teleo-infrastructure/tests/test_probe_gcp_db_parity.py

"""Tests for scripts/probe_gcp_db_parity.py."""

from __future__ import annotations

import json
import sys
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO_ROOT / "scripts"))

import probe_gcp_db_parity as probe  # noqa: E402


def write_gcloud_config(root: Path, *, account: str = "billy@livingip.xyz", project: str = "teleo-501523") -> None:
    (root / "configurations").mkdir(parents=True)
    (root / "active_config").write_text("default\n", encoding="utf-8")
    (root / "configurations" / "config_default").write_text(
        f"[core]\naccount = {account}\nproject = {project}\n",
        encoding="utf-8",
    )


def test_dns_gate_uses_static_config_and_redacts_token_stdout(tmp_path: Path):
    config_dir = tmp_path / "gcloud"
    write_gcloud_config(config_dir)
    accounts = [
        {"account": "billy@livingip.xyz", "status": "ACTIVE"},
        {"account": "billyattnmarket@gmail.com", "status": ""},
    ]

    def fake_runner(command: list[str], _env: dict[str, str], _timeout: int) -> probe.CommandResult:
        if command[1:3] == ["auth", "list"]:
            return probe.CommandResult(0, json.dumps(accounts), "")
        if command[1:3] == ["auth", "print-access-token"]:
            return probe.CommandResult(
                1,
                "secret-token-should-not-leak",
                "Failed to resolve 'oauth2.googleapis.com' ([Errno 8] nodename nor servname provided)",
            )
        return probe.CommandResult(
            1,
            "",
            "HTTPSConnectionPool(host='oauth2.googleapis.com'): Failed to resolve 'oauth2.googleapis.com'",
        )

    result = probe.probe_gcp_parity(
        gcloud_path=Path("/fake/gcloud"),
        source_config=config_dir,
        runner=fake_runner,
        use_temp_config=False,
    )

    assert result["status"] == "blocked_on_network_dns_before_auth_refresh"
    assert result["ready_for_gcp_readonly_parity"] is False
    assert result["mutates_gcp"] is False
    assert result["mutates_db"] is False
    assert result["local_gcloud"]["static_active_config"]["account"] == "billy@livingip.xyz"
    assert result["local_gcloud"]["static_active_config"]["project"] == "teleo-501523"
    assert result["accounts_tested"][0]["token_refresh"]["stdout_redacted"] is True
    assert "secret-token" not in json.dumps(result)
    assert "oauth2.googleapis.com" in result["clear_CTA"]


def test_ready_status_requires_project_vm_and_cloudsql_readbacks(tmp_path: Path):
    config_dir = tmp_path / "gcloud"
    write_gcloud_config(config_dir)

    def fake_runner(command: list[str], _env: dict[str, str], _timeout: int) -> probe.CommandResult:
        suffix = command[1:]
        if suffix[:2] == ["auth", "list"]:
            return probe.CommandResult(0, json.dumps([{"account": "billy@livingip.xyz", "status": "ACTIVE"}]), "")
        if suffix[:2] == ["auth", "print-access-token"]:
            return probe.CommandResult(0, "secret-token-should-not-leak", "")
        if suffix[:2] == ["projects", "describe"]:
            return probe.CommandResult(
                0,
                json.dumps(
                    {
                        "projectId": "teleo-501523",
                        "projectNumber": "123456789",
                        "lifecycleState": "ACTIVE",
                    }
                ),
                "",
            )
        if suffix[:3] == ["compute", "instances", "describe"]:
            return probe.CommandResult(
                0,
                json.dumps(
                    {
                        "name": "teleo-prod-1",
                        "status": "RUNNING",
                        "zone": "https://www.googleapis.com/compute/v1/projects/teleo-501523/zones/europe-west6-a",
                        "machineType": "e2-standard-2",
                        "serviceAccounts": [{"email": "sa-teleo-prod-vm@teleo-501523.iam.gserviceaccount.com"}],
                    }
                ),
                "",
            )
        if suffix[:3] == ["sql", "instances", "list"]:
            return probe.CommandResult(
                0,
                json.dumps(
                    [
                        {
                            "name": "teleo-prod-pgvector",
                            "state": "RUNNABLE",
                            "databaseVersion": "POSTGRES_16",
                            "region": "europe-west6",
                        }
                    ]
                ),
                "",
            )
        return probe.CommandResult(0, "{}", "")

    result = probe.probe_gcp_parity(
        gcloud_path=Path("/fake/gcloud"),
        source_config=config_dir,
        runner=fake_runner,
        use_temp_config=False,
    )

    assert result["status"] == "ready_for_gcp_readonly_parity"
    assert result["ready_for_gcp_readonly_parity"] is True
    assert result["project_readback"]["summary"]["projectId"] == "teleo-501523"
    assert result["runtime_readbacks"]["gce_instance"]["summary"]["status"] == "RUNNING"
    assert result["runtime_readbacks"]["cloudsql_instances"]["summary"][0]["name"] == "teleo-prod-pgvector"
    assert "secret-token" not in json.dumps(result)


def test_markdown_names_gate_and_cta(tmp_path: Path):
    report = {
        "generated_at_utc": "2026-07-10T00:00:00+00:00",
        "status": "blocked_on_network_dns_before_auth_refresh",
        "ready_for_gcp_readonly_parity": False,
        "mutates_gcp": False,
        "mutates_db": False,
        "current_canary": "gcloud account can read teleo-501523",
        "local_gcloud": {
            "static_active_config": {"account": "billy@livingip.xyz", "project": "teleo-501523"},
            "used_temporary_writable_config_copy": True,
            "temporary_config_removed_after_probe": True,
        },
        "accounts_tested": [
            {
                "account": "billy@livingip.xyz",
                "token_refresh": {
                    "ok": False,
                    "returncode": 1,
                    "stdout_redacted": True,
                    "stderr_excerpt": "Failed to resolve oauth2.googleapis.com",
                },
            }
        ],
        "project_readback": {"ok": False, "returncode": 1, "summary": None, "stderr_excerpt": "Failed"},
        "exact_gate": "blocked_on_network_dns_before_auth_refresh",
        "clear_CTA": "Run from a network-enabled shell.",
        "next_non_user_action": ["Rerun the probe."],
        "claim_ceiling": "Read-only only.",
    }
    out = tmp_path / "probe.md"

    probe.write_markdown(out, report)

    text = out.read_text(encoding="utf-8")
    assert "GCP DB Parity Current Probe" in text
    assert "Exact gate: `blocked_on_network_dns_before_auth_refresh`" in text
    assert "Run from a network-enabled shell." in text