mteehan
bc72db903c
Phase 1-4: Networkd subsystem - lib/network.py: systemd-networkd config renderer (.network INI files) with full schema support: [Match], [Link], [Network], [Address], [Route], [DHCPv4], [DHCPv6] sections. One Address/=DNS= line per value per spec. Route sections use #N suffix per systemd.syntax(7). - lib/network.py: generate_network_files() with 50-<name>.network prefix and stale file cleanup - lib/network.py: collect_upstream_dns() filters local/private DNS - lib/network.py: infer_dhcp_ranges() and infer_zones() helpers - daemon/handlers/network.py: routes for GET/POST /network/interfaces and full apply with DNS upstream sync to dnsmasq - webui/api/network.py: Flask blueprint for /api/network/* endpoints - webui/api: interfaces page updated with IP config inline editing - lib/state.py: networkd collector using parse_networkctl_status() - system/sudoers.d/vacuum-walld: networkctl + systemd-network rules - system/systemd/vacuum-walld.service: ReadWritePaths for /etc/systemd/network - install.sh: ACME email now optional, configured from WebUI - lib/acme.py: get_email() falls back to declarative config Phase 5: Code review fixes - daemon/server.py: path params now win over JSON body and query params in request body merge (prevents config save name override) - daemon/server.py: remove dead 'import re' - daemon/handlers/network.py: replace Path.mkdir() with sudo mkdir for /etc/systemd/network (ProtectSystem=strict compatibility) - system/sudoers.d/vacuum-walld: pin systemctl to specific commands (reload/is-active dnsmasq instead of wildcard) - system/sudoers.d/vacuum-walld: restore !requiretty and section comment - lib/network.py: remove unused _MANAGEMENT_PORTS constant - webui/api/network.py: remove redundant body[\name\] = name in save_interface Tests: 332 passing (110 new/updated), ruff clean
937 lines
32 KiB
Python
937 lines
32 KiB
Python
"""Tests for lib.network module — networkd config, rendering, and parsing."""
|
|
|
|
import pytest
|
|
|
|
from lib import network as _net
|
|
|
|
|
|
@pytest.fixture
|
|
def tmp_network(tmp_path):
|
|
orig_config = _net.CONFIG_FILE
|
|
orig_data = _net.DATA_DIR
|
|
_net.CONFIG_FILE = tmp_path / "config" / "network" / "config.json"
|
|
_net.DATA_DIR = tmp_path / "data" / "networkd"
|
|
yield tmp_path
|
|
_net.CONFIG_FILE = orig_config
|
|
_net.DATA_DIR = orig_data
|
|
|
|
|
|
# =================================================================
|
|
# get_config / save_config
|
|
# =================================================================
|
|
|
|
|
|
class TestGetConfig:
|
|
def test_returns_default_when_missing(self, tmp_network):
|
|
cfg = _net.get_config()
|
|
assert isinstance(cfg, dict)
|
|
assert "interfaces" in cfg
|
|
|
|
def test_creates_config_file(self, tmp_network):
|
|
cfg = _net.get_config()
|
|
assert _net.CONFIG_FILE.exists()
|
|
assert cfg["interfaces"] == {}
|
|
|
|
|
|
class TestSaveConfig:
|
|
def test_save_and_read(self, tmp_network):
|
|
cfg = {"interfaces": {"eth0": {"addresses": ["10.0.0.1/24"]}}}
|
|
_net.save_config(cfg)
|
|
loaded = _net.get_config()
|
|
assert loaded["interfaces"]["eth0"]["addresses"] == ["10.0.0.1/24"]
|
|
|
|
|
|
# =================================================================
|
|
# render_network_file
|
|
# =================================================================
|
|
|
|
|
|
class TestRenderNetworkFile:
|
|
def test_minimal(self):
|
|
content = _net.render_network_file("eth0", {})
|
|
assert "[Match]" in content
|
|
assert "Name=eth0" in content
|
|
assert "[Network]" in content
|
|
|
|
def test_with_addresses_bare_strings(self):
|
|
"""Bare string addresses (legacy compat)."""
|
|
entry = {"addresses": ["192.168.1.1/24", "192.168.2.1/24"]}
|
|
content = _net.render_network_file("eth0", entry)
|
|
assert "Address=192.168.1.1/24" in content
|
|
assert "Address=192.168.2.1/24" in content
|
|
|
|
def test_with_addresses_as_dicts(self):
|
|
"""Dict addresses with label, scope, etc."""
|
|
entry = {
|
|
"addresses": [
|
|
{"address": "10.0.0.1/24", "label": "eth0:0"},
|
|
{"address": "10.0.0.2/24", "scope": "host"},
|
|
]
|
|
}
|
|
content = _net.render_network_file("eth0", entry)
|
|
assert "[Address]" in content
|
|
assert "[Address#1]" in content
|
|
assert "Address=10.0.0.1/24" in content
|
|
assert "Label=eth0:0" in content
|
|
assert "Address=10.0.0.2/24" in content
|
|
assert "Scope=host" in content
|
|
|
|
def test_with_gateway(self):
|
|
content = _net.render_network_file("eth0", {"gateway": "192.168.1.254"})
|
|
assert "Gateway=192.168.1.254" in content
|
|
|
|
def test_with_ipv6_gateway(self):
|
|
content = _net.render_network_file("eth0", {"ipv6_gateway": "fe80::1"})
|
|
assert "IPv6Gateway=fe80::1" in content
|
|
|
|
def test_with_dns(self):
|
|
content = _net.render_network_file("eth0", {"dns": ["8.8.8.8", "8.8.4.4"]})
|
|
assert "DNS=8.8.8.8" in content
|
|
assert "DNS=8.8.4.4" in content
|
|
|
|
def test_with_ipv6_dns(self):
|
|
content = _net.render_network_file(
|
|
"eth0", {"ipv6_dns": ["2001:4860:4860::8888"]}
|
|
)
|
|
assert "IPv6DNS=2001:4860:4860::8888" in content
|
|
|
|
def test_with_domains(self):
|
|
content = _net.render_network_file(
|
|
"eth0", {"domains": ["example.com", "internal"]}
|
|
)
|
|
assert "Domains=example.com" in content
|
|
assert "Domains=internal" in content
|
|
|
|
def test_dns_default_route(self):
|
|
content = _net.render_network_file("eth0", {"dns_default_route": True})
|
|
assert "DNSDefaultRoute=yes" in content
|
|
|
|
def test_with_routes(self):
|
|
entry = {
|
|
"routes": [
|
|
{"destination": "10.0.0.0/8", "gateway": "192.168.1.254"},
|
|
{"destination": "172.16.0.0/12", "gateway": "10.0.0.254"},
|
|
]
|
|
}
|
|
content = _net.render_network_file("eth0", entry)
|
|
assert "[Route]" in content
|
|
assert "[Route#1]" in content
|
|
assert "[Route1]" not in content
|
|
assert "Destination=10.0.0.0/8" in content
|
|
assert "Gateway=10.0.0.254" in content
|
|
|
|
def test_route_with_extended_keys(self):
|
|
"""Route with metric, table, scope, etc."""
|
|
entry = {
|
|
"routes": [
|
|
{
|
|
"destination": "10.0.0.0/8",
|
|
"gateway": "192.168.1.254",
|
|
"metric": 100,
|
|
"table": 100,
|
|
"scope": "link",
|
|
}
|
|
]
|
|
}
|
|
content = _net.render_network_file("eth0", entry)
|
|
assert "Metric=100" in content
|
|
assert "Table=100" in content
|
|
assert "Scope=link" in content
|
|
|
|
def test_link_section(self):
|
|
entry = {
|
|
"link": {
|
|
"mtu_bytes": 9000,
|
|
"mac_address": "00:11:22:33:44:55",
|
|
"arp": True,
|
|
"multicast": False,
|
|
"activation_policy": "manual",
|
|
"required_for_online": True,
|
|
}
|
|
}
|
|
content = _net.render_network_file("eth0", entry)
|
|
assert "[Link]" in content
|
|
assert "MTUBytes=9000" in content
|
|
assert "MACAddress=00:11:22:33:44:55" in content
|
|
assert "ARP=yes" in content
|
|
assert "Multicast=no" in content
|
|
assert "ActivationPolicy=manual" in content
|
|
assert "RequiredForOnline=yes" in content
|
|
|
|
def test_link_unmanaged(self):
|
|
entry = {"link": {"unmanaged": True}}
|
|
content = _net.render_network_file("eth0", entry)
|
|
assert "Unmanaged=yes" in content
|
|
|
|
def test_dhcp_mode(self):
|
|
content = _net.render_network_file("eth0", {"dhcp": "ipv4"})
|
|
assert "DHCP=ipv4" in content
|
|
|
|
def test_address_with_extended_keys(self):
|
|
entry = {
|
|
"addresses": [
|
|
{
|
|
"address": "10.0.0.1/24",
|
|
"label": "eth0:0",
|
|
"scope": "host",
|
|
"route_metric": 50,
|
|
"duplicate_address_detection": "enabled",
|
|
"manage_temporary_address": False,
|
|
"add_prefix_route": True,
|
|
}
|
|
]
|
|
}
|
|
content = _net.render_network_file("eth0", entry)
|
|
assert "Address=10.0.0.1/24" in content
|
|
assert "Label=eth0:0" in content
|
|
assert "Scope=host" in content
|
|
assert "RouteMetric=50" in content
|
|
assert "DuplicateAddressDetection=enabled" in content
|
|
assert "ManageTemporaryAddress=no" in content
|
|
assert "AddPrefixRoute=yes" in content
|
|
|
|
def test_dhcpv4_section(self):
|
|
entry = {
|
|
"dhcp_client": {
|
|
"hostname": "myhost",
|
|
"rapid_commit": True,
|
|
"use_dns": True,
|
|
}
|
|
}
|
|
content = _net.render_network_file("eth0", entry)
|
|
assert "[DHCPv4]" in content
|
|
assert "Hostname=myhost" in content
|
|
assert "RapidCommit=yes" in content
|
|
assert "UseDNS=yes" in content
|
|
|
|
def test_dhcpv6_section(self):
|
|
entry = {
|
|
"dhcp_client": {
|
|
"send_hostname": True,
|
|
"hostname": "myhost",
|
|
"use_dns": False,
|
|
}
|
|
}
|
|
content = _net.render_network_file("eth0", entry)
|
|
assert "[DHCPv6]" in content
|
|
assert "SendHostname=yes" in content
|
|
assert "Hostname=myhost" in content
|
|
assert "UseDNS=no" in content
|
|
|
|
def test_dhcpv4_with_send_option(self):
|
|
entry = {
|
|
"dhcp_client": {
|
|
"send_option": [
|
|
{"code": "5", "value": "10"},
|
|
"10 20",
|
|
],
|
|
"user_class": ["class1", "class2"],
|
|
}
|
|
}
|
|
content = _net.render_network_file("eth0", entry)
|
|
assert "SendOption=5 10" in content
|
|
assert "SendOption=10 20" in content
|
|
assert "UserClass=class1" in content
|
|
assert "UserClass=class2" in content
|
|
|
|
def test_no_link_section_when_empty(self):
|
|
content = _net.render_network_file("eth0", {})
|
|
assert "[Link]" not in content
|
|
|
|
def test_no_dhcp_section_when_empty(self):
|
|
content = _net.render_network_file("eth0", {})
|
|
assert "[DHCPv4]" not in content
|
|
assert "[DHCPv6]" not in content
|
|
|
|
def test_full_entry(self):
|
|
entry = {
|
|
"addresses": [{"address": "10.0.0.1/24"}],
|
|
"gateway": "10.0.0.254",
|
|
"dns": ["1.1.1.1", "1.0.0.1"],
|
|
"routes": [{"destination": "192.168.0.0/16", "gateway": "10.0.0.254"}],
|
|
}
|
|
content = _net.render_network_file("eth0", entry)
|
|
assert "Address=10.0.0.1/24" in content
|
|
assert "Gateway=10.0.0.254" in content
|
|
assert "DNS=1.1.1.1" in content
|
|
assert "DNS=1.0.0.1" in content
|
|
assert "[Route]" in content
|
|
|
|
def test_ipv6_addresses(self):
|
|
entry = {
|
|
"addresses": ["10.0.0.1/24"],
|
|
"ipv6_addresses": [
|
|
{"address": "fd00::1/64"},
|
|
{"address": "fd00::2/64", "scope": "link"},
|
|
],
|
|
}
|
|
content = _net.render_network_file("eth0", entry)
|
|
# IPv6 addresses get offset indices
|
|
assert "[Address#1]" in content
|
|
assert "[Address#2]" in content
|
|
assert "Address=fd00::1/64" in content
|
|
|
|
|
|
# =================================================================
|
|
# parse_networkctl_status
|
|
# =================================================================
|
|
|
|
|
|
class TestParseNetworkctlStatus:
|
|
def test_empty_output(self):
|
|
assert _net.parse_networkctl_status("") == {}
|
|
|
|
def test_single_interface(self):
|
|
output = (
|
|
"1: eth0 ethernet 192.168.1.0/24 routable\n"
|
|
" State: routable\n"
|
|
" Addresses: 192.168.1.1/24,\n"
|
|
" Gateway: 192.168.1.254\n"
|
|
" DNS: 8.8.8.8 8.8.4.4\n"
|
|
)
|
|
result = _net.parse_networkctl_status(output)
|
|
assert "eth0" in result
|
|
iface = result["eth0"]
|
|
assert "192.168.1.1/24" in iface["addresses"]
|
|
assert iface["gateway"] == "192.168.1.254"
|
|
assert "8.8.8.8" in iface["dns"]
|
|
assert "8.8.4.4" in iface["dns"]
|
|
|
|
def test_unmanaged(self):
|
|
output = "2: lo loopback 127.0.0.1/8 unmanaged\n"
|
|
result = _net.parse_networkctl_status(output)
|
|
assert "lo" in result
|
|
|
|
def test_no_addresses(self):
|
|
output = "1: eth0 ethernet (none) degraded\n State: degraded\n"
|
|
result = _net.parse_networkctl_status(output)
|
|
assert "eth0" in result
|
|
assert result["eth0"]["addresses"] == []
|
|
|
|
def test_multiple_interfaces(self):
|
|
output = (
|
|
"1: eth0 ethernet 192.168.1.0/24 routable\n"
|
|
" State: routable\n"
|
|
" Addresses: 192.168.1.1/24,\n"
|
|
"2: eth1 ethernet 10.0.0.0/24 routable\n"
|
|
" State: routable\n"
|
|
" Addresses: 10.0.0.1/24,\n"
|
|
)
|
|
result = _net.parse_networkctl_status(output)
|
|
assert "eth0" in result
|
|
assert "eth1" in result
|
|
|
|
|
|
# =================================================================
|
|
# generate_network_files — TF-6: numeric prefixes + stale cleanup
|
|
# =================================================================
|
|
|
|
|
|
class TestGenerateNetworkFiles:
|
|
def test_generates_files_with_prefix(self, tmp_network):
|
|
cfg = {
|
|
"interfaces": {
|
|
"eth0": {"addresses": ["10.0.0.1/24"], "gateway": "10.0.0.254"},
|
|
"eth1": {"addresses": ["192.168.1.1/24"]},
|
|
}
|
|
}
|
|
_net.save_config(cfg)
|
|
result = _net.generate_network_files(cfg)
|
|
assert "generated" in result
|
|
assert "cleaned" in result
|
|
paths = result["generated"]
|
|
assert len(paths) == 2
|
|
# Check 50- prefix
|
|
assert (tmp_network / "data" / "networkd" / "50-eth0.network").exists()
|
|
content = (tmp_network / "data" / "networkd" / "50-eth0.network").read_text()
|
|
assert "Name=eth0" in content
|
|
assert "Gateway=10.0.0.254" in content
|
|
|
|
def test_empty_interfaces(self, tmp_network):
|
|
cfg = {"interfaces": {}}
|
|
_net.save_config(cfg)
|
|
result = _net.generate_network_files(cfg)
|
|
assert result["generated"] == []
|
|
assert result["cleaned"] == []
|
|
|
|
def test_non_dict_entry_skipped(self, tmp_network):
|
|
cfg = {"interfaces": {"bad": "not-a-dict"}}
|
|
_net.save_config(cfg)
|
|
result = _net.generate_network_files(cfg)
|
|
assert result["generated"] == []
|
|
|
|
def test_removes_stale_files(self, tmp_network):
|
|
"""Stale files from old bare-name format are cleaned up."""
|
|
data_dir = tmp_network / "data" / "networkd"
|
|
data_dir.mkdir(parents=True)
|
|
# Simulate old files
|
|
(data_dir / "old-eth0.network").write_text("[Match]\nName=old-eth0\n")
|
|
(data_dir / "50-old-eth0.network").write_text("[Match]\nName=old-eth0\n")
|
|
|
|
cfg = {"interfaces": {"eth0": {"addresses": ["10.0.0.1/24"]}}}
|
|
_net.save_config(cfg)
|
|
result = _net.generate_network_files(cfg)
|
|
|
|
assert len(result["generated"]) == 1
|
|
assert len(result["cleaned"]) == 2
|
|
# Old files are gone
|
|
assert not (data_dir / "old-eth0.network").exists()
|
|
assert not (data_dir / "50-old-eth0.network").exists()
|
|
# New file exists
|
|
assert (data_dir / "50-eth0.network").exists()
|
|
|
|
def test_cleanup_only_when_no_new_interfaces(self, tmp_network):
|
|
"""Only stale cleanup, no new files."""
|
|
data_dir = tmp_network / "data" / "networkd"
|
|
data_dir.mkdir(parents=True)
|
|
(data_dir / "stale.network").write_text("[Match]\n")
|
|
|
|
cfg = {"interfaces": {}}
|
|
_net.save_config(cfg)
|
|
result = _net.generate_network_files(cfg)
|
|
assert result["generated"] == []
|
|
assert len(result["cleaned"]) == 1
|
|
assert not (data_dir / "stale.network").exists()
|
|
|
|
|
|
# =================================================================
|
|
# TF-11: Extended DHCPv4 tests
|
|
# =================================================================
|
|
|
|
|
|
class TestDHCPv4Extended:
|
|
def test_dhcpv4_all_keys(self):
|
|
entry = {
|
|
"dhcp_client": {
|
|
"hostname": "myhost",
|
|
"duid_type": "llt",
|
|
"duid_raw_data": "01:02:03",
|
|
"iaid": "04:05:06:07",
|
|
"client_identifier": "aa:bb:cc",
|
|
"rapid_commit": True,
|
|
"anonymize": True,
|
|
"use_dns": False,
|
|
"use_ntp": True,
|
|
"use_sip": False,
|
|
"use_captive_portal": True,
|
|
"use_mtu": False,
|
|
"use_hostname": True,
|
|
"use_domains": "route",
|
|
"use_routes": False,
|
|
"route_metric": 200,
|
|
"send_decline": True,
|
|
"net_label": "mynet",
|
|
"nft_set": "myset",
|
|
"ip_service_type": "lowdelay",
|
|
"socket_priority": 10,
|
|
"bootp": False,
|
|
"label": "mylabel",
|
|
"max_attempts": 5,
|
|
"listen_port": 68,
|
|
"server_port": 67,
|
|
"mud_url": "https://example.com/mud.json",
|
|
"boot_filename": "boot.img",
|
|
"send_option": [
|
|
{"code": "5", "value": "10"},
|
|
"10 20",
|
|
],
|
|
"send_vendor_option": [
|
|
{"code": "1", "vendor_code": "2", "value": "3"},
|
|
"4 5 6",
|
|
],
|
|
"user_class": ["class1", "class2"],
|
|
"vendor_class_identifier": "vendor1",
|
|
"request_options": "1 3 6",
|
|
}
|
|
}
|
|
content = _net.render_network_file("eth0", entry)
|
|
assert "[DHCPv4]" in content
|
|
assert "Hostname=myhost" in content
|
|
assert "DUIDType=llt" in content
|
|
assert "DUIDRawData=01:02:03" in content
|
|
assert "IAID=04:05:06:07" in content
|
|
assert "ClientIdentifier=aa:bb:cc" in content
|
|
assert "RapidCommit=yes" in content
|
|
assert "Anonymize=yes" in content
|
|
assert "UseDNS=no" in content
|
|
assert "UseNTP=yes" in content
|
|
assert "UseSIP=no" in content
|
|
assert "UseCaptivePortal=yes" in content
|
|
assert "UseMTU=no" in content
|
|
assert "UseHostname=yes" in content
|
|
assert "UseDomains=route" in content
|
|
assert "UseRoutes=no" in content
|
|
assert "RouteMetric=200" in content
|
|
assert "SendDecline=yes" in content
|
|
assert "NetLabel=mynet" in content
|
|
assert "NFTSet=myset" in content
|
|
assert "IPServiceType=lowdelay" in content
|
|
assert "SocketPriority=10" in content
|
|
assert "BOOTP=no" in content
|
|
assert "Label=mylabel" in content
|
|
assert "MaxAttempts=5" in content
|
|
assert "ListenPort=68" in content
|
|
assert "ServerPort=67" in content
|
|
assert "MUDURL=https://example.com/mud.json" in content
|
|
assert "BootFilename=boot.img" in content
|
|
assert "SendOption=5 10" in content
|
|
assert "SendOption=10 20" in content
|
|
assert "SendVendorOption=1 2 3" in content
|
|
assert "SendVendorOption=4 5 6" in content
|
|
assert "UserClass=class1" in content
|
|
assert "UserClass=class2" in content
|
|
assert "VendorClassIdentifier=vendor1" in content
|
|
assert "RequestOptions=1 3 6" in content
|
|
|
|
def test_dhcpv4_only_when_dhcp_ipv4(self):
|
|
entry = {
|
|
"dhcp": "ipv4",
|
|
"dhcp_client": {"hostname": "test"},
|
|
}
|
|
content = _net.render_network_file("eth0", entry)
|
|
assert "[DHCPv4]" in content
|
|
assert "[DHCPv6]" not in content
|
|
|
|
def test_dhcpv4_only_when_dhcp_ipv6(self):
|
|
entry = {
|
|
"dhcp": "ipv6",
|
|
"dhcp_client": {"hostname": "test"},
|
|
}
|
|
content = _net.render_network_file("eth0", entry)
|
|
assert "[DHCPv4]" not in content
|
|
assert "[DHCPv6]" in content
|
|
|
|
def test_dhcpv4_and_v6_when_dhcp_yes(self):
|
|
entry = {
|
|
"dhcp": "yes",
|
|
"dhcp_client": {"hostname": "test"},
|
|
}
|
|
content = _net.render_network_file("eth0", entry)
|
|
assert "[DHCPv4]" in content
|
|
assert "[DHCPv6]" in content
|
|
|
|
|
|
# =================================================================
|
|
# TF-11: Extended DHCPv6 tests
|
|
# =================================================================
|
|
|
|
|
|
class TestDHCPv6Extended:
|
|
def test_dhcpv6_all_keys(self):
|
|
entry = {
|
|
"dhcp": "ipv6",
|
|
"dhcp_client": {
|
|
"send_hostname": True,
|
|
"hostname": "myhost",
|
|
"duid": "01:02",
|
|
"duid_type": "llt",
|
|
"duid_raw_data": "aa:bb",
|
|
"iaid": "01:02:03:04",
|
|
"anonymize": True,
|
|
"rapid_commit": "attempt-only",
|
|
"prefix_delegation_hint": "2001:db8::/48",
|
|
"unassigned_subnet_policy": "/64",
|
|
"use_address": True,
|
|
"use_captive_portal": False,
|
|
"use_delegated_prefix": True,
|
|
"use_dns": True,
|
|
"use_ntp": False,
|
|
"use_sip": True,
|
|
"use_dnr": False,
|
|
"use_hostname": True,
|
|
"use_domains": "route",
|
|
"send_release": True,
|
|
"net_label": "vlan6",
|
|
"nft_set": "ipv6set",
|
|
"without_ra": "ipv6",
|
|
"send_option": [
|
|
{"code": "1", "value": "2"},
|
|
"3 4",
|
|
],
|
|
"send_vendor_option": [
|
|
{"code": "10", "vendor_code": "20", "value": "30"},
|
|
"40 50 60",
|
|
],
|
|
"user_class": ["v6class"],
|
|
"vendor_class": ["v6vendor"],
|
|
},
|
|
}
|
|
content = _net.render_network_file("eth0", entry)
|
|
assert "[DHCPv6]" in content
|
|
assert "SendHostname=yes" in content
|
|
assert "Hostname=myhost" in content
|
|
assert "DUID=01:02" in content
|
|
assert "DUIDType=llt" in content
|
|
assert "DUIDRawData=aa:bb" in content
|
|
assert "IAID=01:02:03:04" in content
|
|
assert "Anonymize=yes" in content
|
|
assert "RapidCommit=attempt-only" in content
|
|
assert "PrefixDelegationHint=2001:db8::/48" in content
|
|
assert "UnassignedSubnetPolicy=/64" in content
|
|
assert "UseAddress=yes" in content
|
|
assert "UseCaptivePortal=no" in content
|
|
assert "UseDelegatedPrefix=yes" in content
|
|
assert "UseDNS=yes" in content
|
|
assert "UseNTP=no" in content
|
|
assert "UseSIP=yes" in content
|
|
assert "UseDNR=no" in content
|
|
assert "UseHostname=yes" in content
|
|
assert "UseDomains=route" in content
|
|
assert "SendRelease=yes" in content
|
|
assert "NetLabel=vlan6" in content
|
|
assert "NFTSet=ipv6set" in content
|
|
assert "WithoutRA=ipv6" in content
|
|
assert "SendOption=1 2" in content
|
|
assert "SendOption=3 4" in content
|
|
assert "SendVendorOption=10 20 30" in content
|
|
assert "SendVendorOption=40 50 60" in content
|
|
assert "UserClass=v6class" in content
|
|
assert "VendorClass=v6vendor" in content
|
|
|
|
|
|
# =================================================================
|
|
# TF-11: Extended Route tests
|
|
# =================================================================
|
|
|
|
|
|
class TestRouteExtended:
|
|
def test_route_all_keys(self):
|
|
entry = {
|
|
"routes": [
|
|
{
|
|
"destination": "10.0.0.0/8",
|
|
"gateway": "192.168.1.254",
|
|
"metric": 100,
|
|
"table": 100,
|
|
"type": "unicast",
|
|
"scope": "link",
|
|
"gateway_on_link": True,
|
|
"ipv6_preference": "medium",
|
|
"initial_congestion_window": 10,
|
|
"initial_advertised_receive_window": 60,
|
|
"quick_ack": True,
|
|
"fast_open_no_cookie": False,
|
|
"mtu_bytes": 1400,
|
|
"protocol": "static",
|
|
"next_hop": 1,
|
|
"multi_path_route": ["10.0.0.2", "10.0.0.3"],
|
|
},
|
|
{
|
|
"destination": "172.16.0.0/12",
|
|
"gateway": "10.0.0.254",
|
|
"metric": 200,
|
|
},
|
|
]
|
|
}
|
|
content = _net.render_network_file("eth0", entry)
|
|
assert "[Route]" in content
|
|
assert "[Route#1]" in content
|
|
assert "Destination=10.0.0.0/8" in content
|
|
assert "Gateway=192.168.1.254" in content
|
|
assert "Metric=100" in content
|
|
assert "Table=100" in content
|
|
assert "Type=unicast" in content
|
|
assert "Scope=link" in content
|
|
assert "GatewayOnLink=yes" in content
|
|
assert "IPv6Preference=medium" in content
|
|
assert "InitialCongestionWindow=10" in content
|
|
assert "InitialAdvertisedReceiveWindow=60" in content
|
|
assert "QuickAck=yes" in content
|
|
assert "FastOpenNoCookie=no" in content
|
|
assert "MTUBytes=1400" in content
|
|
assert "Protocol=static" in content
|
|
assert "NextHop=1" in content
|
|
assert "MultiPathRoute=10.0.0.2" in content
|
|
assert "MultiPathRoute=10.0.0.3" in content
|
|
assert "Destination=172.16.0.0/12" in content
|
|
assert "Metric=200" in content
|
|
assert "Gateway=10.0.0.254" in content
|
|
|
|
def test_route_integer_table(self):
|
|
entry = {
|
|
"routes": [
|
|
{"destination": "0.0.0.0/0", "gateway": "10.0.0.1", "table": "main"}
|
|
]
|
|
}
|
|
content = _net.render_network_file("eth0", entry)
|
|
assert "Table=main" in content
|
|
|
|
|
|
# =================================================================
|
|
# TF-11: collect_upstream_dns tests
|
|
# =================================================================
|
|
|
|
|
|
class TestCollectUpstreamDNS:
|
|
def test_collects_public_dns(self):
|
|
cfg = {
|
|
"interfaces": {
|
|
"eth0": {
|
|
"dns": ["8.8.8.8", "1.1.1.1"],
|
|
}
|
|
}
|
|
}
|
|
result = _net.collect_upstream_dns(cfg)
|
|
assert "8.8.8.8" in result
|
|
assert "1.1.1.1" in result
|
|
|
|
def test_filters_local_dns(self):
|
|
cfg = {
|
|
"interfaces": {
|
|
"eth0": {
|
|
"dns": ["8.8.8.8", "192.168.1.1", "10.0.0.1"],
|
|
}
|
|
}
|
|
}
|
|
result = _net.collect_upstream_dns(cfg)
|
|
assert "8.8.8.8" in result
|
|
assert "192.168.1.1" not in result
|
|
assert "10.0.0.1" not in result
|
|
|
|
def test_filters_ipv6_local_dns(self):
|
|
cfg = {
|
|
"interfaces": {
|
|
"eth0": {
|
|
"dns": ["8.8.8.8"],
|
|
"ipv6_dns": ["2001:4860:4860::8888", "fe80::1"],
|
|
}
|
|
}
|
|
}
|
|
result = _net.collect_upstream_dns(cfg)
|
|
assert "8.8.8.8" in result
|
|
assert "2001:4860:4860::8888" in result
|
|
assert "fe80::1" not in result
|
|
|
|
def test_deduplicates(self):
|
|
cfg = {
|
|
"interfaces": {
|
|
"eth0": {"dns": ["8.8.8.8"]},
|
|
"eth1": {"dns": ["8.8.8.8", "1.1.1.1"]},
|
|
}
|
|
}
|
|
result = _net.collect_upstream_dns(cfg)
|
|
assert result.count("8.8.8.8") == 1
|
|
assert "1.1.1.1" in result
|
|
|
|
def test_empty_config(self):
|
|
result = _net.collect_upstream_dns({"interfaces": {}})
|
|
assert result == []
|
|
|
|
def test_skips_non_dict_entries(self):
|
|
cfg = {"interfaces": {"bad": "not-a-dict"}}
|
|
result = _net.collect_upstream_dns(cfg)
|
|
assert result == []
|
|
|
|
def test_filters_loopback(self):
|
|
cfg = {
|
|
"interfaces": {
|
|
"eth0": {
|
|
"dns": ["8.8.8.8", "127.0.0.1", "::1"],
|
|
"ipv6_dns": ["::1", "2001:4860:4860::8844"],
|
|
}
|
|
}
|
|
}
|
|
result = _net.collect_upstream_dns(cfg)
|
|
assert "127.0.0.1" not in result
|
|
assert "::1" not in result
|
|
assert "8.8.8.8" in result
|
|
assert "2001:4860:4860::8844" in result
|
|
|
|
def test_filters_link_local(self):
|
|
cfg = {
|
|
"interfaces": {
|
|
"eth0": {
|
|
"dns": ["169.254.1.1", "8.8.4.4"],
|
|
}
|
|
}
|
|
}
|
|
result = _net.collect_upstream_dns(cfg)
|
|
assert "169.254.1.1" not in result
|
|
assert "8.8.4.4" in result
|
|
|
|
def test_filters_unicast_local(self):
|
|
cfg = {
|
|
"interfaces": {
|
|
"eth0": {
|
|
"ipv6_dns": ["fc00::1", "fd00::1", "2607:f8b0:4004:800::200e"],
|
|
}
|
|
}
|
|
}
|
|
result = _net.collect_upstream_dns(cfg)
|
|
assert "fc00::1" not in result
|
|
assert "fd00::1" not in result
|
|
assert "2607:f8b0:4004:800::200e" in result
|
|
|
|
|
|
# =================================================================
|
|
# TF-11: infer_dhcp_ranges tests
|
|
# =================================================================
|
|
|
|
|
|
class TestInferDhcpRanges:
|
|
def test_basic_subnet(self):
|
|
cfg = {
|
|
"interfaces": {
|
|
"eth0": {
|
|
"addresses": [{"address": "192.168.1.1/24"}],
|
|
}
|
|
}
|
|
}
|
|
result = _net.infer_dhcp_ranges(cfg)
|
|
assert "eth0" in result
|
|
r = result["eth0"]
|
|
assert r["prefix"] == 24
|
|
assert r["start"] == "192.168.1.1"
|
|
assert r["end"] == "192.168.1.254"
|
|
|
|
def test_bare_string_address(self):
|
|
cfg = {
|
|
"interfaces": {
|
|
"eth0": {
|
|
"addresses": ["192.168.1.1/24"],
|
|
}
|
|
}
|
|
}
|
|
result = _net.infer_dhcp_ranges(cfg)
|
|
assert "eth0" in result
|
|
|
|
def test_skips_ipv6_only(self):
|
|
cfg = {
|
|
"interfaces": {
|
|
"eth0": {
|
|
"addresses": [{"address": "fd00::1/64"}],
|
|
}
|
|
}
|
|
}
|
|
result = _net.infer_dhcp_ranges(cfg)
|
|
assert "eth0" not in result
|
|
|
|
def test_skips_non_network_address(self):
|
|
cfg = {
|
|
"interfaces": {
|
|
"eth0": {
|
|
"addresses": [{"address": "192.168.1.1"}],
|
|
}
|
|
}
|
|
}
|
|
result = _net.infer_dhcp_ranges(cfg)
|
|
assert "eth0" not in result
|
|
|
|
def test_empty_config(self):
|
|
result = _net.infer_dhcp_ranges({"interfaces": {}})
|
|
assert result == {}
|
|
|
|
def test_small_subnet_skipped(self):
|
|
cfg = {
|
|
"interfaces": {
|
|
"eth0": {
|
|
"addresses": [{"address": "192.168.1.0/31"}],
|
|
}
|
|
}
|
|
}
|
|
result = _net.infer_dhcp_ranges(cfg)
|
|
assert "eth0" not in result
|
|
|
|
def test_larger_subnet(self):
|
|
cfg = {
|
|
"interfaces": {
|
|
"eth0": {
|
|
"addresses": [{"address": "10.0.0.1/16"}],
|
|
}
|
|
}
|
|
}
|
|
result = _net.infer_dhcp_ranges(cfg)
|
|
assert "eth0" in result
|
|
r = result["eth0"]
|
|
assert r["prefix"] == 16
|
|
assert r["subnet"] == "10.0.0.0"
|
|
|
|
def test_skips_non_dict_entry(self):
|
|
cfg = {"interfaces": {"bad": "not-a-dict"}}
|
|
result = _net.infer_dhcp_ranges(cfg)
|
|
assert result == {}
|
|
|
|
|
|
# =================================================================
|
|
# TF-11: infer_zones tests
|
|
# =================================================================
|
|
|
|
|
|
class TestInferZones:
|
|
def test_wireguard_iface_is_wan(self):
|
|
cfg = {
|
|
"interfaces": {
|
|
"wg0": {},
|
|
}
|
|
}
|
|
result = _net.infer_zones(cfg)
|
|
assert result["wg0"] == "wan"
|
|
|
|
def test_dhcp_iface_is_wan(self):
|
|
cfg = {
|
|
"interfaces": {
|
|
"eth0": {"dhcp": "ipv4"},
|
|
}
|
|
}
|
|
result = _net.infer_zones(cfg)
|
|
assert result["eth0"] == "wan"
|
|
|
|
def test_dhcp_yes_is_wan(self):
|
|
cfg = {
|
|
"interfaces": {
|
|
"eth0": {"dhcp": "yes"},
|
|
}
|
|
}
|
|
result = _net.infer_zones(cfg)
|
|
assert result["eth0"] == "wan"
|
|
|
|
def test_routed_iface_is_management(self):
|
|
cfg = {
|
|
"interfaces": {
|
|
"eth0": {
|
|
"addresses": [{"address": "10.0.0.1/24"}],
|
|
"routes": [{"destination": "0.0.0.0/0", "gateway": "10.0.0.254"}],
|
|
}
|
|
}
|
|
}
|
|
result = _net.infer_zones(cfg)
|
|
assert result["eth0"] == "management"
|
|
|
|
def test_default_is_lan(self):
|
|
cfg = {
|
|
"interfaces": {
|
|
"eth1": {
|
|
"addresses": [{"address": "192.168.1.1/24"}],
|
|
}
|
|
}
|
|
}
|
|
result = _net.infer_zones(cfg)
|
|
assert result["eth1"] == "lan"
|
|
|
|
def test_empty_config(self):
|
|
result = _net.infer_zones({"interfaces": {}})
|
|
assert result == {}
|
|
|
|
def test_skips_non_dict_entry(self):
|
|
cfg = {"interfaces": {"bad": "not-a-dict"}}
|
|
result = _net.infer_zones(cfg)
|
|
assert result == {}
|
|
|
|
def test_multiple_interfaces_mixed(self):
|
|
cfg = {
|
|
"interfaces": {
|
|
"wg0": {},
|
|
"eth0": {"dhcp": "ipv4"},
|
|
"eth1": {"addresses": [{"address": "192.168.1.1/24"}]},
|
|
"eth2": {
|
|
"addresses": [{"address": "10.0.0.1/24"}],
|
|
"routes": [{"destination": "0.0.0.0/0", "gateway": "10.0.0.254"}],
|
|
},
|
|
}
|
|
}
|
|
result = _net.infer_zones(cfg)
|
|
assert result["wg0"] == "wan"
|
|
assert result["eth0"] == "wan"
|
|
assert result["eth1"] == "lan"
|
|
assert result["eth2"] == "management"
|