148 lines
5.7 KiB
Python
148 lines
5.7 KiB
Python
"""Tests for scripts/collect_telegram_visible_direct_claim_preflight.py."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(REPO_ROOT / "scripts"))
|
|
|
|
import collect_telegram_visible_direct_claim_preflight as preflight # noqa: E402
|
|
|
|
|
|
def args_for(packet: Path, tmp_path: Path) -> argparse.Namespace:
|
|
return argparse.Namespace(
|
|
packet=packet,
|
|
ssh_target="root@example.invalid",
|
|
ssh_key=tmp_path / "key",
|
|
connect_timeout=1,
|
|
timeout=5,
|
|
remote_json_base64=None,
|
|
out=tmp_path / "out.json",
|
|
markdown_out=tmp_path / "out.md",
|
|
)
|
|
|
|
|
|
def good_remote_payload() -> dict:
|
|
return {
|
|
"deploy_head": {"returncode": 0, "stdout": "abc123", "stderr": ""},
|
|
"last_deploy_sha": {"returncode": 0, "stdout": "abc123", "stderr": ""},
|
|
"packet_present": True,
|
|
"packet_bytes": 1024,
|
|
"service": {
|
|
"returncode": 0,
|
|
"stderr": "",
|
|
"state": {
|
|
"ActiveState": "active",
|
|
"SubState": "running",
|
|
"MainPID": "1908033",
|
|
"NRestarts": "0",
|
|
"ExecMainStartTimestamp": "Fri 2026-07-10 01:04:22 UTC",
|
|
},
|
|
},
|
|
"db_counts": {
|
|
"returncode": 0,
|
|
"stderr": "",
|
|
"counts": {
|
|
"public.claims": 1837,
|
|
"public.sources": 4145,
|
|
"public.claim_evidence": 4670,
|
|
"public.claim_edges": 4916,
|
|
"kb_stage.kb_proposals": 26,
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
def test_preflight_passes_with_ready_packet_and_remote_readback(tmp_path: Path):
|
|
def fake_runner(_command: list[str], _input: str, _timeout: int) -> preflight.CommandResult:
|
|
return preflight.CommandResult(0, json.dumps(good_remote_payload()), "")
|
|
|
|
result = preflight.collect_preflight(args_for(preflight.DEFAULT_PACKET, tmp_path), runner=fake_runner)
|
|
|
|
assert result["status"] == "pass"
|
|
assert result["ready_for_authorized_telegram_visible_dc_run"] is True
|
|
assert result["telegram_visible_messages_sent"] is False
|
|
assert result["production_apply_executed"] is False
|
|
assert result["mutates_db"] is False
|
|
assert all(result["packet_checks"].values())
|
|
assert all(result["remote_checks"].values())
|
|
assert result["remote_source"] == "ssh_subprocess"
|
|
|
|
|
|
def test_preflight_can_use_supplied_remote_payload_without_subprocess_ssh(tmp_path: Path):
|
|
args = args_for(preflight.DEFAULT_PACKET, tmp_path)
|
|
args.remote_json_base64 = base64.b64encode(json.dumps(good_remote_payload()).encode()).decode()
|
|
|
|
def failing_runner(_command: list[str], _input: str, _timeout: int) -> preflight.CommandResult:
|
|
raise AssertionError("runner should not be called when remote_json_base64 is supplied")
|
|
|
|
result = preflight.collect_preflight(args, runner=failing_runner)
|
|
|
|
assert result["status"] == "pass"
|
|
assert result["remote_source"] == "supplied_remote_json_base64"
|
|
assert result["remote"]["deploy_head"]["stdout"] == "abc123"
|
|
|
|
|
|
def test_preflight_fails_on_invalid_supplied_remote_payload(tmp_path: Path):
|
|
args = args_for(preflight.DEFAULT_PACKET, tmp_path)
|
|
args.remote_json_base64 = "not valid base64"
|
|
|
|
result = preflight.collect_preflight(args)
|
|
|
|
assert result["status"] == "fail"
|
|
assert result["ready_for_authorized_telegram_visible_dc_run"] is False
|
|
assert result["remote_returncode"] == 2
|
|
assert "failed to parse --remote-json-base64" in result["remote_stderr"]
|
|
assert "Pre-send Telegram-visible direct-claim benchmark" in result["blocker_readback"]["current_canary"]
|
|
assert "remote readback" in result["blocker_readback"]["next_non_user_action"]
|
|
|
|
|
|
def test_preflight_fails_if_deploy_stamp_drift_detected(tmp_path: Path):
|
|
payload = good_remote_payload()
|
|
payload["last_deploy_sha"]["stdout"] = "different"
|
|
|
|
def fake_runner(_command: list[str], _input: str, _timeout: int) -> preflight.CommandResult:
|
|
return preflight.CommandResult(0, json.dumps(payload), "")
|
|
|
|
result = preflight.collect_preflight(args_for(preflight.DEFAULT_PACKET, tmp_path), runner=fake_runner)
|
|
|
|
assert result["status"] == "fail"
|
|
assert result["ready_for_authorized_telegram_visible_dc_run"] is False
|
|
assert result["remote_checks"]["deploy_head_matches_last_deploy"] is False
|
|
|
|
|
|
def test_markdown_reports_counts_and_claim_ceiling(tmp_path: Path):
|
|
def fake_runner(_command: list[str], _input: str, _timeout: int) -> preflight.CommandResult:
|
|
return preflight.CommandResult(0, json.dumps(good_remote_payload()), "")
|
|
|
|
result = preflight.collect_preflight(args_for(preflight.DEFAULT_PACKET, tmp_path), runner=fake_runner)
|
|
out = tmp_path / "preflight.md"
|
|
|
|
preflight.write_markdown(out, result)
|
|
|
|
text = out.read_text(encoding="utf-8")
|
|
assert "Telegram-Visible Direct-Claim Preflight" in text
|
|
assert "Ready for authorized Telegram-visible DC run: `True`" in text
|
|
assert "Remote source: `ssh_subprocess`" in text
|
|
assert "`public.claims`: `1837`" in text
|
|
assert "does not send Telegram messages" in text
|
|
|
|
|
|
def test_markdown_reports_clear_blocker_cta_on_failure(tmp_path: Path):
|
|
def failing_runner(_command: list[str], _input: str, _timeout: int) -> preflight.CommandResult:
|
|
return preflight.CommandResult(255, "", "ssh denied\n")
|
|
|
|
result = preflight.collect_preflight(args_for(preflight.DEFAULT_PACKET, tmp_path), runner=failing_runner)
|
|
out = tmp_path / "preflight-fail.md"
|
|
|
|
preflight.write_markdown(out, result)
|
|
|
|
text = out.read_text(encoding="utf-8")
|
|
assert "## Blocker Readback" in text
|
|
assert "Exact gate: `ssh denied`" in text
|
|
assert "--remote-json-base64 <payload>" in text
|