394 lines
14 KiB
Python
394 lines
14 KiB
Python
"""Tests for daemon/handlers/network.py — handler endpoint logic."""
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from daemon.handlers.network import (
|
|
apply_all,
|
|
get_infer_dhcp_ranges,
|
|
get_infer_zones,
|
|
get_interface,
|
|
get_interfaces,
|
|
reload_interface,
|
|
save_interface,
|
|
set_sysctl,
|
|
)
|
|
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
|
|
|
|
|
|
# =================================================================
|
|
# TF-12: Handler tests
|
|
# =================================================================
|
|
|
|
|
|
class TestSaveInterface:
|
|
def test_save_interface_saves_config(self, tmp_network):
|
|
with (
|
|
patch("daemon.handlers.network.run") as mock_run,
|
|
patch(
|
|
"daemon.handlers.network.DATA_DIR", tmp_network / "data" / "networkd"
|
|
),
|
|
):
|
|
mock_run.return_value = "1: eth0 ethernet routable\n State: routable\n"
|
|
save_interface(
|
|
None,
|
|
{"name": "eth0", "addresses": ["10.0.0.1/24"], "gateway": "10.0.0.254"},
|
|
)
|
|
|
|
cfg = _net.get_config()
|
|
assert "eth0" in cfg["interfaces"]
|
|
assert cfg["interfaces"]["eth0"]["addresses"] == ["10.0.0.1/24"]
|
|
|
|
def test_save_interface_renders_file(self, tmp_network):
|
|
with (
|
|
patch("daemon.handlers.network.run") as mock_run,
|
|
patch(
|
|
"daemon.handlers.network.DATA_DIR", tmp_network / "data" / "networkd"
|
|
),
|
|
):
|
|
mock_run.return_value = "1: eth0 ethernet routable\n State: routable\n"
|
|
save_interface(
|
|
None,
|
|
{"name": "eth0", "addresses": ["10.0.0.1/24"]},
|
|
)
|
|
|
|
data_dir = tmp_network / "data" / "networkd"
|
|
assert (data_dir / "99-eth0.network").exists()
|
|
content = (data_dir / "99-eth0.network").read_text()
|
|
assert "Name=eth0" in content
|
|
assert "Address=10.0.0.1/24" in content
|
|
|
|
def test_save_interface_requires_name(self, tmp_network):
|
|
with pytest.raises(ValueError, match="name"):
|
|
save_interface(None, {"addresses": ["10.0.0.1/24"]})
|
|
|
|
def test_save_interface_requires_body(self):
|
|
with pytest.raises(ValueError, match="body"):
|
|
save_interface(None, None)
|
|
|
|
def test_save_interface_rejects_invalid_name(self, tmp_network):
|
|
invalid_names = [
|
|
"../../etc/passwd",
|
|
"eth 0",
|
|
"",
|
|
"eth/0",
|
|
"eth..0",
|
|
]
|
|
for invalid in invalid_names:
|
|
with (
|
|
patch("daemon.handlers.network.run") as mock_run,
|
|
patch(
|
|
"daemon.handlers.network.DATA_DIR",
|
|
tmp_network / "data" / "networkd",
|
|
),
|
|
):
|
|
mock_run.return_value = (
|
|
"1: eth0 ethernet routable\n State: routable\n"
|
|
)
|
|
with pytest.raises(ValueError, match="name"):
|
|
save_interface(None, {"name": invalid})
|
|
|
|
|
|
class TestReloadInterfaceValidation:
|
|
def test_reload_interface_rejects_invalid_name(self):
|
|
invalid_names = [
|
|
"../../etc/passwd",
|
|
"eth 0",
|
|
"",
|
|
"eth/0",
|
|
]
|
|
for invalid in invalid_names:
|
|
with pytest.raises(ValueError, match="name"):
|
|
reload_interface(None, {"name": invalid})
|
|
|
|
|
|
class TestGetInterfaceValidation:
|
|
def test_get_interface_rejects_invalid_name(self, tmp_network):
|
|
invalid_names = [
|
|
"../../etc/passwd",
|
|
"eth 0",
|
|
"",
|
|
"eth/0",
|
|
]
|
|
for invalid in invalid_names:
|
|
with pytest.raises(ValueError, match="name"):
|
|
get_interface(None, {"name": invalid})
|
|
|
|
|
|
class TestReloadInterface:
|
|
def test_reload_interface(self):
|
|
with patch("daemon.handlers.network.run") as mock_run:
|
|
mock_run.return_value = "reloaded"
|
|
result = reload_interface(None, {"name": "eth0"})
|
|
|
|
assert result["name"] == "eth0"
|
|
assert result["reloaded"] is True
|
|
mock_run.assert_called_with(
|
|
["networkctl", "reconfigure", "eth0"], sudo=True
|
|
)
|
|
|
|
def test_reload_interface_requires_name(self):
|
|
with pytest.raises(ValueError, match="name"):
|
|
reload_interface(None, None)
|
|
|
|
def test_reload_interface_missing_name(self):
|
|
with pytest.raises(ValueError, match="name"):
|
|
reload_interface(None, {})
|
|
|
|
|
|
class TestApplyAll:
|
|
def test_apply_all_generates_files(self, tmp_network):
|
|
_net.save_config(
|
|
{
|
|
"interfaces": {
|
|
"eth0": {"addresses": ["10.0.0.1/24"]},
|
|
"eth1": {"addresses": ["192.168.1.1/24"]},
|
|
}
|
|
}
|
|
)
|
|
|
|
with (
|
|
patch("daemon.handlers.network.generate_network_files") as mock_gen,
|
|
patch("daemon.handlers.network.run") as mock_run,
|
|
patch("daemon.handlers.network.collect_upstream_dns", return_value=[]),
|
|
):
|
|
mock_gen.return_value = {
|
|
"generated": [_net.DATA_DIR / "99-eth0.network"],
|
|
"cleaned": [],
|
|
}
|
|
mock_run.return_value = ""
|
|
|
|
result = apply_all(None, None)
|
|
|
|
assert result["applied"] == 1
|
|
assert len(result["files"]) == 1
|
|
|
|
def test_apply_all_syncs_dns(self, tmp_network):
|
|
_net.save_config(
|
|
{
|
|
"interfaces": {
|
|
"eth0": {
|
|
"dhcp": "ipv4",
|
|
"dns": ["8.8.8.8", "192.168.1.1"],
|
|
}
|
|
}
|
|
}
|
|
)
|
|
|
|
with (
|
|
patch("daemon.handlers.network.generate_network_files") as mock_gen,
|
|
patch("daemon.handlers.network.run") as mock_run,
|
|
patch("daemon.handlers.network.set_upstreams") as mock_set_upstreams,
|
|
patch("daemon.handlers.network.collect_upstream_dns") as mock_collect,
|
|
):
|
|
mock_gen.return_value = {
|
|
"generated": [_net.DATA_DIR / "99-eth0.network"],
|
|
"cleaned": [],
|
|
}
|
|
mock_run.return_value = ""
|
|
mock_collect.return_value = ["8.8.8.8"]
|
|
|
|
apply_all(None, None)
|
|
|
|
mock_set_upstreams.assert_called_once_with(["8.8.8.8"])
|
|
|
|
def test_apply_all_handles_dns_sync_failure(self, tmp_network):
|
|
_net.save_config({"interfaces": {"eth0": {"dns": ["8.8.8.8"]}}})
|
|
|
|
with (
|
|
patch("daemon.handlers.network.generate_network_files") as mock_gen,
|
|
patch("daemon.handlers.network.run") as mock_run,
|
|
patch(
|
|
"daemon.handlers.network.set_upstreams",
|
|
side_effect=RuntimeError("fail"),
|
|
),
|
|
patch("daemon.handlers.network.collect_upstream_dns") as mock_collect,
|
|
):
|
|
mock_gen.return_value = {
|
|
"generated": [_net.DATA_DIR / "99-eth0.network"],
|
|
"cleaned": [],
|
|
}
|
|
mock_run.return_value = ""
|
|
mock_collect.return_value = ["8.8.8.8"]
|
|
|
|
result = apply_all(None, None)
|
|
assert "applied" in result
|
|
|
|
def test_apply_all_removes_stale_system_files(self, tmp_network):
|
|
_net.save_config({"interfaces": {"eth0": {"addresses": ["10.0.0.1/24"]}}})
|
|
sys_dir = tmp_network / "etc" / "systemd" / "network"
|
|
sys_dir.mkdir(parents=True)
|
|
(sys_dir / "stale-file.network").write_text("[Match]\nName=old\n")
|
|
|
|
with (
|
|
patch("daemon.handlers.network.generate_network_files") as mock_gen,
|
|
patch("daemon.handlers.network.run") as mock_run,
|
|
patch("daemon.handlers.network.collect_upstream_dns", return_value=[]),
|
|
):
|
|
mock_gen.return_value = {
|
|
"generated": [_net.DATA_DIR / "99-eth0.network"],
|
|
"cleaned": [],
|
|
}
|
|
mock_run.return_value = ""
|
|
|
|
class FakePath:
|
|
def __init__(self, p="/etc/systemd/network") -> None:
|
|
self._p = sys_dir if p == "/etc/systemd/network" else Path(p)
|
|
|
|
def exists(self):
|
|
return True
|
|
|
|
def iterdir(self):
|
|
return iter(self._p.iterdir())
|
|
|
|
def __truediv__(self, other):
|
|
return self._p / other
|
|
|
|
def mkdir(self, *args, **kwargs) -> None:
|
|
self._p.mkdir(parents=True, exist_ok=True)
|
|
|
|
with patch("daemon.handlers.network.Path", FakePath):
|
|
apply_all(None, None)
|
|
|
|
assert (sys_dir / "stale-file.network").exists()
|
|
|
|
|
|
class TestGetInterfaces:
|
|
def test_get_interfaces_returns_merged_data(self, tmp_network):
|
|
_net.save_config({"interfaces": {"eth0": {"addresses": ["10.0.0.1/24"]}}})
|
|
|
|
with patch("daemon.handlers.network.run") as mock_run:
|
|
mock_run.return_value = (
|
|
"1: eth0 ethernet 10.0.0.0/24 routable\n"
|
|
" State: routable\n"
|
|
" Addresses: 10.0.0.1/24,\n"
|
|
)
|
|
result = get_interfaces(None, None)
|
|
|
|
assert "interfaces" in result
|
|
assert "eth0" in result["interfaces"]
|
|
assert "config" in result["interfaces"]["eth0"]
|
|
assert "runtime" in result["interfaces"]["eth0"]
|
|
|
|
def test_get_interfaces_handles_networkctl_failure(self, tmp_network):
|
|
_net.save_config({"interfaces": {"eth0": {"addresses": ["10.0.0.1/24"]}}})
|
|
|
|
with patch(
|
|
"daemon.handlers.network.run", side_effect=RuntimeError("no networkctl")
|
|
):
|
|
result = get_interfaces(None, None)
|
|
|
|
assert "interfaces" in result
|
|
assert "eth0" in result["interfaces"]
|
|
|
|
|
|
class TestGetInterface:
|
|
def test_get_single_interface(self, tmp_network):
|
|
_net.save_config({"interfaces": {"eth0": {"addresses": ["10.0.0.1/24"]}}})
|
|
|
|
with patch("daemon.handlers.network.run") as mock_run:
|
|
mock_run.return_value = "1: eth0 ethernet\n State: routable\n"
|
|
result = get_interface(None, {"name": "eth0"})
|
|
|
|
assert result["name"] == "eth0"
|
|
assert "config" in result
|
|
assert result["config"]["addresses"] == ["10.0.0.1/24"]
|
|
|
|
def test_get_interface_not_found(self, tmp_network):
|
|
_net.save_config({"interfaces": {}})
|
|
|
|
with pytest.raises(Exception, match="not found"):
|
|
get_interface(None, {"name": "eth0"})
|
|
|
|
def test_get_interface_requires_name(self):
|
|
with pytest.raises(ValueError, match="required"):
|
|
get_interface(None, None)
|
|
|
|
|
|
class TestInferEndpoints:
|
|
def test_infer_dhcp_ranges_endpoint(self, tmp_network):
|
|
_net.save_config(
|
|
{
|
|
"interfaces": {
|
|
"eth0": {"addresses": [{"address": "192.168.1.1/24"}]},
|
|
}
|
|
}
|
|
)
|
|
result = get_infer_dhcp_ranges(None, None)
|
|
assert "ranges" in result
|
|
assert "eth0" in result["ranges"]
|
|
|
|
def test_infer_zones_endpoint(self, tmp_network):
|
|
_net.save_config(
|
|
{
|
|
"interfaces": {
|
|
"wg0": {},
|
|
"eth0": {"addresses": [{"address": "192.168.1.1/24"}]},
|
|
}
|
|
}
|
|
)
|
|
result = get_infer_zones(None, None)
|
|
assert "zones" in result
|
|
assert result["zones"]["wg0"] == "wan"
|
|
assert result["zones"]["eth0"] == "lan"
|
|
|
|
|
|
class TestSetSysctl:
|
|
def test_set_sysctl_success(self):
|
|
with (
|
|
patch("daemon.handlers.network.run") as mock_run,
|
|
patch.object(Path, "read_text", return_value="1"),
|
|
):
|
|
mock_run.return_value = "" # sysctl -w call
|
|
result = set_sysctl(None, {"name": "net.ipv4.ip_forward", "value": "1"})
|
|
|
|
assert result["name"] == "net.ipv4.ip_forward"
|
|
assert result["value"] == "1"
|
|
assert mock_run.call_count == 1
|
|
assert mock_run.call_args_list[0].args == (
|
|
["sysctl", "-w", "net.ipv4.ip_forward=1"],
|
|
)
|
|
assert mock_run.call_args_list[0].kwargs == {"sudo": True}
|
|
|
|
def test_set_sysctl_rejects_slash_in_name(self):
|
|
with pytest.raises(ValueError, match="valid sysctl key"):
|
|
set_sysctl(None, {"name": "net.ipv4/ip_forward", "value": "1"})
|
|
|
|
def test_set_sysctl_rejects_double_dot(self):
|
|
with pytest.raises(ValueError, match="valid sysctl key"):
|
|
set_sysctl(None, {"name": "net..ipv4", "value": "1"})
|
|
|
|
def test_set_sysctl_requires_name(self):
|
|
with pytest.raises(ValueError, match="name"):
|
|
set_sysctl(None, {"value": "1"})
|
|
|
|
def test_set_sysctl_requires_value(self):
|
|
with pytest.raises(ValueError, match="value"):
|
|
set_sysctl(None, {"name": "net.ipv4.ip_forward"})
|
|
|
|
def test_set_sysctl_requires_body(self):
|
|
with pytest.raises(ValueError, match="body"):
|
|
set_sysctl(None, None)
|
|
|
|
def test_set_sysctl_verify_failure(self):
|
|
with (
|
|
patch("daemon.handlers.network.run") as mock_run,
|
|
patch.object(Path, "read_text", return_value="0"),
|
|
):
|
|
mock_run.return_value = "" # sysctl -w call succeeds
|
|
with pytest.raises(RuntimeError, match="verify failed"):
|
|
set_sysctl(None, {"name": "net.ipv4.ip_forward", "value": "1"})
|