teleo-infrastructure/tests/test_run_approve_claim_clone_canary_macos_path.py

"""Platform-path regression tests for the approved-change clone canary."""

from __future__ import annotations

import hashlib
import json
import subprocess
import sys
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "scripts"))
WRAPPER = ROOT / "scripts" / "run_approve_claim_isolated_container_canary.sh"
FINALIZER = ROOT / "scripts" / "finalize_approve_claim_isolated_canary.py"

import apply_proposal as apply  # noqa: E402
import finalize_approve_claim_isolated_canary as finalizer  # noqa: E402
import run_approve_claim_clone_canary as canary  # noqa: E402


def test_role_psql_resolves_docker_before_scrubbing_path(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    captured: dict[str, object] = {}

    monkeypatch.setattr(
        canary.shutil,
        "which",
        lambda executable: "/opt/homebrew/bin/docker" if executable == "docker" else None,
    )

    def fake_run(command: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]:
        captured["command"] = command
        captured["env"] = kwargs["env"]
        return subprocess.CompletedProcess(command, 0, "", "")

    monkeypatch.setattr(canary, "_run", fake_run)

    canary._role_psql(
        "local-postgres",
        "teleo",
        "select 1;",
        role="kb_apply",
        password="test-only-password",
    )

    assert captured["command"][0] == "/opt/homebrew/bin/docker"
    assert captured["env"] == {
        "PGPASSWORD": "test-only-password",
        "PATH": "/usr/bin:/bin:/usr/local/bin",
    }


def test_apply_tool_resolves_docker_before_scrubbing_path(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    captured: dict[str, object] = {}
    monkeypatch.setattr(
        apply.shutil,
        "which",
        lambda executable: "/opt/homebrew/bin/docker" if executable == "docker" else None,
    )

    def fake_run(command: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]:
        captured["command"] = command
        captured["env"] = kwargs["env"]
        return subprocess.CompletedProcess(command, 0, "ok", "")

    monkeypatch.setattr(apply.subprocess, "run", fake_run)
    args = type(
        "Args",
        (),
        {
            "container": "local-postgres",
            "role": "kb_apply",
            "host": "127.0.0.1",
            "db": "teleo",
        },
    )()

    assert apply.run_psql(args, "select 1;", "test-only-password") == "ok"
    assert captured["command"][0] == "/opt/homebrew/bin/docker"
    assert captured["env"] == {
        "PGPASSWORD": "test-only-password",
        "PATH": "/usr/bin:/bin:/usr/local/bin",
    }


def test_outer_wrapper_enforces_local_source_and_networkless_tmpfs() -> None:
    wrapper = WRAPPER.read_text(encoding="utf-8")
    finalizer = FINALIZER.read_text(encoding="utf-8")
    text = wrapper + "\n" + finalizer

    assert 'source_canary_label" != "working-leo-approved-source"' in text
    assert "--network none" in text
    assert "--tmpfs /var/lib/postgresql/data:rw,nosuid,size=512m" in text
    assert 'CANARY_NETWORK_MODE="$container_network_mode"' in text
    assert 'CANARY_MOUNTS_JSON="$container_mounts_json"' in text
    assert 'checks["outer_source_tier_enforced"]' in text
    assert 'checks["outer_container_network_disabled"]' in text
    assert 'checks["outer_container_has_no_docker_volumes"]' in text
    assert 'source_tier != "live-readonly" or service_observable' in text
    assert "<<'PY'" not in wrapper
    assert "finalize_approve_claim_isolated_canary.py" in wrapper


def test_outer_finalizer_runs_from_a_normal_file_without_heredoc(tmp_path: Path) -> None:
    content = FINALIZER.read_bytes()
    report = tmp_path / "canary.json"
    report.write_text(
        json.dumps(
            {
                "status": "pass",
                "checks": {},
                "source_binding": {
                    "unchanged_during_run": True,
                    "files": [
                        {
                            "repo_relative_path": FINALIZER.relative_to(ROOT).as_posix(),
                            "sha256": hashlib.sha256(content).hexdigest(),
                            "size_bytes": len(content),
                        }
                    ],
                },
            }
        ),
        encoding="utf-8",
    )
    counts = json.dumps({"public.claims": 2, "kb_stage.kb_proposals": 0})
    data = finalizer.finalize(
        report,
        {
            "SOURCE_COUNTS_BEFORE": counts,
            "SOURCE_COUNTS_AFTER": counts,
            "SOURCE_CONTAINER": "isolated-source",
            "SOURCE_DB": "teleo",
            "SERVICE_BEFORE": "Available=false\nReason=systemctl_unavailable\n",
            "SERVICE_AFTER": "Available=false\nReason=systemctl_unavailable\n",
            "REPO_ROOT": str(ROOT),
            "OUTER_CONTAINER_REMOVE_RC": "0",
            "OUTER_WORKDIR_REMOVE_RC": "0",
            "CONTAINER_NAME_READBACK": "",
            "CONTAINER_NAME_READBACK_RC": "0",
            "CONTAINER_LABEL_READBACK": "",
            "CONTAINER_LABEL_READBACK_RC": "0",
            "CONTAINER_LABEL_ORPHAN_COUNT": "0",
            "OUTER_WORKDIR_LEXISTS": "false",
            "CANARY_RC": "0",
            "SOURCE_TIER": "isolated-local",
            "SOURCE_NETWORK_MODE": "none",
            "SOURCE_CANARY_LABEL": "working-leo-approved-source",
            "CANARY_NETWORK_MODE": "none",
            "CANARY_MOUNTS_JSON": "[]",
            "CANARY_LABEL": "proposal-apply-lifecycle",
            "CANARY_INSTANCE_LABEL": "working-leo-approve-claim-test",
        },
    )

    assert data["status"] == "pass"
    assert data["current_tier"] == "T2_runtime"
    assert data["checks"]["outer_isolation_complete"] is True
    assert data["outer_isolation"]["cleanup_readback"]["label_scoped_orphan_count"] == 0