pigway-frame-flow/tests/test_production_deploy.py

"""Deployment gates, isolation checks, and rollback use mocked Docker only."""

import importlib.util
import json
from pathlib import Path
from types import SimpleNamespace

import pytest


@pytest.fixture
def tool():
    spec = importlib.util.spec_from_file_location(
        "production_deploy", Path(__file__).parents[1] / "scripts/production-deploy.py"
    )
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    instance = module.Deploy()
    instance.env = {
        "PROD_PROJECT": "production-frame",
        "PROD_CONTAINER": "production-frame",
        "PROD_IMAGE_REPO": "pigway-frame-flow",
        "PROD_DATA_DIR": "/confirmed/data",
        "PROD_CONFIG_DIR": "/confirmed/config",
        "PROD_WEB_BIND": "192.0.2.1",
        "PROD_WEB_PORT": "18787",
        "PROD_STATE_VOLUME": "production-frame-cd",
        "PROD_MOCK": "false",
        "PROD_WRITE_ENABLED": "false",
    }
    return instance


def existing(tool):
    return {
        "Image": "sha256:previous",
        "State": {"Running": True},
        "Config": {
            "User": "10333:10333",
            "Env": ["BUILD_COMMIT=previous", "FRAME_FLOW_TOKEN=fixture-token"],
            "Labels": {
                "com.docker.compose.project": tool.env["PROD_PROJECT"],
                "com.docker.compose.service": "pigway-frame-flow",
            },
        },
        "Mounts": [
            {"Type": "bind", "Destination": target, "Source": tool.env[key], "RW": True}
            for target, key in [
                ("/data", "PROD_DATA_DIR"),
                ("/config", "PROD_CONFIG_DIR"),
            ]
        ],
        "HostConfig": {
            "PortBindings": {
                "18787/tcp": [
                    {"HostIp": tool.env["PROD_WEB_BIND"], "HostPort": "18787"}
                ]
            },
            "NetworkMode": "production-frame_default",
            "RestartPolicy": {"Name": "unless-stopped"},
        },
    }


def test_preflight_preserves_paths_and_has_no_write_commands(tool, monkeypatch):
    monkeypatch.setattr(tool, "inspect", lambda _: existing(tool))
    calls = []
    monkeypatch.setattr(
        tool,
        "run",
        lambda *args, **kwargs: calls.append(args) or SimpleNamespace(returncode=1),
    )
    config = tool.preflight()
    service = config["services"]["pigway-frame-flow"]
    assert service["image"] == "sha256:previous"
    assert all(m["bind"]["create_host_path"] is False for m in service["volumes"])
    assert service["environment"]["FRAME_FLOW_TOKEN"] == "fixture-token"
    assert calls == [("volume", "inspect", "production-frame-cd")]


@pytest.mark.parametrize("change", ["path", "project", "port", "network"])
def test_preflight_rejects_wrong_host_resources(tool, monkeypatch, change):
    old = existing(tool)
    if change == "path":
        old["Mounts"][0]["Source"] = "/unrelated/data"
    elif change == "project":
        old["Config"]["Labels"]["com.docker.compose.project"] = "ffci-fixture"
    elif change == "port":
        old["HostConfig"]["PortBindings"]["18787/tcp"][0]["HostPort"] = "9999"
    else:
        old["HostConfig"]["NetworkMode"] = "host"
    monkeypatch.setattr(tool, "inspect", lambda _: old)
    with pytest.raises(ValueError):
        tool.preflight()


@pytest.mark.parametrize("failure", [None, "candidate", "rollback"])
def test_deploy_rolls_back_exact_baseline_without_deleting_data(
    tool, monkeypatch, failure
):
    old = {"services": {"pigway-frame-flow": {"image": "sha256:previous"}}}
    new = {"services": {"pigway-frame-flow": {"image": "candidate"}}}
    activated, calls, saved = [], [], []
    monkeypatch.setattr(tool, "preflight", lambda: old)
    monkeypatch.setattr(tool, "lock", lambda: calls.append("lock"))
    monkeypatch.setattr(tool, "unlock", lambda: calls.append("unlock"))
    monkeypatch.setattr(tool, "run", lambda *args, **kw: calls.append(args))
    monkeypatch.setattr(tool, "build", lambda _: new)
    monkeypatch.setattr(tool, "diagnostics", lambda: None)
    monkeypatch.setattr(tool, "activate", lambda c: activated.append(c))
    monkeypatch.setattr(
        tool, "state_io", lambda state: saved.append(json.loads(json.dumps(state)))
    )
    verifies = []

    def verify(config):
        verifies.append(config)
        if config == new and failure:
            raise RuntimeError("candidate not ready")
        if config == old and failure == "rollback" and len(verifies) > 1:
            raise RuntimeError("rollback not ready")
        return {"commit": "previous", "version": "0.1.0"}

    monkeypatch.setattr(tool, "verify", verify)
    if failure:
        with pytest.raises(RuntimeError):
            tool.deploy()
        assert activated == [new, old]
        assert saved[0]["previous"] == old
        assert ("unlock" in calls) is (failure != "rollback")
    else:
        tool.deploy()
        assert activated == [new]
        assert saved[-1]["status"] == "success"
    assert not any(
        isinstance(c, tuple) and ("down" in c or "prune" in c) for c in calls
    )


def test_activate_escapes_dollar_secrets_and_never_down(tool, monkeypatch):
    calls = []

    def run(*args, **kwargs):
        calls.append(args)
        if "config" in args:
            path = Path(args[args.index("-f") + 1])
            assert "$$literal" in path.read_text()

    monkeypatch.setattr(tool, "run", run)
    tool.activate(
        {
            "services": {
                "pigway-frame-flow": {"environment": {"FRAME_FLOW_TOKEN": "$literal"}}
            }
        }
    )
    assert len(calls) == 2 and "up" in calls[1] and "--no-build" in calls[1]


def test_workflow_deploy_is_verify_gated_and_manual_plan_default(tool):
    import yaml

    workflow = yaml.safe_load(
        (Path(__file__).parents[1] / ".github/workflows/verify.yml").read_text()
    )
    job = workflow["jobs"]["deploy"]
    assert job["needs"] == "verify" and "success()" in job["if"]
    assert "refs/heads/main" in job["if"] and "pull_request" not in job["if"]
    assert job["env"]["PROD_DEPLOY_ENABLED"] == "${{ vars.PROD_DEPLOY_ENABLED }}"
    assert (
        workflow[True]["workflow_dispatch"]["inputs"]["deploy_action"]["default"]
        == "plan"
    )
    assert job["steps"][-1]["if"] == "always()"


def test_recovery_handles_missing_or_crashed_production_container(tool, monkeypatch):
    previous = {
        "name": tool.env["PROD_PROJECT"],
        "services": {
            "pigway-frame-flow": {
                "container_name": tool.env["PROD_CONTAINER"],
                "image": "sha256:old",
                "volumes": [
                    {"source": tool.env["PROD_DATA_DIR"], "target": "/data"},
                    {"source": tool.env["PROD_CONFIG_DIR"], "target": "/config"},
                ],
            }
        },
    }
    state = {"owner": tool.owner, "status": "prepared", "previous": previous}
    calls = []

    def run(*args, **kwargs):
        calls.append(args)
        if args[0] == "inspect":
            assert args[1].endswith("-deploy-lock")
            return SimpleNamespace(
                returncode=0,
                stdout=json.dumps(
                    [
                        {
                            "Id": "owned-lock",
                            "Image": "sha256:old",
                            "Config": {
                                "Labels": {"frameflow.deploy.owner": tool.owner}
                            },
                        }
                    ]
                ).encode(),
            )
        return SimpleNamespace(
            returncode=0,
            stdout=json.dumps(
                [{"Labels": {"frameflow.deploy.project": tool.env["PROD_PROJECT"]}}]
            ).encode(),
        )

    monkeypatch.setattr(tool, "run", run)

    def state_io(payload=None):
        if payload is not None:
            state.update(payload)
        return json.dumps(state).encode()

    monkeypatch.setattr(tool, "state_io", state_io)
    monkeypatch.setattr(
        tool, "rollback", lambda config: calls.append(("rollback", config))
    )
    tool.recover()
    assert ("rollback", previous) in calls
    assert ("rm", "owned-lock") in calls
    assert state["status"] == "rolled_back"


def test_lock_conflict_never_activates_production(tool, monkeypatch):
    monkeypatch.setattr(tool, "preflight", lambda: {})
    monkeypatch.setattr(tool, "activate", lambda _: pytest.fail("Production touched"))

    def fail(*args, **kwargs):
        raise RuntimeError("Lock already reserved")

    monkeypatch.setattr(tool, "run", fail)
    with pytest.raises(RuntimeError):
        tool.deploy()