test: cover the generated configs, the PXE API and PXE files

pyproject.toml

@@ -51,6 +51,8 @@ ignore = [
 [tool.pytest.ini_options]
 DJANGO_SETTINGS_MODULE = "mdbsite.settings"
 python_files = ["tests/test_*.py"]
+# WhiteNoise, since tests run without collectstatic.
+filterwarnings = ["ignore:No directory at:UserWarning"]
 
 [build-system]
 requires = ["hatchling"]

tests/test_api.py

@@ -41,3 +41,63 @@ def test_validate_puppet_host_by_secret(api_client, user_with_token, host):
     res = api_client.post(reverse("api-validate-host-secret"), data, format="json")
     assert res.status_code == 200, res.content
     assert not Host.objects.get(pk=host.pk).pxe_installable
+
+
+def post_validate(api_client, token, certname, pxe_key):
+    api_client.credentials(HTTP_AUTHORIZATION=f"Token {token.key}")
+    return api_client.post(
+        reverse("api-validate-host-secret"),
+        {"certname": certname, "pxe_key": pxe_key},
+        format="json",
+    )
+
+
+@pytest.mark.parametrize(
+    ("change", "certname", "error"),
+    [
+        ({}, "nosuchhost.edb.technology", "not found"),
+        ({}, "tut.nosuchdomain.example", "not found"),
+        ({"pxe_installable": False}, "tut.edb.technology", "not installable"),
+        ({"pxe_key": ""}, "tut.edb.technology", "has no pxe_key"),
+    ],
+)
+def test_validate_rejects(api_client, user_with_token, host, change, certname, error):
+    _, token = user_with_token
+    Host.objects.filter(pk=host.pk).update(**change)
+
+    res = post_validate(api_client, token, certname, host.pxe_key)
+
+    assert res.status_code == 400
+    assert error in str(res.content)
+
+
+def test_validate_rejects_wrong_key(api_client, user_with_token, host):
+    _, token = user_with_token
+    res = post_validate(api_client, token, "tut.edb.technology", "wrong")
+    assert res.status_code == 400
+    assert "does not match" in str(res.content)
+    # A wrong guess must not use up the installation.
+    host.refresh_from_db()
+    assert host.pxe_installable
+    assert host.pxe_key
+
+
+def test_validate_key_works_only_once(api_client, user_with_token, host):
+    _, token = user_with_token
+    key = host.pxe_key
+    assert (
+        post_validate(api_client, token, "tut.edb.technology", key).status_code == 200
+    )
+    assert (
+        post_validate(api_client, token, "tut.edb.technology", key).status_code == 400
+    )
+
+
+def test_validate_requires_a_token(api_client, host):
+    res = api_client.post(
+        reverse("api-validate-host-secret"),
+        {"certname": "tut.edb.technology", "pxe_key": host.pxe_key},
+        format="json",
+    )
+    assert res.status_code == 401
+    assert Host.objects.get(pk=host.pk).pxe_installable

tests/test_generated_configs.py

@@ -0,0 +1,130 @@
+"""The dhcpd.conf and reverse zones generated from the test_data fixture."""
+
+import pytest
+from django.core.management import call_command
+from freezegun import freeze_time
+
+from mdb.models import DhcpConfig, Interface, Ip4Subnet, Ip6Subnet
+from mdb.services.zone_sync import ZoneSyncService
+
+EXPECTED_DHCPD_CONF = """\
+# Autogenerated configuration 2024-01-01 00:00:00
+default-lease-time 600;
+max-lease-time 7200;
+log-facility local7;
+ddns-update-style none;
+
+# clients
+subnet 10.0.3.0 netmask 255.255.255.0 {
+\toption domain-name "edb.technology";
+\toption domain-name-servers 158.36.190.194, 158.36.190.199, 8.8.8.8;
+\toption routers 158.36.186.1;
+\tdeny unknown-clients;
+\trange 10.0.3.2 10.0.3.254;
+}
+
+host tut {
+\thardware ethernet f0:de:f1:58:be:1a;
+\tfixed-address tut.edb.technology;
+\tfilename "pxelinux.0";
+}
+"""
+
+SOA_AND_NS = """\
+$TTL 3600
+@ IN SOA sega.neuf.no. sysadmin.neuf.no. (
+\t{serial}\t; serial
+\t28800\t; refresh
+\t7200\t; retry
+\t604800\t; expire
+\t86400 )\t; minimum ttl
+;
+;
+@\tIN\tNS\tsega.neuf.no.
+@\tIN\tNS\tatari.neuf.no.
+;
+"""
+
+EXPECTED_IP4_REVERSE = (
+    SOA_AND_NS.format(serial=2024010101)
+    + "2                   \tIN\tPTR\ttut.edb.technology.\n"
+)
+
+EXPECTED_IP6_REVERSE = (
+    SOA_AND_NS.format(serial=1)
+    + "2.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0\tPTR\ttut.edb.technology.\n"
+)
+
+
+@pytest.fixture
+def fixtures(db):
+    # Loading the interface bumps the subnet's serial to the frozen day's first.
+    with freeze_time("2024-01-01 00:00:00"):
+        call_command("loaddata", "test_data", verbosity=0)
+
+
+@pytest.fixture
+def service(tmp_path):
+    svc = ZoneSyncService()
+    svc.checkzone_dir = str(tmp_path)
+    return svc
+
+
+def body(zone_file):
+    """A zone file without its header comments, which name the file and the time."""
+    return zone_file[zone_file.index("$TTL") :]
+
+
+@freeze_time("2024-01-01 00:00:00")
+def test_dhcpd_configuration(fixtures):
+    assert DhcpConfig.objects.get().dhcpd_configuration() == EXPECTED_DHCPD_CONF
+
+
+@freeze_time("2024-01-01 00:00:00")
+def test_dhcpd_configuration_authoritative(fixtures):
+    DhcpConfig.objects.update(authoritative=True)
+    conf = DhcpConfig.objects.get().dhcpd_configuration()
+    assert conf.splitlines()[1] == "authoritative;"
+
+
+def test_dhcpd_configuration_leaves_out_non_dhcp_hosts(fixtures):
+    Interface.objects.update(dhcp_client=False)
+    assert "host tut" not in DhcpConfig.objects.get().dhcpd_configuration()
+
+
+def test_dhcpd_configuration_without_pxe_filename(fixtures):
+    Interface.objects.update(pxe_filename="")
+    conf = DhcpConfig.objects.get().dhcpd_configuration()
+    assert "host tut {" in conf
+    assert "filename" not in conf
+
+
+def test_dhcpd_configuration_without_dynamic_range(fixtures):
+    Ip4Subnet.objects.update(dhcp_dynamic=False)
+    assert "range" not in DhcpConfig.objects.get().dhcpd_configuration()
+
+
+def test_ip4_reverse_zone(fixtures, service):
+    zone = Ip4Subnet.objects.get()
+    contents = zone.zone_file_contents()
+    assert body(contents) == EXPECTED_IP4_REVERSE
+    status, output = service.check_zone(zone, contents)
+    assert status == 0, output
+
+
+def test_ip4_reverse_zone_leaves_out_unassigned_addresses(fixtures):
+    Interface.objects.update(ip4address=None)
+    assert "PTR" not in Ip4Subnet.objects.get().zone_file_contents()
+
+
+def test_ip6_reverse_zone(fixtures, service):
+    zone = Ip6Subnet.objects.get()
+    contents = zone.zone_file_contents()
+    assert body(contents) == EXPECTED_IP6_REVERSE
+    status, output = service.check_zone(zone, contents)
+    assert status == 0, output
+
+
+def test_ip6_reverse_zone_name():
+    zone = Ip6Subnet(network="2001:16d8:eeb6:3", netmask=64)
+    assert zone.reverse_zone_name == "3.0.0.0.6.b.e.e.8.d.6.1.1.0.0.2.ip6.arpa"

tests/test_pxe.py

@@ -1,9 +1,11 @@
 import pytest
 from django.conf import settings
+from django.contrib import messages
 from django.core.management import call_command
+from django.urls import reverse
 
-from mdb.models import Host, OperatingSystem
+from mdb.models import Host, Interface, OperatingSystem
-from mdb.pxe import render_pxelinux_cfg
+from mdb.pxe import get_pxe_filename, render_pxelinux_cfg
 
 
 @pytest.fixture
@@ -43,3 +45,85 @@ def test_render_pxeconfig_with_os_pxe_config(host):
 @pytest.mark.django_db
 def test_host_as_pxe_files(host):
     assert len(host.as_pxe_files()) == 1
+
+
+PXE_FILE = "01-f0-de-f1-58-be-1a"  # from the fixture interface's MAC address
+
+
+@pytest.fixture
+def tftp_root(settings, tmp_path):
+    settings.MDB_PXE_TFTP_ROOT = str(tmp_path)
+    return tmp_path
+
+
+def test_saving_installable_host_writes_pxe_file(host, tftp_root):
+    Interface.objects.update(pxe_filename="", dhcp_client=False)
+    host.save()
+
+    path = tftp_root / PXE_FILE
+    assert host.pxe_key in path.read_text()
+    assert path.stat().st_mode & 0o777 == 0o644
+    # The interface is set up to boot from the network.
+    interface = host.interface_set.get()
+    assert interface.pxe_filename == "pxelinux.0"
+    assert interface.dhcp_client
+
+
+def test_saving_host_keeps_custom_pxe_filename(host, tftp_root):
+    Interface.objects.update(pxe_filename="custom.efi")
+    host.save()
+    assert host.interface_set.get().pxe_filename == "custom.efi"
+
+
+def test_saving_non_installable_host_deletes_pxe_file(host, tftp_root):
+    (tftp_root / PXE_FILE).write_text("old")
+    host.pxe_installable = False
+    host.save()
+    assert not (tftp_root / PXE_FILE).exists()
+
+
+def test_saving_host_without_key_generates_one(host, tftp_root):
+    host.pxe_key = ""
+    host.save()
+    key = Host.objects.get(pk=host.pk).pxe_key
+    assert len(key) == 32
+    assert key in (tftp_root / PXE_FILE).read_text()
+
+
+def test_pxe_filename_by_ip_address(host):
+    interface = host.interface_set.get()  # 10.0.3.2
+    assert get_pxe_filename(interface, "ip_addr") == "0A000302"
+
+
+def test_admin_marks_hosts_installable(host, tftp_root, admin_client):
+    Host.objects.update(pxe_installable=False)
+
+    res = admin_client.post(
+        reverse("admin:mdb_host_changelist"),
+        {"action": "set_installable", "_selected_action": [host.pk]},
+        follow=True,
+    )
+
+    assert res.status_code == 200
+    assert Host.objects.get(pk=host.pk).pxe_installable
+    assert (tftp_root / PXE_FILE).exists()
+    assert not [m for m in res.context["messages"] if m.level == messages.WARNING]
+
+
+def test_admin_warns_about_interfaces_without_ip(host, tftp_root, admin_client):
+    Host.objects.update(pxe_installable=False)
+    Interface.objects.create(
+        name="eth1", macaddr="00:11:22:33:44:55", host=host, domain_id=1
+    )
+
+    res = admin_client.post(
+        reverse("admin:mdb_host_changelist"),
+        {"action": "set_installable", "_selected_action": [host.pk]},
+        follow=True,
+    )
+
+    warnings = [str(m) for m in res.context["messages"] if m.level == messages.WARNING]
+    assert warnings == [
+        "Host tut might not be installable since interface(s) eth1 has no IP set."
+    ]
+    assert Host.objects.get(pk=host.pk).pxe_installable

tests/test_zone_sync.py

@@ -7,6 +7,7 @@ from freezegun import freeze_time
 
 from mdb.models import (
     DhcpConfig,
+    DomainARecord,
     Host,
     HostType,
     Interface,
@@ -320,3 +321,21 @@ def test_zone_file_contents_includes_ip4_in_spf_record_when_interface_has_addres
     add_host(db_domain, "mailhost", allow_email=True, ip4address=address)
     content = db_domain.zone_file_contents()
     assert f'mailhost\tTXT\t"v=spf1 mx ip4:{address.address} -all"\n' in content
+
+
+def test_sync_zone_does_not_write_a_zone_that_fails_validation(db_domain, tmp_path):
+    with open(db_domain.domain_filename, "w") as f:
+        f.write("the live zone\n")
+    DomainARecord.objects.create(domain=db_domain, name="bad", target="not-an-address")
+    svc = ZoneSyncService(debug=False)
+    svc.checkzone_dir = str(tmp_path)
+
+    assert svc.sync_zone(db_domain) is False
+
+    with open(db_domain.domain_filename) as f:
+        assert f.read() == "the live zone\n"
+    db_domain.refresh_from_db()
+    assert db_domain.domain_active_serial == 0
+    change = svc.changes["example.org"]
+    assert change["success"] is False
+    assert "not-an-address" in change["errors"]