jfriberg
1b35702b1b
Co-authored-by: Joachim Friberg <joachim.friberg@ip-solutions.se> Reviewed-on: phirna/zima-apps#6
291 lines
10 KiB
Python
291 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
|
|
|
|
ROOT_DIR = Path(__file__).resolve().parents[1]
|
|
BACKEND_DIR = ROOT_DIR / "backend"
|
|
sys.path.insert(0, str(BACKEND_DIR))
|
|
|
|
from app.docker_api import DockerApiError, DockerUsageResolver
|
|
from app.dns_sync import DnsSyncError
|
|
from app.models import IpEntry
|
|
from app.service import ConflictError, DependencyError, EntryService
|
|
from app.storage import EntryStorage
|
|
|
|
|
|
class FakeDockerClient:
|
|
def list_containers(self, include_stopped: bool = True):
|
|
assert include_stopped is True
|
|
return [
|
|
{"Id": "a1", "Names": ["/alpha"]},
|
|
{"Id": "b2", "Names": ["/beta"]},
|
|
]
|
|
|
|
def inspect_container(self, container_id: str):
|
|
if container_id == "a1":
|
|
return {
|
|
"Name": "/alpha",
|
|
"NetworkSettings": {
|
|
"Ports": {
|
|
"8080/tcp": [{"HostIp": "10.0.4.2", "HostPort": "8080"}],
|
|
"8443/tcp": [{"HostIp": "0.0.0.0", "HostPort": "8443"}],
|
|
}
|
|
},
|
|
}
|
|
if container_id == "b2":
|
|
return {
|
|
"Name": "/beta",
|
|
"NetworkSettings": {
|
|
"Ports": {
|
|
"9000/tcp": [{"HostIp": "10.0.4.3", "HostPort": "9000"}],
|
|
}
|
|
},
|
|
}
|
|
raise AssertionError(f"unexpected container_id: {container_id}")
|
|
|
|
|
|
class FakeUsageResolver:
|
|
def __init__(self, mapping=None, should_fail=False):
|
|
self._mapping = mapping or {}
|
|
self._should_fail = should_fail
|
|
|
|
def resolve_ip_usage(self, ips: set[str]):
|
|
if self._should_fail:
|
|
raise DockerApiError("docker unreachable")
|
|
response = {ip: set() for ip in ips}
|
|
for ip in ips:
|
|
response[ip] = set(self._mapping.get(ip, set()))
|
|
return response
|
|
|
|
|
|
class FakeIpManager:
|
|
def __init__(self):
|
|
self.present = set()
|
|
|
|
def is_present(self, ip: str, cidr: int, device: str):
|
|
return (ip, cidr, device) in self.present
|
|
|
|
def ensure_present(self, ip: str, cidr: int, device: str):
|
|
self.present.add((ip, cidr, device))
|
|
|
|
def ensure_absent(self, ip: str, cidr: int, device: str):
|
|
self.present.discard((ip, cidr, device))
|
|
|
|
|
|
class FakeDnsProvider:
|
|
def __init__(self, fail_upsert=False, fail_delete=False):
|
|
self.records = {}
|
|
self.upserts = []
|
|
self.deletes = []
|
|
self.fail_upsert = fail_upsert
|
|
self.fail_delete = fail_delete
|
|
|
|
def upsert_a_record(self, fqdn: str, ip: str, ttl: int):
|
|
if self.fail_upsert:
|
|
raise DnsSyncError("upsert failed")
|
|
self.records[fqdn] = ip
|
|
self.upserts.append((fqdn, ip, ttl))
|
|
|
|
def delete_a_record(self, fqdn: str):
|
|
if self.fail_delete:
|
|
raise DnsSyncError("delete failed")
|
|
self.records.pop(fqdn, None)
|
|
self.deletes.append(fqdn)
|
|
|
|
|
|
def assert_true(condition, message):
|
|
if not condition:
|
|
raise AssertionError(message)
|
|
|
|
|
|
def build_service(
|
|
tmp_path: Path,
|
|
entries=None,
|
|
usage_resolver=None,
|
|
ip_manager=None,
|
|
dns_provider=None,
|
|
dns_base_domain="home.arpa",
|
|
):
|
|
storage = EntryStorage(str(tmp_path / "entries.json"))
|
|
if entries:
|
|
storage.save_entries(entries)
|
|
|
|
resolver = usage_resolver or FakeUsageResolver()
|
|
ipm = ip_manager or FakeIpManager()
|
|
|
|
return EntryService(
|
|
storage=storage,
|
|
usage_resolver=resolver,
|
|
ip_manager=ipm,
|
|
interface_provider=lambda: ["eth0", "eth1"],
|
|
dns_provider=dns_provider,
|
|
dns_base_domain=dns_base_domain,
|
|
dns_ttl_seconds=120,
|
|
)
|
|
|
|
|
|
def test_exact_hostip_match_only():
|
|
resolver = DockerUsageResolver(FakeDockerClient())
|
|
usage = resolver.resolve_ip_usage({"10.0.4.2", "10.0.4.3"})
|
|
|
|
assert_true(usage["10.0.4.2"] == {"alpha"}, "10.0.4.2 should match alpha")
|
|
assert_true(usage["10.0.4.3"] == {"beta"}, "10.0.4.3 should match beta")
|
|
|
|
|
|
def test_disable_blocked_when_used(tmp_path: Path):
|
|
entry = IpEntry(id="one", name="lan-a", ip="10.0.4.2", cidr=16, device="eth0", enabled=True)
|
|
resolver = FakeUsageResolver(mapping={"10.0.4.2": {"alpha"}})
|
|
ip_manager = FakeIpManager()
|
|
ip_manager.present.add(("10.0.4.2", 16, "eth0"))
|
|
|
|
service = build_service(tmp_path, entries=[entry], usage_resolver=resolver, ip_manager=ip_manager)
|
|
|
|
blocked = False
|
|
try:
|
|
service.set_enabled("one", enabled=False)
|
|
except ConflictError:
|
|
blocked = True
|
|
|
|
assert_true(blocked, "disable must be blocked when entry is used")
|
|
assert_true(("10.0.4.2", 16, "eth0") in ip_manager.present, "IP must stay present when disable is blocked")
|
|
|
|
|
|
def test_disable_blocked_when_docker_check_fails(tmp_path: Path):
|
|
entry = IpEntry(id="two", name="lan-b", ip="10.0.4.8", cidr=16, device="eth0", enabled=True)
|
|
resolver = FakeUsageResolver(should_fail=True)
|
|
ip_manager = FakeIpManager()
|
|
ip_manager.present.add(("10.0.4.8", 16, "eth0"))
|
|
|
|
service = build_service(tmp_path, entries=[entry], usage_resolver=resolver, ip_manager=ip_manager)
|
|
|
|
blocked = False
|
|
try:
|
|
service.set_enabled("two", enabled=False)
|
|
except DependencyError:
|
|
blocked = True
|
|
|
|
assert_true(blocked, "disable must fail-closed when Docker usage check fails")
|
|
|
|
|
|
def test_delete_blocked_when_enabled(tmp_path: Path):
|
|
entry = IpEntry(id="three", name="lan-c", ip="10.0.4.9", cidr=16, device="eth0", enabled=True)
|
|
service = build_service(tmp_path, entries=[entry])
|
|
|
|
blocked = False
|
|
try:
|
|
service.delete_entry("three")
|
|
except ConflictError:
|
|
blocked = True
|
|
|
|
assert_true(blocked, "delete must be blocked while entry is enabled")
|
|
|
|
|
|
def test_reconcile_reapplies_enabled(tmp_path: Path):
|
|
entry = IpEntry(id="four", name="lan-d", ip="10.0.4.10", cidr=16, device="eth0", enabled=True)
|
|
ip_manager = FakeIpManager()
|
|
service = build_service(tmp_path, entries=[entry], ip_manager=ip_manager)
|
|
|
|
errors = service.reconcile_enabled_entries()
|
|
|
|
assert_true(errors == [], "reconcile should succeed without errors")
|
|
assert_true(("10.0.4.10", 16, "eth0") in ip_manager.present, "enabled IP must be re-applied on startup reconcile")
|
|
|
|
|
|
def test_dns_upsert_on_enable_when_used(tmp_path: Path):
|
|
entry = IpEntry(id="dns1", name="Lan App", ip="10.0.4.20", cidr=16, device="eth0", enabled=False)
|
|
resolver = FakeUsageResolver(mapping={"10.0.4.20": {"nginx"}})
|
|
ip_manager = FakeIpManager()
|
|
dns = FakeDnsProvider()
|
|
service = build_service(tmp_path, entries=[entry], usage_resolver=resolver, ip_manager=ip_manager, dns_provider=dns)
|
|
|
|
service.set_enabled("dns1", enabled=True)
|
|
|
|
assert_true(dns.records.get("lan-app.home.arpa") == "10.0.4.20", "DNS record should be created on enable+used")
|
|
|
|
|
|
def test_dns_no_upsert_on_enable_when_unused(tmp_path: Path):
|
|
entry = IpEntry(id="dns2", name="Lan App 2", ip="10.0.4.21", cidr=16, device="eth0", enabled=False)
|
|
resolver = FakeUsageResolver(mapping={})
|
|
ip_manager = FakeIpManager()
|
|
dns = FakeDnsProvider()
|
|
service = build_service(tmp_path, entries=[entry], usage_resolver=resolver, ip_manager=ip_manager, dns_provider=dns)
|
|
|
|
service.set_enabled("dns2", enabled=True)
|
|
|
|
assert_true("lan-app-2.home.arpa" not in dns.records, "Unused entries must not create DNS record")
|
|
|
|
|
|
def test_dns_reconcile_deletes_when_no_longer_used(tmp_path: Path):
|
|
entry = IpEntry(id="dns3", name="Lan App 3", ip="10.0.4.22", cidr=16, device="eth0", enabled=True)
|
|
resolver = FakeUsageResolver(mapping={"10.0.4.22": {"nginx"}})
|
|
dns = FakeDnsProvider()
|
|
service = build_service(tmp_path, entries=[entry], usage_resolver=resolver, dns_provider=dns)
|
|
|
|
service.reconcile_dns_records()
|
|
assert_true(dns.records.get("lan-app-3.home.arpa") == "10.0.4.22", "record should exist after used reconcile")
|
|
|
|
resolver._mapping = {}
|
|
service.reconcile_dns_records()
|
|
assert_true("lan-app-3.home.arpa" not in dns.records, "record should be removed when usage disappears")
|
|
|
|
|
|
def test_dns_fail_closed_rolls_back_enable(tmp_path: Path):
|
|
entry = IpEntry(id="dns4", name="Lan App 4", ip="10.0.4.23", cidr=16, device="eth0", enabled=False)
|
|
resolver = FakeUsageResolver(mapping={"10.0.4.23": {"nginx"}})
|
|
ip_manager = FakeIpManager()
|
|
dns = FakeDnsProvider(fail_upsert=True)
|
|
service = build_service(tmp_path, entries=[entry], usage_resolver=resolver, ip_manager=ip_manager, dns_provider=dns)
|
|
|
|
failed = False
|
|
try:
|
|
service.set_enabled("dns4", enabled=True)
|
|
except ConflictError:
|
|
failed = True
|
|
|
|
assert_true(failed, "enable must fail-closed when DNS upsert fails")
|
|
current = service.list_entries().items[0]
|
|
assert_true(not current.enabled, "entry must roll back to disabled on DNS failure")
|
|
assert_true(not ip_manager.is_present("10.0.4.23", 16, "eth0"), "IP presence must roll back on DNS failure")
|
|
|
|
|
|
def test_dns_fail_closed_blocks_delete(tmp_path: Path):
|
|
entry = IpEntry(id="dns5", name="Lan App 5", ip="10.0.4.24", cidr=16, device="eth0", enabled=False)
|
|
resolver = FakeUsageResolver(mapping={})
|
|
dns = FakeDnsProvider(fail_delete=True)
|
|
service = build_service(tmp_path, entries=[entry], usage_resolver=resolver, dns_provider=dns)
|
|
|
|
failed = False
|
|
try:
|
|
service.delete_entry("dns5")
|
|
except ConflictError:
|
|
failed = True
|
|
|
|
assert_true(failed, "delete must fail-closed when DNS cleanup fails")
|
|
assert_true(len(service.list_entries().items) == 1, "entry must remain when delete fails")
|
|
|
|
|
|
def main():
|
|
test_exact_hostip_match_only()
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
tmp_path = Path(tmp)
|
|
test_disable_blocked_when_used(tmp_path)
|
|
test_disable_blocked_when_docker_check_fails(tmp_path)
|
|
test_delete_blocked_when_enabled(tmp_path)
|
|
test_reconcile_reapplies_enabled(tmp_path)
|
|
test_dns_upsert_on_enable_when_used(tmp_path)
|
|
test_dns_no_upsert_on_enable_when_unused(tmp_path)
|
|
test_dns_reconcile_deletes_when_no_longer_used(tmp_path)
|
|
test_dns_fail_closed_rolls_back_enable(tmp_path)
|
|
test_dns_fail_closed_blocks_delete(tmp_path)
|
|
|
|
print("Integration tests passed")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|