pigway-frame-flow/tests/test_production_bootstrap.py

"""First production bootstrap safety and failure paths, mocked Docker only."""

import json
from types import SimpleNamespace

import pytest
from test_production_deploy import existing
from test_production_deploy import tool as deploy_fixture


@pytest.fixture
def tool():
    return deploy_fixture.__wrapped__()


def residue(tool):
    value = existing(tool)
    value.update(Id="residue-id")
    value["State"] = {
        "Status": "created",
        "Running": False,
        "StartedAt": "0001-01-01T00:00:00Z",
    }
    for mount in value["Mounts"]:
        mount["Source"] = "/workspace/way/pigway-frame-flow" + mount["Destination"]
    return value


def prepare(tool, monkeypatch, old=None):
    tool.env["PROD_MOCK"] = "true"
    monkeypatch.setattr(tool, "container_optional", lambda: old)
    monkeypatch.setattr(
        tool,
        "host_paths",
        lambda paths: [{"exists": False, "empty": False} for _ in paths],
    )
    monkeypatch.setattr(
        tool, "run", lambda *args, **kw: SimpleNamespace(stdout=b"", returncode=0)
    )


@pytest.mark.parametrize("kind", ["absent", "residue"])
def test_bootstrap_accepts_only_empty_first_install(tool, monkeypatch, kind):
    old = residue(tool) if kind == "residue" else None
    prepare(tool, monkeypatch, old)
    assert tool.bootstrap_check() == old


@pytest.mark.parametrize(
    "kind",
    [
        "running",
        "stopped",
        "old-path",
        "extra-mount",
        "identity",
        "target-nonempty",
        "target-symlink",
        "journal",
    ],
)
def test_bootstrap_fail_closed(tool, monkeypatch, kind):
    old = residue(tool)
    prepare(tool, monkeypatch, old)
    if kind == "running":
        old["State"].update(Status="running", Running=True)
    elif kind == "stopped":
        old["State"].update(Status="exited", StartedAt="2026-10-01T00:00:00Z")
    elif kind == "extra-mount":
        old["Mounts"].append({"Type": "volume", "Destination": "/unknown"})
    elif kind == "identity":
        old["Config"]["Labels"] = {}
    elif kind == "journal":
        monkeypatch.setattr(
            tool,
            "run",
            lambda *a, **kw: SimpleNamespace(
                stdout=tool.env["PROD_STATE_VOLUME"].encode()
            ),
        )
    else:
        monkeypatch.setattr(
            tool,
            "host_paths",
            lambda paths: [
                {
                    "exists": kind == "old-path"
                    or not paths[0].startswith("/workspace"),
                    "empty": False,
                }
                for _ in paths
            ],
        )
    with pytest.raises(ValueError):
        tool.bootstrap_check()


@pytest.mark.parametrize("mock,writes", [("false", "false"), ("true", "true")])
def test_bootstrap_requires_mock_and_no_writes(tool, monkeypatch, mock, writes):
    prepare(tool, monkeypatch)
    tool.env.update(PROD_MOCK=mock, PROD_WRITE_ENABLED=writes)
    with pytest.raises(ValueError, match="mock=true"):
        tool.bootstrap_check()


@pytest.mark.parametrize("failure", [None, "ready", "smoke", "commit"])
def test_bootstrap_verifies_restart_and_preserves_data_on_failure(
    tool, monkeypatch, failure
):
    prepare(tool, monkeypatch)
    candidate = tool.bootstrap_config()
    candidate["services"]["pigway-frame-flow"]["image"] = "release-image"
    calls, saved, verifies = [], [], []
    monkeypatch.setattr(tool, "secret_config", lambda: {})
    monkeypatch.setattr(tool, "build", lambda config: candidate)
    monkeypatch.setattr(tool, "bootstrap_check", lambda: None)
    monkeypatch.setattr(tool, "lock", lambda: calls.append(("lock",)))
    monkeypatch.setattr(tool, "unlock", lambda: calls.append(("unlock",)))
    monkeypatch.setattr(tool, "state_io", lambda state: saved.append(dict(state)))
    monkeypatch.setattr(
        tool, "activate", lambda config: calls.append(("activate", config))
    )
    monkeypatch.setattr(tool, "preflight", lambda: calls.append(("preflight",)))
    monkeypatch.setattr(tool, "diagnostics", lambda: None)
    monkeypatch.setattr(
        tool,
        "inspect",
        lambda _: {"Image": "sha256:new", "State": {"Health": {"Status": "healthy"}}},
    )

    def run(*args, **kw):
        calls.append(args)
        output = b""
        if args[:2] == ("image", "inspect"):
            output = b'[{"Id":"sha256:new"}]'
        return SimpleNamespace(
            stdout=output, returncode=1 if args[:2] == ("network", "inspect") else 0
        )

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

    def verify(config):
        verifies.append(config)
        if failure:
            raise RuntimeError(failure)

    monkeypatch.setattr(tool, "verify", verify)
    if failure:
        with pytest.raises(RuntimeError, match=failure):
            tool.bootstrap()
        assert saved[-1]["status"] == "failed"
    else:
        tool.bootstrap()
        assert len(verifies) == 2
        assert ("restart", tool.env["PROD_CONTAINER"]) in calls
        assert saved[-1]["status"] == "success"
    assert not any("down" in c or "prune" in c for c in calls)
    assert not any(c[0] in ("rm", "volume") and "remove" in c for c in calls)


def test_bootstrap_config_is_adoptable_by_preflight(tool, monkeypatch):
    prepare(tool, monkeypatch)
    config = tool.bootstrap_config()
    runtime = existing(tool)
    monkeypatch.setattr(tool, "inspect", lambda _: runtime)
    monkeypatch.setattr(tool, "run", lambda *a, **kw: SimpleNamespace(returncode=1))
    adopted = tool.preflight()
    service = config["services"]["pigway-frame-flow"]
    other = adopted["services"]["pigway-frame-flow"]
    for key in ("volumes", "ports", "container_name", "restart", "user"):
        assert service[key] == other[key]
    assert config["name"] == adopted["name"]


def test_cleanup_only_removes_owned_container_and_empty_owned_network(
    tool, monkeypatch
):
    prepare(tool, monkeypatch)
    old = residue(tool)
    old["Config"]["Labels"]["frameflow.bootstrap.owner"] = tool.owner
    monkeypatch.setattr(tool, "container_optional", lambda: old)
    tool.state = {"network_created": True}
    calls = []

    def run(*args, **kw):
        calls.append(args)
        return SimpleNamespace(
            returncode=0,
            stdout=json.dumps(
                [
                    {
                        "Id": "network-id",
                        "Containers": {},
                        "Labels": {
                            "com.docker.compose.project": tool.env["PROD_PROJECT"]
                        },
                    }
                ]
            ).encode(),
        )

    monkeypatch.setattr(tool, "run", run)
    tool.bootstrap_cleanup()
    assert calls == [
        ("rm", "-f", "residue-id"),
        ("network", "inspect", "production-frame_default"),
        ("network", "rm", "network-id"),
    ]
    old["Config"]["Labels"].pop("frameflow.bootstrap.owner")
    with pytest.raises(ValueError):
        tool.bootstrap_cleanup()


def test_diagnostics_masks_all_secrets(tool, monkeypatch, capsys):
    tool.secret_values = ["test-secret"]
    monkeypatch.setattr(
        tool,
        "run",
        lambda *a, **kw: SimpleNamespace(
            returncode=1, stdout=b"test-secret https://private.invalid/key", stderr=b""
        ),
    )
    tool.diagnostics()
    output = capsys.readouterr().out
    assert "test-secret" not in output and "private.invalid" not in output


@pytest.mark.parametrize("failure", ["ready", "smoke", "commit"])
def test_actual_verify_rejects_ready_smoke_and_commit_failure(
    tool, monkeypatch, failure
):
    monkeypatch.setattr(
        tool.verify.__func__.__globals__["time"], "sleep", lambda _: None
    )
    config = {
        "services": {"pigway-frame-flow": {"environment": {"BUILD_COMMIT": "expected"}}}
    }

    def run(*args, **kw):
        if "-i" in args and failure == "smoke":
            raise RuntimeError("smoke failed")
        if "-i" not in args and failure == "ready":
            return SimpleNamespace(returncode=1)
        return SimpleNamespace(returncode=0, stdout=b'{"commit":"wrong"}')

    monkeypatch.setattr(tool, "run", run)
    with pytest.raises(RuntimeError):
        tool.verify(config)


def test_workflow_bootstrap_manual_only_and_ci_namespace_separate(tool):
    from pathlib import Path

    import yaml

    root = Path(__file__).parents[1]
    workflow = yaml.safe_load((root / ".github/workflows/verify.yml").read_text())
    inputs = workflow[True]["workflow_dispatch"]["inputs"]["deploy_action"]
    assert inputs["options"] == ["plan", "bootstrap", "deploy", "rollback"]
    assert inputs["default"] == "plan"
    assert workflow["jobs"]["deploy"]["needs"] == "verify"
    assert "success()" in workflow["jobs"]["deploy"]["if"]
    tool.env["PROD_PROJECT"] = "ffci-invalid"
    with pytest.raises(ValueError):
        tool.settings()


def test_automatic_push_without_production_never_creates_it(tool, monkeypatch, capsys):
    import importlib.util
    from pathlib import Path

    spec = importlib.util.spec_from_file_location(
        "bootstrap_main", Path(__file__).parents[1] / "scripts/production-deploy.py"
    )
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    tool.env.update(PROD_DEPLOY_ENABLED="true", GITHUB_EVENT_NAME="push")
    monkeypatch.setattr(module, "Deploy", lambda: tool)
    monkeypatch.setattr(tool, "container_optional", lambda: None)
    monkeypatch.setattr(tool, "deploy", lambda: pytest.fail("Production touched"))
    monkeypatch.setattr(tool, "bootstrap", lambda: pytest.fail("Production touched"))
    module.main()
    assert "bootstrap required" in capsys.readouterr().out
    tool.env["CD_ACTION"] = "bootstrap"
    with pytest.raises(SystemExit, match="workflow_dispatch"):
        module.main()


def test_bootstrap_recovery_cleans_only_owned_prepared_transaction(tool, monkeypatch):
    prepare(tool, monkeypatch)
    calls = []
    tool.state = {
        "owner": tool.owner,
        "kind": "bootstrap",
        "status": "prepared",
        "network_created": True,
    }

    def run(*args, **kw):
        calls.append(args)
        value = [{"Labels": {"frameflow.deploy.project": tool.env["PROD_PROJECT"]}}]
        if args[0] == "inspect":
            value = [
                {
                    "Id": "lock-id",
                    "Image": "built-image",
                    "Config": {"Labels": {"frameflow.deploy.owner": tool.owner}},
                }
            ]
        return SimpleNamespace(returncode=0, stdout=json.dumps(value).encode())

    monkeypatch.setattr(tool, "run", run)
    monkeypatch.setattr(tool, "bootstrap_cleanup", lambda: calls.append(("cleanup",)))

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

    monkeypatch.setattr(tool, "state_io", state_io)
    tool.recover()
    assert tool.state["status"] == "failed"
    assert ("cleanup",) in calls and ("rm", "lock-id") in calls


@pytest.mark.parametrize("status", ["failed", "prepared", "success"])
def test_retry_journal_only_failed_matching_empty_bootstrap(tool, monkeypatch, status):
    prepare(tool, monkeypatch)
    record = {
        "kind": "bootstrap",
        "status": status,
        "candidate": tool.bootstrap_config(),
    }

    def run(*args, **kw):
        if args[:2] == ("volume", "ls"):
            return SimpleNamespace(stdout=tool.env["PROD_STATE_VOLUME"].encode())
        return SimpleNamespace(
            stdout=json.dumps(
                [{"Labels": {"frameflow.deploy.project": tool.env["PROD_PROJECT"]}}]
            ).encode()
        )

    monkeypatch.setattr(tool, "run", run)
    monkeypatch.setattr(tool, "state_io", lambda: json.dumps(record).encode())
    if status == "failed":
        assert tool.bootstrap_check() is None
        record["candidate"]["services"]["pigway-frame-flow"]["volumes"][0]["source"] = (
            "/foreign/data"
        )
    with pytest.raises(ValueError):
        tool.bootstrap_check()


def test_synology_initialization_mounts_only_existing_parents_and_creates_leaves(
    tool, monkeypatch, tmp_path
):
    calls = []
    monkeypatch.setattr(tool, "run", lambda *a, **kw: calls.append((a, kw)))
    tool.initialize_directories()
    args, kw = calls[0]
    assert "-v" not in args and "--network" in args
    assert [args[i + 1] for i, value in enumerate(args) if value == "--mount"] == [
        "type=bind,source=/confirmed,target=/parent-data",
        "type=bind,source=/confirmed,target=/parent-config",
    ]
    code = args[args.index("-c") + 1]
    import subprocess
    import sys

    paths = [str(tmp_path / "data"), str(tmp_path / "config")]
    # Keep the production permission operations in the script; replace only UID for local test.
    code = code.replace("os.chown(p,10333,10333)", "None")
    result = subprocess.run(
        [sys.executable, "-c", code],
        input=json.dumps(paths).encode(),
        capture_output=True,
    )
    assert result.returncode == 0
    assert all(
        (tmp_path / name).stat().st_mode & 0o777 == 0o750 for name in ("data", "config")
    )
    (tmp_path / "data" / "keep").write_text("must survive")
    result = subprocess.run(
        [sys.executable, "-c", code],
        input=json.dumps(paths).encode(),
        capture_output=True,
    )
    assert (
        result.returncode != 0
        and (tmp_path / "data" / "keep").read_text() == "must survive"
    )
    tool.env["PROD_DATA_DIR"] = "/data"
    with pytest.raises(ValueError, match="NAS root"):
        tool.initialize_directories()