diff --git a/pyproject.toml b/pyproject.toml index f5a97b3..1e96ee2 100644 --- a/pyproject.toml +++ b/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"] diff --git a/tests/test_api.py b/tests/test_api.py index caa0c6e..1f2b339 100644 --- a/tests/test_api.py +++ b/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 diff --git a/tests/test_generated_configs.py b/tests/test_generated_configs.py new file mode 100644 index 0000000..ef1d8f2 --- /dev/null +++ b/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" diff --git a/tests/test_pxe.py b/tests/test_pxe.py index 07ed268..d0339fd 100644 --- a/tests/test_pxe.py +++ b/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.pxe import render_pxelinux_cfg +from mdb.models import Host, Interface, OperatingSystem +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 diff --git a/tests/test_zone_sync.py b/tests/test_zone_sync.py index a27f4d6..e795b5a 100644 --- a/tests/test_zone_sync.py +++ b/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"]