playground/docker_to_caddy_test.py

#!/usr/bin/env -S uv run --script
# /// script
# dependencies = ["pytest"]
# ///

import copy
from pathlib import Path
import subprocess
import sys
from unittest.mock import patch

import pytest

import docker_to_caddy as script


def container(name: str = "app", domain: str = "ascii.app.abdus.dev") -> dict:
    return {
        "Name": f"/{name}",
        "Config": {"Labels": {"caddy.domain": domain}},
        "State": {"Running": True, "Status": "running"},
        "HostConfig": {"PortBindings": {
            "3000/tcp": [{"HostIp": "127.0.0.1", "HostPort": "4510"}],
        }},
        "NetworkSettings": {"Ports": {
            "3000/tcp": [{"HostIp": "127.0.0.1", "HostPort": "4510"}],
        }},
    }


@pytest.mark.parametrize("address, expected", [
    ("127.0.0.1", "127.0.0.1:4510"),
    ("0.0.0.0", "127.0.0.1:4510"),
    ("::", "[::1]:4510"),
    ("::1", "[::1]:4510"),
    ("192.168.1.2", "192.168.1.2:4510"),
])
def test_bindings(address: str, expected: str) -> None:
    app = container()
    app["NetworkSettings"]["Ports"]["3000/tcp"][0]["HostIp"] = address
    assert f"reverse_proxy {expected}" in script.render_config(containers=[app])


@pytest.mark.parametrize("selector, succeeds", [(None, False), ("3000", True), ("9000", False)])
def test_port_selection(selector: str | None, succeeds: bool) -> None:
    app = container()
    app["NetworkSettings"]["Ports"]["4000/tcp"] = [
        {"HostIp": "::1", "HostPort": "4600"},
    ]
    if selector is not None:
        app["Config"]["Labels"]["caddy.port"] = selector
    if succeeds:
        assert "127.0.0.1:4510" in script.render_config(containers=[app])
    else:
        with pytest.raises(ValueError):
            script.render_config(containers=[app])


@pytest.mark.parametrize("domain", ["", "bad host.dev", "example.dev {", "*.abdus.dev", "https://abdus.dev"])
def test_invalid_domains(domain: str) -> None:
    with pytest.raises(ValueError):
        script.render_config(containers=[container(domain=domain)])


def test_duplicate_domains() -> None:
    with pytest.raises(ValueError):
        script.render_config(containers=[container(), container(name="other")])


def test_missing_published_ports() -> None:
    app = container()
    app["NetworkSettings"]["Ports"] = {}
    with pytest.raises(ValueError):
        script.render_config(containers=[app])


def test_host_networking() -> None:
    app = container()
    app["HostConfig"]["NetworkMode"] = "host"
    with pytest.raises(ValueError):
        script.render_config(containers=[app])


def test_stopped_unlabeled_and_empty() -> None:
    stopped = container()
    stopped["State"] = {"Running": False, "Status": "exited"}
    unlabeled = container()
    unlabeled["Config"]["Labels"] = None
    assert script.render_config(containers=[stopped, unlabeled]) == script.render_config(containers=[])


def test_deterministic_and_ipv4_preference() -> None:
    first = container(name="a", domain="a.abdus.dev")
    second = container(name="z", domain="z.abdus.dev")
    first["NetworkSettings"]["Ports"]["3000/tcp"].append(
        {"HostIp": "::", "HostPort": "4510"},
    )
    forward = script.render_config(containers=[first, second])
    assert forward == script.render_config(containers=[second, first])
    assert "reverse_proxy 127.0.0.1:4510" in forward
    assert forward.index("a.abdus.dev") < forward.index("z.abdus.dev")


@pytest.mark.parametrize("left, right, protocol, port, expected", [
    ("0.0.0.0", "127.0.0.1", "tcp", "4510", True),
    ("127.0.0.1", "127.0.0.2", "tcp", "4510", False),
    ("::", "127.0.0.1", "tcp", "4510", True),
    ("127.0.0.1", "127.0.0.1", "udp", "4510", False),
    ("127.0.0.1", "127.0.0.1", "tcp", "0", False),
])
def test_clashes(left: str, right: str, protocol: str, port: str, expected: bool) -> None:
    first = container()
    first["HostConfig"]["PortBindings"]["3000/tcp"][0]["HostIp"] = left
    first["NetworkSettings"]["Ports"] = copy.deepcopy(first["HostConfig"]["PortBindings"])
    second = container(name="stopped")
    second["Config"]["Labels"] = None
    second["State"] = {"Running": False, "Status": "exited"}
    second["NetworkSettings"]["Ports"] = {}
    second["HostConfig"]["PortBindings"] = {
        f"3000/{protocol}": [{"HostIp": right, "HostPort": port}],
    }
    with patch.object(script.logging, "warning") as warning:
        script.warn_port_clashes(containers=[first, second])
    assert warning.call_count == int(expected)


def test_output_and_failure_preservation(tmp_path: Path) -> None:
    output = tmp_path / "generated.caddy"
    output.write_text(data="previous", encoding="utf-8")
    output.chmod(mode=0o640)
    with patch("sys.argv", ["docker_to_caddy.py", "--output", str(output)]):
        with patch.object(script, "inspect_containers", return_value=[container(), container()]):
            assert script.main() == 1
        assert output.read_text() == "previous"
        with patch.object(script, "inspect_containers", side_effect=subprocess.CalledProcessError(
            returncode=1, cmd="docker", stderr="permission denied",
        )):
            assert script.main() == 1
        assert output.read_text() == "previous"
    script.write_config(config="replacement", output=output)
    assert output.read_text() == "replacement"
    assert output.stat().st_mode & 0o777 == 0o640


def test_stdout(capsys: pytest.CaptureFixture[str]) -> None:
    with patch("sys.argv", ["docker_to_caddy.py"]):
        with patch.object(script, "inspect_containers", return_value=[container()]):
            assert script.main() == 0
    assert capsys.readouterr().out == script.render_config(containers=[container()])


def test_docker_discovery() -> None:
    responses = [
        subprocess.CompletedProcess(args=[], returncode=0, stdout="id1\nid2\n"),
        subprocess.CompletedProcess(args=[], returncode=0, stdout='[{"Name":"/one"},{"Name":"/two"}]'),
    ]
    with patch.object(script.subprocess, "run", side_effect=responses) as run:
        assert script.inspect_containers() == [{"Name": "/one"}, {"Name": "/two"}]
    assert run.call_args_list[1].kwargs["args"] == ["docker", "container", "inspect", "id1", "id2"]
    with patch.object(script.subprocess, "run", return_value=subprocess.CompletedProcess(
        args=[], returncode=0, stdout="",
    )) as run:
        assert script.inspect_containers() == []
        assert run.call_count == 1


if __name__ == "__main__":
    sys.exit(pytest.main(args=[str(Path(__file__).resolve()), *sys.argv[1:]]))