pigway-frame-flow/tests/test_server_release.py

"""Production release preflight must reject unsafe inherited container settings."""

import importlib.util
from pathlib import Path

import pytest


@pytest.fixture
def release_tool():
    spec = importlib.util.spec_from_file_location(
        "server_release", Path(__file__).parents[1] / "scripts/server-release.py"
    )
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


@pytest.mark.parametrize(
    "unsafe", ["mock", "writes", "photos", "system_disk", "root", "caps"]
)
def test_release_rejects_unsafe_runtime(release_tool, monkeypatch, unsafe):
    container = {
        "Config": {
            "User": "10333:10333",
            "Env": [
                "KOKONNA_MOCK=false",
                "KOKONNA_WRITE_ENABLED=true",
                "BUILD_COMMIT=test",
            ],
        },
        "State": {"Running": True},
        "HostConfig": {"ReadonlyRootfs": True, "CapDrop": ["ALL"]},
        "Mounts": [
            {"Destination": "/data", "Source": str(release_tool.APP / "data")},
            {"Destination": "/config", "Source": str(release_tool.APP / "config")},
        ],
    }
    if unsafe == "mock":
        container["Config"]["Env"][0] = "KOKONNA_MOCK=true"
    elif unsafe == "writes":
        container["Config"]["Env"][1] = "KOKONNA_WRITE_ENABLED=false"
    elif unsafe == "photos":
        container["Mounts"].append(
            {"Destination": "/photos", "Source": "/volume1/photo"}
        )
    elif unsafe == "system_disk":
        container["Mounts"][0]["Source"] = "/var/lib/frame-flow"
    elif unsafe == "root":
        container["Config"]["User"] = "root"
    else:
        container["HostConfig"]["CapDrop"] = []
    monkeypatch.setattr(release_tool, "inspect", lambda: container)
    with pytest.raises(AssertionError):
        release_tool.verify_runtime("test")


def test_release_readiness_retries_on_actual_listener(release_tool, monkeypatch):
    import io
    import urllib.error

    calls = []

    def listener(url, timeout):
        calls.append(url)
        if url != "http://192.0.2.10:81/ready" or len(calls) == 1:
            raise urllib.error.URLError("Connection refused during restart")
        return io.BytesIO(b'{"status":"ready"}')

    monkeypatch.setattr(release_tool.urllib.request, "urlopen", listener)
    monkeypatch.setattr(release_tool.time, "sleep", lambda _: None)
    release_tool.wait_ready()
    assert len(calls) == 2