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 != release_tool.ORIGIN + "/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


def test_release_requires_private_storage_identity_before_host_actions(
    release_tool, monkeypatch
):
    calls = []
    monkeypatch.setattr(release_tool.os, "geteuid", lambda: 0)
    monkeypatch.setattr(release_tool, "run", lambda *args: calls.append(args))
    monkeypatch.delenv("FRAME_FLOW_MOUNT_UUID", raising=False)
    with pytest.raises(AssertionError, match="FRAME_FLOW_MOUNT_UUID"):
        release_tool.main()
    assert calls == []


def test_release_rejects_wrong_storage_uuid(release_tool, monkeypatch):
    monkeypatch.setattr(release_tool.os, "geteuid", lambda: 0)
    monkeypatch.setenv("FRAME_FLOW_MOUNT_UUID", "expected-storage-uuid")
    monkeypatch.setenv("FRAME_FLOW_ORIGIN", release_tool.ORIGIN)
    monkeypatch.setattr(release_tool, "run", lambda *args: "different-storage-uuid")
    with pytest.raises(AssertionError):
        release_tool.main()


@pytest.mark.parametrize("invalid", [None, "id", "commit", "version", "content"])
def test_prebuilt_image_is_bound_to_source_before_deployment(
    release_tool, monkeypatch, invalid
):
    import json
    import subprocess

    digest = "sha256:" + "a" * 64
    monkeypatch.setenv("FRAME_FLOW_PREBUILT_IMAGE_ID", digest)
    metadata = {
        "Id": digest,
        "Config": {
            "Labels": {
                "org.opencontainers.image.revision": "fixture-commit",
                "org.opencontainers.image.version": "0.1.22",
            }
        },
    }
    if invalid == "id":
        metadata["Id"] = "sha256:" + "b" * 64
    if invalid == "commit":
        metadata["Config"]["Labels"]["org.opencontainers.image.revision"] = "wrong"
    if invalid == "version":
        metadata["Config"]["Labels"]["org.opencontainers.image.version"] = "wrong"
    monkeypatch.setattr(release_tool, "run", lambda *args: json.dumps([metadata]))
    calls = []

    def isolated(command, **kwargs):
        calls.append(command)
        assert "--network=none" in command and "--read-only" in command
        assert "--cap-drop=ALL" in command and "--user=10333:10333" in command
        assert "-v" not in command and "--mount" not in command
        assert json.loads(kwargs["input"]) == {"LICENSE": "fixture-hash"}
        if invalid == "content":
            raise subprocess.CalledProcessError(1, command)

    monkeypatch.setattr(release_tool.subprocess, "run", isolated)
    if invalid:
        with pytest.raises((AssertionError, subprocess.CalledProcessError)):
            release_tool.verify_prebuilt_image(
                "fixture-image", "fixture-commit", "0.1.22", {"LICENSE": "fixture-hash"}
            )
    else:
        assert release_tool.verify_prebuilt_image(
            "fixture-image", "fixture-commit", "0.1.22", {"LICENSE": "fixture-hash"}
        )
    if invalid in ("id", "commit", "version"):
        assert not calls