teleo-infrastructure/tests/test_gcp_generated_db_direct_claim_suite.py

import argparse
import asyncio
import json
import os
import subprocess
from pathlib import Path

import pytest

from scripts.run_gcp_generated_db_direct_claim_suite import (
    REVIEWED_CLOUDSQL_TOOL_SHA256,
    REVIEWED_MANIFEST_SQL_SHA256,
    build_cloudsql_wrapper,
    database_fingerprint,
    direct_prompts,
    normalize_database_manifest_rows,
    run_suite,
    validate_database_identity,
    validate_parity_receipt,
)
from scripts.run_leo_clone_bound_handler_checkpoint import (
    READ_ONLY_BRIDGE_COMMANDS,
    read_tool_proof,
    validate_read_only_bridge_argv,
)


def test_direct_prompt_catalog_is_the_exact_six_case_suite() -> None:
    prompts = direct_prompts()

    assert [prompt["id"] for prompt in prompts] == [f"DC-{index:02d}" for index in range(1, 7)]
    assert "status" not in READ_ONLY_BRIDGE_COMMANDS
    validate_read_only_bridge_argv(["list-proposals", "--help"])
    validate_read_only_bridge_argv(["list-proposals", "--status", "rejected"])


def test_wrapper_binds_private_read_only_database_and_records_calls(tmp_path: Path) -> None:
    wrapper = build_cloudsql_wrapper(
        cloudsql_tool=tmp_path / "cloudsql_memory_tool.py",
        tool_log=tmp_path / "calls.jsonl",
        target_db="teleo_clone_test",
        host="10.61.0.3",
        project="teleo-501523",
        password_secret="test-secret-name",
        run_nonce="nonce",
    )

    assert "dbname=$TARGET_DATABASE" in wrapper
    assert '--db "$TARGET_DATABASE"' in wrapper
    assert '--canonical-db "$TARGET_DATABASE"' in wrapper
    assert "sslmode=require" in wrapper
    assert "begin transaction read only" in wrapper
    assert 'export PGOPTIONS="-c default_transaction_read_only=on"' in wrapper
    assert "default_transaction_read_only" in wrapper
    assert "host(inet_server_addr())" in wrapper
    assert '"phase": "start"' in wrapper
    assert '"phase": "end"' in wrapper
    assert "unset PGPASSWORD" in wrapper
    assert "systemctl restart" not in wrapper
    assert "send_message" not in wrapper


def test_database_identity_requires_private_tls_and_read_only_target() -> None:
    validate_database_identity(
        {
            "current_database": "teleo_clone_test",
            "system_identifier": "1234",
            "server_address": "10.61.0.3",
            "ssl": True,
            "transaction_read_only": "on",
            "default_transaction_read_only": "on",
        },
        "teleo_clone_test",
    )

    with pytest.raises(RuntimeError, match="not RFC1918-private"):
        validate_database_identity(
            {
                "current_database": "teleo_clone_test",
                "system_identifier": "1234",
                "server_address": "34.1.2.3",
                "ssl": True,
                "transaction_read_only": "on",
                "default_transaction_read_only": "on",
            },
            "teleo_clone_test",
        )


def test_manifest_normalization_excludes_nondeterministic_timing() -> None:
    output = "\n".join(
        [
            'BEGIN',
            '{"kind":"identity","database":"teleo_clone_test","captured_at":"2026-07-12T00:00:00Z"}',
            '{"kind":"schemas","items":["kb_stage","public"]}',
            '{"kind":"table","schema":"public","table":"claims","row_count":2,"rowset_md5":"abc"}',
            '{"kind":"performance","query":"count_claims","elapsed_ms":1.2,"observed_rows":2}',
            'ROLLBACK',
        ]
    )

    rows = normalize_database_manifest_rows(output)

    assert all(row.get("kind") != "performance" for row in rows)
    assert next(row for row in rows if row["kind"] == "identity").get("captured_at") is None


def test_database_fingerprint_uses_full_normalized_manifest(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    manifest_sql = tmp_path / "manifest.sql"
    manifest_sql.write_text("select 1;\n")
    args = type(
        "Args",
        (),
        {
            "host": "10.61.0.3",
            "target_db": "teleo_clone_test",
            "manifest_sql": manifest_sql,
        },
    )()
    output = "\n".join(
        [
            '{"kind":"identity","database":"teleo_clone_test","server_address":"10.61.0.3",'
            '"ssl":true,"transaction_read_only":"on","captured_at":"one"}',
            '{"kind":"schemas","items":["kb_stage","public"]}',
            '{"kind":"table","schema":"public","table":"claims","row_count":2,"rowset_md5":"abc"}',
        ]
    )
    monkeypatch.setattr("scripts.run_gcp_generated_db_direct_claim_suite.database_password", lambda _args: "secret")
    monkeypatch.setattr("scripts.run_gcp_generated_db_direct_claim_suite.run", lambda *_args, **_kwargs: output)

    receipt = database_fingerprint(args)

    assert receipt["table_count"] == 1
    assert receipt["total_rows"] == 2
    assert receipt["database"] == "teleo_clone_test"
    assert len(receipt["sha256"]) == 64


def test_reviewed_helper_hashes_match_repo_files() -> None:
    import hashlib

    assert (
        hashlib.sha256(Path("hermes-agent/leoclean-bin/cloudsql_memory_tool.py").read_bytes()).hexdigest()
        == REVIEWED_CLOUDSQL_TOOL_SHA256
    )
    assert hashlib.sha256(Path("ops/postgres_parity_manifest.sql").read_bytes()).hexdigest() == REVIEWED_MANIFEST_SQL_SHA256


def parity_payload(target_db: str = "teleo_clone_test") -> dict[str, object]:
    return {
        "artifact": "canonical_postgres_parity_verification",
        "status": "pass",
        "scope": "local_restore",
        "problems": [],
        "details": {
            "target_database": target_db,
            "source_table_count": 2,
            "target_table_count": 2,
            "source_total_rows": 3,
            "target_total_rows": 3,
            "table_mismatches": [],
            "application_role_mismatches": {},
            "required_extension_mismatches": {},
            "performance_problems": [],
            "structural_hashes": {"columns": {"source": "same", "target": "same"}},
        },
    }


def test_parity_receipt_is_bound_to_generated_target(tmp_path: Path) -> None:
    receipt_path = tmp_path / "parity.json"
    receipt_path.write_text(json.dumps(parity_payload()))

    receipt = validate_parity_receipt(receipt_path, "teleo_clone_test")

    assert receipt["target_database"] == "teleo_clone_test"
    assert receipt["target_table_count"] == 2

    with pytest.raises(RuntimeError, match="target_database"):
        validate_parity_receipt(receipt_path, "teleo_clone_other")


def test_generated_wrapper_executes_only_clone_bound_default_read_only_tool(tmp_path: Path) -> None:
    fake_bin = tmp_path / "bin"
    fake_bin.mkdir()
    gcloud = fake_bin / "gcloud"
    gcloud.write_text("#!/bin/sh\nprintf 'fake-password\\n'\n")
    gcloud.chmod(0o700)
    psql = fake_bin / "psql"
    psql.write_text(
        "#!/bin/sh\ncat >/dev/null\nprintf '%s\\n' "
        "'{\"current_database\":\"teleo_clone_test\",\"system_identifier\":\"system-1\","
        "\"server_address\":\"10.61.0.3\",\"ssl\":true,\"transaction_read_only\":\"on\","
        "\"default_transaction_read_only\":\"on\"}'\n"
    )
    psql.chmod(0o700)
    tool = tmp_path / "tool.py"
    tool.write_text(
        "import json, os, sys\n"
        "print(json.dumps({'argv': sys.argv[1:], 'pgoptions': os.environ.get('PGOPTIONS'), "
        "'has_password': bool(os.environ.get('PGPASSWORD'))}))\n"
    )
    log = tmp_path / "calls.jsonl"
    wrapper = tmp_path / "teleo-kb"
    wrapper.write_text(
        build_cloudsql_wrapper(
            cloudsql_tool=tool,
            tool_log=log,
            target_db="teleo_clone_test",
            host="10.61.0.3",
            project="teleo-501523",
            password_secret="test-secret",
            run_nonce="test-nonce",
            system_exec_path=f"{fake_bin}:/usr/bin:/bin",
        )
    )
    wrapper.chmod(0o700)

    completed = subprocess.run(
        [str(wrapper), "decision-matrix-status"],
        text=True,
        capture_output=True,
        env={**os.environ, "HOME": str(tmp_path)},
        check=False,
    )

    assert completed.returncode == 0, completed.stderr
    tool_receipt = json.loads(completed.stdout)
    assert tool_receipt["pgoptions"] == "-c default_transaction_read_only=on"
    assert tool_receipt["has_password"] is True
    assert tool_receipt["argv"][:6] == [
        "--host",
        "10.61.0.3",
        "--db",
        "teleo_clone_test",
        "--canonical-db",
        "teleo_clone_test",
    ]
    events = [json.loads(line) for line in log.read_text().splitlines()]
    assert [event["phase"] for event in events] == ["start", "end"]
    assert events[0]["database_identity"]["default_transaction_read_only"] == "on"
    assert "fake-password" not in log.read_text()
    proof = read_tool_proof(
        log,
        "teleo-prod-1",
        "teleo_clone_test",
        run_nonce="test-nonce",
        database_identity={"system_identifier": "system-1"},
        require_database_read_only=True,
        require_default_read_only=True,
    )
    assert proof["all_bound_to_supplied_target"] is True
    events[0]["database_identity"]["default_transaction_read_only"] = "off"
    unsafe_log = tmp_path / "unsafe-calls.jsonl"
    unsafe_log.write_text("".join(json.dumps(event) + "\n" for event in events))
    unsafe_proof = read_tool_proof(
        unsafe_log,
        "teleo-prod-1",
        "teleo_clone_test",
        run_nonce="test-nonce",
        database_identity={"system_identifier": "system-1"},
        require_database_read_only=True,
        require_default_read_only=True,
    )
    assert unsafe_proof["all_bound_to_supplied_target"] is False


def test_run_suite_retains_failure_report_when_postflight_fails(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    output = tmp_path / "report.json"
    placeholder = tmp_path / "placeholder"
    placeholder.write_text("x")
    args = argparse.Namespace(
        target_db="teleo_clone_test",
        cloudsql_tool=placeholder,
        manifest_sql=placeholder,
        parity_receipt=placeholder,
        service="leo.service",
        live_profile=tmp_path,
        output=output,
        chat_id="1",
        user_id="2",
        user_name="test",
        run_id="test-run",
        turn_timeout=1,
        host="10.61.0.3",
        project="teleo-501523",
        password_secret="secret-name",
    )
    identity_calls = 0

    def fake_identity(_args: argparse.Namespace) -> dict[str, object]:
        nonlocal identity_calls
        identity_calls += 1
        if identity_calls > 1:
            raise RuntimeError("postflight unavailable")
        return {
            "current_database": "teleo_clone_test",
            "system_identifier": "system-1",
            "server_address": "10.61.0.3",
            "ssl": True,
            "transaction_read_only": "on",
            "default_transaction_read_only": "on",
        }

    monkeypatch.setattr("scripts.run_gcp_generated_db_direct_claim_suite.validate_reviewed_file", lambda *args: "hash")
    monkeypatch.setattr(
        "scripts.run_gcp_generated_db_direct_claim_suite.validate_parity_receipt",
        lambda *args: {"target_database": "teleo_clone_test"},
    )
    monkeypatch.setattr("scripts.run_gcp_generated_db_direct_claim_suite.service_state", lambda *args: {"MainPID": "1"})
    monkeypatch.setattr(
        "scripts.run_gcp_generated_db_direct_claim_suite.profile_hashes",
        lambda *args: {"wrapper": "hash"},
    )
    monkeypatch.setattr("scripts.run_gcp_generated_db_direct_claim_suite.database_identity", fake_identity)
    monkeypatch.setattr(
        "scripts.run_gcp_generated_db_direct_claim_suite.database_fingerprint",
        lambda *args: {"sha256": "fingerprint"},
    )
    monkeypatch.setattr(
        "scripts.run_gcp_generated_db_direct_claim_suite.canonical_status",
        lambda *args: {"high_signal_rows": {"claims": 1}},
    )
    monkeypatch.setattr(
        "scripts.run_gcp_generated_db_direct_claim_suite.bound.copy_profile",
        lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError("runner setup failed")),
    )

    report = asyncio.run(run_suite(args))
    retained = json.loads(output.read_text())

    assert report["status"] == "fail"
    assert retained["status"] == "fail"
    assert {error["phase"] for error in retained["errors"]} >= {"execution", "database_identity_after"}
    assert "gateway_child_cleanup" in retained