zima-apps/Apps/docker-ip-addr-manager/tests/integration_tests.py

#!/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.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))


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):
    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"],
    )


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 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)

    print("Integration tests passed")


if __name__ == "__main__":
    main()