Revamp zone sync and add tests

tests/conftest.py

@@ -0,0 +1,64 @@
+import pytest
+from django.core.management import call_command
+
+from mdb.models import (
+    Domain,
+    DomainAAAARecord,
+    DomainARecord,
+    DomainCnameRecord,
+    DomainTxtRecord,
+    MailExchange,
+    Nameserver,
+)
+from mdb.services.zone_sync import ZoneSyncService
+
+
+@pytest.fixture
+def domain(db):
+    call_command("loaddata", "test_data", verbosity=0)
+    return Domain.objects.get(domain_name="edb.technology")
+
+
+@pytest.fixture
+def service(tmp_path):
+    svc = ZoneSyncService(debug=True)
+    svc.checkzone_dir = str(tmp_path)
+    return svc
+
+
+@pytest.fixture
+def db_domain(db, tmp_path):
+    domain = Domain.objects.create(
+        domain_name="example.org",
+        domain_soa="ns1.example.org",
+        domain_admin="admin@example.org",
+        domain_ipaddr="1.2.3.4",
+        domain_ip6addr="2001:db8::1",
+        domain_filename=str(tmp_path / "db.example.org"),
+        domain_ttl=3600,
+        domain_serial=2024010101,
+    )
+    ns1 = Nameserver.objects.create(hostname="ns1.example.org")
+    ns2 = Nameserver.objects.create(hostname="ns2.example.org")
+    domain.domain_nameservers.set([ns1, ns2])
+    mx = MailExchange.objects.create(priority=10, hostname="mail.example.org")
+    domain.domain_mailexchanges.set([mx])
+    DomainARecord.objects.create(domain=domain, name="ns1", target="1.2.3.4")
+    DomainARecord.objects.create(domain=domain, name="ns2", target="5.6.7.8")
+    DomainARecord.objects.create(domain=domain, name="mail", target="9.10.11.12")
+    DomainARecord.objects.create(domain=domain, name="www", target="1.2.3.4")
+    DomainAAAARecord.objects.create(domain=domain, name="www", target="2001:db8::1")
+    DomainCnameRecord.objects.create(domain=domain, name="ftp", target="www.example.org.")
+    DomainTxtRecord.objects.create(domain=domain, name="@", target='"v=spf1 mx ~all"')
+    return domain
+
+
+def load_command_class():
+    path = (
+        pathlib.Path(__file__).parent.parent
+        / "mdb/management/commands/zone-sync.py"
+    )
+    spec = importlib.util.spec_from_file_location("zone_sync_command", path)
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod.Command

tests/test_zone_sync.py

@@ -0,0 +1,185 @@
+import os
+from unittest.mock import MagicMock
+
+import pytest
+from freezegun import freeze_time
+
+from mdb.services.zone_sync import ZoneSyncService
+
+EXPECTED_ZONE_FILE_CONTENTS = """\
+; serial:2024010101
+; zone file for example.org
+; 2024-01-01 00:00:00
+; filename: {filename}
+$TTL 3600
+@ IN SOA ns1.example.org. admin.example.org. (
+\t2024010101\t; serial
+\t28800\t; refresh
+\t7200\t; retry
+\t604800\t; expire
+\t86400 )\t; minimum ttl
+;
+@\tIN\tNS\tns1.example.org.
+@\tIN\tNS\tns2.example.org.
+@\t\tMX\t10 mail.example.org.
+@\tIN\tA\t1.2.3.4
+@\tIN\tAAAA\t2001:db8::1
+; SRV records
+; A records
+ns1 IN A 1.2.3.4
+ns2 IN A 5.6.7.8
+mail IN A 9.10.11.12
+www IN A 1.2.3.4
+; AAAA records
+www IN AAAA 2001:db8::1
+; CNAME records \nftp\tIN\tCNAME\twww.example.org.
+; TXT records \n@ TXT "v=spf1 mx ~all"
+; HOST records
+"""
+
+EXPECTED_NORMALIZED = (
+    "$TTL 3600\t; 1 hour\n"
+    "example.org.\t\tIN SOA\tns1.example.org. admin.example.org. (\n"
+    "\t\t\t\t2024010101 ; serial\n"
+    "\t\t\t\t28800      ; refresh (8 hours)\n"
+    "\t\t\t\t7200       ; retry (2 hours)\n"
+    "\t\t\t\t604800     ; expire (1 week)\n"
+    "\t\t\t\t86400      ; minimum (1 day)\n"
+    "\t\t\t\t)\n"
+    "\t\t\tNS\tns1.example.org.\n"
+    "\t\t\tNS\tns2.example.org.\n"
+    "\t\t\tA\t1.2.3.4\n"
+    "\t\t\tMX\t10 mail.example.org.\n"
+    '\t\t\tTXT\t"v=spf1 mx ~all"\n'
+    "\t\t\tAAAA\t2001:db8::1\n"
+    "ftp.example.org.\tCNAME\twww.example.org.\n"
+    "mail.example.org.\tA\t9.10.11.12\n"
+    "ns1.example.org.\tA\t1.2.3.4\n"
+    "ns2.example.org.\tA\t5.6.7.8\n"
+    "www.example.org.\tA\t1.2.3.4\n"
+    "\t\t\tAAAA\t2001:db8::1"
+)
+
+
+SAMPLE_ZONE = """\
+$TTL 3600
+example.com. IN SOA ns1.example.com. admin.example.com. (
+    2024010101 ; serial
+    3600       ; refresh
+    900        ; retry
+    604800     ; expire
+    86400      ; minimum
+)
+example.com.        IN NS    ns1.example.com.
+example.com.        IN NS    ns2.example.com.
+example.com.        IN MX    10 mail.example.com.
+example.com.        IN MX    20 mail2.example.com.
+example.com.        IN A     1.2.3.4
+example.com.        IN AAAA  2001:db8::1
+example.com.        IN TXT   "v=spf1 mx ~all"
+ns1.example.com.    IN A     1.2.3.4
+ns2.example.com.    IN A     5.6.7.8
+mail.example.com.   IN A     9.10.11.12
+www.example.com.    IN CNAME example.com.
+_dmarc.example.com. IN TXT   "v=DMARC1; p=none; rua=mailto:dmarc@example.com"
+"""
+
+
+def make_zone(domain_name="example.com", active_serial=2024010101, serial=2024010102):
+    zone = MagicMock()
+    zone.domain_name = domain_name
+    zone.domain_active_serial = active_serial
+    zone.domain_serial = serial
+    return zone
+
+
+def test_calculate_diff_identical_returns_empty(service):
+    zone = make_zone()
+    assert service.calculate_diff("line\n", "line\n", zone) == ""
+
+
+def test_calculate_diff_format(service):
+    zone = make_zone(domain_name="example.com", active_serial=1, serial=2)
+    diff = service.calculate_diff("new line\n", "old line\n", zone)
+    assert "*** 1-example.com" in diff
+    assert "--- 2-example.com" in diff
+    assert "! old line" in diff
+    assert "! new line" in diff
+
+
+def test_normalize_zone_is_idempotent(service):
+    zone = make_zone()
+    normalized = service.normalize_zone(zone, SAMPLE_ZONE)
+    assert service.normalize_zone(zone, normalized) == normalized
+
+
+def test_normalize_zone_fallback_when_binary_missing(service):
+    service.compilezone_bin = "/nonexistent/named-compilezone"
+    zone = make_zone()
+    assert service.normalize_zone(zone, SAMPLE_ZONE) == SAMPLE_ZONE
+
+
+def test_check_zone_valid(service):
+    zone = make_zone()
+    status, _ = service.check_zone(zone, SAMPLE_ZONE)
+    assert status == 0
+
+
+def test_check_zone_invalid(service):
+    zone = make_zone()
+    status, _ = service.check_zone(zone, "this is not a valid zone file")
+    assert status != 0
+
+
+@freeze_time("2024-01-01 00:00:00")
+def test_zone_file_contents_includes_all_record_types(db_domain):
+    content = db_domain.zone_file_contents()
+    assert content == EXPECTED_ZONE_FILE_CONTENTS.format(
+        filename=db_domain.domain_filename
+    )
+
+
+def test_zone_file_contents_normalizes_without_error(db_domain, service):
+    normalized = service.normalize_zone(db_domain, db_domain.zone_file_contents())
+    assert normalized == EXPECTED_NORMALIZED
+
+
+def test_check_zone_accepts_zone_file_contents(db_domain, service):
+    content = service.normalize_zone(db_domain, db_domain.zone_file_contents())
+    status, output = service.check_zone(db_domain, content)
+    assert status == 0, output
+    assert "loaded serial 2024010101" in output
+    assert output.endswith("OK")
+
+
+def test_find_zone_returns_matching_domain(db_domain, service):
+    assert service.find_zone("example.org") == db_domain
+
+
+@pytest.mark.django_db
+def test_find_zone_returns_none_for_unknown(service):
+    assert service.find_zone("nonexistent.example") is None
+
+
+def test_sync_zone_writes_zone_file_and_updates_serial(db_domain, tmp_path):
+    svc = ZoneSyncService(debug=False)
+    svc.checkzone_dir = str(tmp_path)
+
+    svc.sync_zone(db_domain)
+
+    assert os.path.exists(db_domain.domain_filename)
+    db_domain.refresh_from_db()
+    assert db_domain.domain_active_serial == db_domain.domain_serial
+
+
+def test_sync_zone_records_success_and_diff(db_domain, tmp_path):
+    svc = ZoneSyncService(debug=False)
+    svc.checkzone_dir = str(tmp_path)
+
+    svc.sync_zone(db_domain)
+
+    change = svc.changes["example.org"]
+    assert change["success"] is True
+    assert change["errors"] is None
+    assert change["old_serial"] == 0
+    assert change["new_serial"] == 2024010101

tests/test_zone_sync_command.py

@@ -0,0 +1,109 @@
+from io import StringIO
+from unittest.mock import MagicMock, patch
+
+import pytest
+from mdb.management.commands.zone_sync import Command as CommandClass
+
+
+@pytest.fixture
+def cmd():
+    c = CommandClass()
+    c.stdout = StringIO()
+    c.stderr = StringIO()
+    return c
+
+
+def make_service(changes=None):
+    svc = MagicMock()
+    svc.changes = changes or {
+        "example.org": {"errors": None, "diff": "- old\n+ new", "success": True}
+    }
+    svc.run_reload_commands.return_value = (True, "", "")
+    return svc
+
+
+# --- run_reload_commands ---
+
+def test_run_reload_commands_calls_all_on_success(service):
+    with patch("mdb.services.zone_sync.getstatusoutput", return_value=(0, "")) as mock:
+        success, _, _ = service.run_reload_commands()
+    assert success is True
+    assert mock.call_count == 3
+
+
+def test_run_reload_commands_stops_on_first_failure(service):
+    with patch(
+        "mdb.services.zone_sync.getstatusoutput",
+        side_effect=[(0, ""), (1, "connection refused"), (0, "")],
+    ) as mock:
+        success, failed_cmd, output = service.run_reload_commands()
+    assert success is False
+    assert "reload" in failed_cmd
+    assert output == "connection refused"
+    assert mock.call_count == 2  # thaw never reached
+
+
+# --- reload_and_notify ---
+
+def test_reload_and_notify_success_writes_ok(cmd):
+    svc = make_service()
+    with patch("mdb.management.commands.zone_sync.mail_admins"):
+        cmd.reload_and_notify(svc, errors="", diffs="diff", zones="example.org")
+    assert "ok" in cmd.stdout.getvalue()
+
+
+def test_reload_and_notify_success_sends_mail(cmd):
+    svc = make_service()
+    with patch("mdb.management.commands.zone_sync.mail_admins") as mock_mail:
+        cmd.reload_and_notify(svc, errors="", diffs="diff", zones="example.org")
+    mock_mail.assert_called_once()
+    assert "Success" in mock_mail.call_args.kwargs["subject"]
+
+
+def test_reload_and_notify_failure_writes_fail_and_exits(cmd):
+    svc = make_service()
+    svc.run_reload_commands.return_value = (False, "/usr/sbin/rndc freeze", "timed out")
+    with patch("mdb.management.commands.zone_sync.mail_admins"):
+        with patch("os.getlogin", return_value="testuser"):
+            with pytest.raises(SystemExit) as exc:
+                cmd.reload_and_notify(svc, errors="", diffs="", zones="example.org")
+    assert exc.value.code == 1
+    assert "fail" in cmd.stdout.getvalue()
+
+
+def test_reload_and_notify_failure_mail_includes_command_and_output(cmd):
+    svc = make_service()
+    svc.run_reload_commands.return_value = (False, "/usr/sbin/rndc freeze", "timed out")
+    with patch("mdb.management.commands.zone_sync.mail_admins") as mock_mail:
+        with patch("os.getlogin", return_value="testuser"):
+            with pytest.raises(SystemExit):
+                cmd.reload_and_notify(svc, errors="", diffs="", zones="example.org")
+    subject = mock_mail.call_args.kwargs["subject"]
+    message = mock_mail.call_args.kwargs["message"]
+    assert "FAIL" in subject
+    assert "rndc freeze" in message
+    assert "timed out" in message
+
+
+# --- handle ---
+
+@pytest.mark.django_db
+def test_handle_exits_if_zone_not_found(cmd):
+    with pytest.raises(SystemExit) as exc:
+        cmd.handle(debug=True, force=False, zone="nonexistent.example")
+    assert exc.value.code == 1
+    assert "nonexistent.example" in cmd.stderr.getvalue()
+
+
+@pytest.mark.django_db
+def test_handle_skips_reload_in_debug_mode(cmd, db_domain):
+    with patch.object(cmd, "reload_and_notify") as mock_reload:
+        cmd.handle(debug=True, force=True, zone="example.org")
+    mock_reload.assert_not_called()
+
+
+@pytest.mark.django_db
+def test_handle_calls_reload_and_notify_when_changes_exist(cmd, db_domain):
+    with patch.object(cmd, "reload_and_notify") as mock_reload:
+        cmd.handle(debug=False, force=True, zone="example.org")
+    mock_reload.assert_called_once()