govoplan/tests/test_managed_ingress_drill.py

from __future__ import annotations

import importlib.util
from pathlib import Path
import json
import subprocess
import sys
import unittest
from unittest.mock import patch


ROOT = Path(__file__).resolve().parents[1]


def _load_module():
    path = ROOT / "tools/checks/managed-ingress-drill.py"
    spec = importlib.util.spec_from_file_location("managed_ingress_drill", path)
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


INGRESS = _load_module()


class ManagedIngressDrillTests(unittest.TestCase):
    def test_config_is_streamed_into_a_daemon_visible_volume(self) -> None:
        completed = subprocess.CompletedProcess([], 0, "", "")
        with patch.object(INGRESS, "_run", return_value=completed) as run:
            INGRESS._write_volume_file(
                image="registry.example/caddy@sha256:" + "1" * 64,
                volume="config-volume",
                filename="Caddyfile",
                content=":8080 { respond /health 200 }\n",
            )

        argv = run.call_args.args[0]
        self.assertIn("type=volume,src=config-volume,dst=/govoplan-config", argv)
        self.assertIn("0:0", argv)
        self.assertNotIn("type=bind", " ".join(argv))
        self.assertEqual(
            ":8080 { respond /health 200 }\n",
            run.call_args.kwargs["input_text"],
        )

    def test_config_filename_cannot_escape_the_volume(self) -> None:
        with self.assertRaisesRegex(ValueError, "invalid config filename"):
            INGRESS._write_volume_file(
                image="registry.example/caddy@sha256:" + "1" * 64,
                volume="config-volume",
                filename="../Caddyfile",
                content="",
            )

    def test_drill_has_no_runner_local_bind_mounts(self) -> None:
        source = (ROOT / "tools/checks/managed-ingress-drill.py").read_text(
            encoding="utf-8"
        )

        self.assertNotIn("type=bind", source)
        self.assertIn('"--network-alias",\n                    "load-balancer"', source)
        self.assertNotIn('"127.0.0.1::8080"', source)
        self.assertIn("requested_http_port", source)
        self.assertIn("requested_https_port", source)
        self.assertIn('"--cap-add",\n                "NET_BIND_SERVICE"', source)

    def test_published_port_reads_the_docker_mapping(self) -> None:
        completed = subprocess.CompletedProcess(
            [],
            0,
            json.dumps(
                {
                    "8443/tcp": [
                        {"HostIp": "127.0.0.1", "HostPort": "49152"}
                    ]
                }
            ),
            "",
        )
        with patch.object(INGRESS, "_run", return_value=completed) as run:
            port = INGRESS._published_port("ingress", 8443)

        self.assertEqual(49152, port)
        self.assertEqual(
            [
                "docker",
                "inspect",
                "--format",
                "{{json .HostConfig.PortBindings}}",
                "ingress",
            ],
            run.call_args.args[0],
        )

    def test_published_port_rejects_non_loopback_binding(self) -> None:
        completed = subprocess.CompletedProcess(
            [],
            0,
            '{"8443/tcp":[{"HostIp":"0.0.0.0","HostPort":"49152"}]}',
            "",
        )
        with patch.object(INGRESS, "_run", return_value=completed):
            with self.assertRaisesRegex(RuntimeError, "loopback binding"):
                INGRESS._published_port("ingress", 8443)

    def test_probe_runs_as_a_network_sibling_from_a_digest_image(self) -> None:
        completed = subprocess.CompletedProcess([], 0, "", "")
        image = "registry.example/runtime-api@sha256:" + "1" * 64
        with patch.object(INGRESS, "_run", return_value=completed) as run:
            INGRESS._probe_ingress(
                image=image,
                network="deployment-network",
                container="ingress",
            )

        argv = run.call_args.args[0]
        self.assertEqual("docker", argv[0])
        self.assertIn("deployment-network", argv)
        self.assertIn(image, argv)
        self.assertIn('(\"ingress\", port)', argv[-1])
        self.assertIn("server_hostname=\"localhost\"", argv[-1])
        self.assertNotIn("localhost:49152", argv[-1])

    def test_probe_diagnostics_include_container_stderr(self) -> None:
        probe_failure = subprocess.CalledProcessError(1, ["docker", "run"])
        state = subprocess.CompletedProcess([], 0, '{"Running":false}', "")
        logs = subprocess.CompletedProcess([], 0, "", "caddy startup failed")
        with patch.object(
            INGRESS,
            "_run",
            side_effect=[probe_failure, state, logs],
        ), patch.object(INGRESS.sys, "stderr") as stderr:
            with self.assertRaises(subprocess.CalledProcessError):
                INGRESS._probe_ingress(
                    image="registry.example/runtime-api@sha256:" + "1" * 64,
                    network="deployment-network",
                    container="ingress",
                )

        rendered = "".join(call.args[0] for call in stderr.write.call_args_list)
        self.assertIn('"Running":false', rendered)
        self.assertIn("caddy startup failed", rendered)

    def test_standalone_workflow_is_dispatch_only_and_digest_bounded(self) -> None:
        workflow = (
            ROOT / ".gitea/workflows/runtime-ingress-drill.yml"
        ).read_text(encoding="utf-8")

        self.assertIn("workflow_dispatch:", workflow)
        self.assertNotIn("\n  push:", workflow)
        self.assertIn("--probe-image \"$PROBE_IMAGE\"", workflow)
        self.assertIn("GOVOPLAN_REGISTRY_TOKEN", workflow)


if __name__ == "__main__":
    unittest.main()