645 lines
24 KiB
Python
645 lines
24 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import hashlib
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from scripts import run_gcp_generated_db_working_leo_suite as suite
|
|
|
|
|
|
def parity_payload(target_db: str = "teleo_clone_working_leo") -> dict[str, object]:
|
|
return {
|
|
"artifact": "canonical_postgres_parity_verification",
|
|
"status": "pass",
|
|
"scope": "disposable_cloudsql_copy",
|
|
"problems": [],
|
|
"details": {
|
|
"target_database": target_db,
|
|
"source_table_count": 39,
|
|
"target_table_count": 39,
|
|
"source_total_rows": 100,
|
|
"target_total_rows": 100,
|
|
"table_mismatches": [],
|
|
"application_role_mismatches": {},
|
|
"required_extension_mismatches": {},
|
|
"performance_problems": [],
|
|
"structural_hashes": {
|
|
"columns": {"source": "same", "target": "same"},
|
|
"constraints": {"source": "same", "target": "same"},
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
def write_parity(tmp_path: Path, target_db: str = "teleo_clone_working_leo") -> Path:
|
|
path = tmp_path / "parity.json"
|
|
path.write_text(json.dumps(parity_payload(target_db)), encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def write_pinned_json(tmp_path: Path, name: str, payload: dict[str, object]) -> tuple[Path, str]:
|
|
path = tmp_path / name
|
|
content = json.dumps(payload, sort_keys=True).encode()
|
|
path.write_bytes(content)
|
|
return path, hashlib.sha256(content).hexdigest()
|
|
|
|
|
|
def target_identity_payload(**overrides: object) -> dict[str, object]:
|
|
payload: dict[str, object] = {
|
|
"artifact": suite.TARGET_IDENTITY_ARTIFACT,
|
|
"status": "pass",
|
|
"problems": [],
|
|
"target_database": "teleo_clone_working_leo",
|
|
"project": "teleo-501523",
|
|
"instance": "teleo-pgvector-standby",
|
|
"server_address": "10.61.0.3",
|
|
"system_identifier": "target-system-123",
|
|
"captured_at_utc": "2026-07-12T00:00:00Z",
|
|
}
|
|
payload.update(overrides)
|
|
return payload
|
|
|
|
|
|
def source_baseline_payload() -> dict[str, object]:
|
|
tables = suite.composition.PRODUCTION_TABLES
|
|
return {
|
|
"artifact": suite.SOURCE_BASELINE_ARTIFACT,
|
|
"status": "pass",
|
|
"problems": [],
|
|
"project": "teleo-501523",
|
|
"instance": "teleo-pgvector-standby",
|
|
"captured_at_utc": "2026-07-12T00:00:00Z",
|
|
"guard_snapshot": {
|
|
"database_identity": {
|
|
"current_database": "teleo_source",
|
|
"system_identifier": "source-system-456",
|
|
},
|
|
"counts": {table: 1 for table in tables},
|
|
"rowset_md5": {table: f"md5-{index}" for index, table in enumerate(tables)},
|
|
},
|
|
"gate_schema": {
|
|
"approval_table": "kb_stage.kb_proposal_approvals",
|
|
"review_principals_table": "kb_stage.kb_review_principals",
|
|
"review_role_exists": True,
|
|
"apply_role_exists": True,
|
|
"approve_function": "approve",
|
|
"assert_function": "assert",
|
|
"finish_function": "finish",
|
|
},
|
|
}
|
|
|
|
|
|
def capability_receipt(
|
|
role: str, target_db: str = "teleo_clone_working_leo", *, writable: bool | None = None
|
|
) -> str:
|
|
capabilities = {
|
|
"kb_read": (False, False, False, False, False),
|
|
"kb_stage_writer": (True, False, False, False, False),
|
|
"kb_review": (False, True, False, False, False),
|
|
"kb_apply": (False, False, True, True, True),
|
|
}
|
|
insert_proposals, approve, assert_apply, finish_apply, insert_canonical = capabilities[role]
|
|
writable = role != "kb_read" if writable is None else writable
|
|
payload = {
|
|
"session_user": role,
|
|
"current_user": role,
|
|
"current_database": target_db,
|
|
"transaction_read_only": "off" if writable else "on",
|
|
"default_transaction_read_only": "off" if writable else "on",
|
|
"is_superuser": False,
|
|
"can_select_proposals": True,
|
|
"can_insert_proposals": insert_proposals,
|
|
"can_update_proposals": False,
|
|
"can_approve": approve,
|
|
"can_assert_apply": assert_apply,
|
|
"can_finish_apply": finish_apply,
|
|
"can_insert_canonical": insert_canonical,
|
|
}
|
|
return suite.ROLE_RECEIPT_PREFIX + json.dumps(payload, sort_keys=True)
|
|
|
|
|
|
def executor_args(**overrides: object) -> argparse.Namespace:
|
|
values: dict[str, object] = {
|
|
"target_db": "teleo_clone_working_leo",
|
|
"host": "10.61.0.3",
|
|
"project": "teleo-501523",
|
|
"instance": "teleo-pgvector-standby",
|
|
"password_secret": "read-secret",
|
|
"read_role": "kb_read",
|
|
"stage_role": "kb_stage_writer",
|
|
"review_role": "kb_review",
|
|
"apply_role": "kb_apply",
|
|
"stage_password_secret": "stage-secret",
|
|
"review_password_secret": "review-secret",
|
|
"apply_password_secret": "apply-secret",
|
|
"allow_stage": True,
|
|
"allow_review": True,
|
|
"allow_apply": True,
|
|
}
|
|
values.update(overrides)
|
|
return argparse.Namespace(**values)
|
|
|
|
|
|
def test_reuses_exact_twenty_prompt_catalog_and_existing_scorer_contracts() -> None:
|
|
catalog = suite.full_catalog()
|
|
checks = suite.composition_check_contract()
|
|
|
|
assert len(catalog) == 20
|
|
assert [row["id"] for row in catalog] == [
|
|
"OE-01",
|
|
"OE-02",
|
|
"OE-03",
|
|
"OE-04",
|
|
"OE-05",
|
|
"CS-01",
|
|
"CS-02",
|
|
"CS-03",
|
|
"CS-04",
|
|
"CS-05",
|
|
"CS-06",
|
|
"CS-07",
|
|
"CS-08",
|
|
"CS-09",
|
|
"DC-01",
|
|
"DC-02",
|
|
"DC-03",
|
|
"DC-04",
|
|
"DC-05",
|
|
"DC-06",
|
|
]
|
|
assert len(checks) == 34
|
|
assert "source_packet_validated" in checks
|
|
assert "isolated_restart_and_memory_survived" in checks
|
|
|
|
|
|
def test_validation_binds_parity_and_reviewed_hashes_to_generated_database(tmp_path: Path) -> None:
|
|
args = argparse.Namespace(
|
|
target_db="teleo_clone_working_leo",
|
|
cloudsql_tool=Path("hermes-agent/leoclean-bin/cloudsql_memory_tool.py"),
|
|
manifest_sql=Path("ops/postgres_parity_manifest.sql"),
|
|
parity_receipt=write_parity(tmp_path),
|
|
)
|
|
|
|
result = suite.validate_local_inputs(args)
|
|
|
|
assert result["parity_receipt"]["target_database"] == args.target_db
|
|
assert result["catalog_contract"]["count"] == 20
|
|
assert result["composition_contract"]["inherited_check_count"] == 34
|
|
assert result["composition_contract"]["inherited_34_of_34_claimable"] is False
|
|
assert result["composition_contract"]["gcp_required_check_count"] == 32
|
|
assert result["execute_input_contract"]["ready"] is False
|
|
assert "--target-identity-receipt" in result["execute_input_contract"]["missing"]
|
|
assert result["reviewed_inputs"]["cloudsql_tool_sha256"] == suite.gcp.REVIEWED_CLOUDSQL_TOOL_SHA256
|
|
assert result["reviewed_inputs"]["manifest_sql_sha256"] == suite.gcp.REVIEWED_MANIFEST_SQL_SHA256
|
|
|
|
args.target_db = "teleo_clone_other"
|
|
with pytest.raises(RuntimeError, match="target_database"):
|
|
suite.validate_local_inputs(args)
|
|
|
|
|
|
def test_executor_binds_every_call_and_defaults_read_only_except_controller_phases(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
calls: list[dict[str, object]] = []
|
|
|
|
def fake_run(command: list[str], **kwargs: object) -> str:
|
|
if command[0] == "gcloud":
|
|
return "private-value\n"
|
|
if "env" in kwargs:
|
|
kwargs["env"] = dict(kwargs["env"])
|
|
calls.append({"command": command, **kwargs})
|
|
connection = command[1]
|
|
role = connection.split("user=", 1)[1].split(" ", 1)[0]
|
|
writable = kwargs["env"]["PGOPTIONS"].endswith("off")
|
|
return capability_receipt(role, writable=writable) + "\nok\n"
|
|
|
|
monkeypatch.setattr(suite.gcp, "run", fake_run)
|
|
executor = suite.CloudSqlExecutor(executor_args())
|
|
|
|
executor.read("select 1;")
|
|
executor.stage("insert into kb_stage.kb_proposals default values;")
|
|
executor.review("select kb_stage.approve_strict_proposal();")
|
|
executor.apply("select kb_stage.finish_approved_proposal();")
|
|
|
|
assert len(calls) == 4
|
|
assert all("dbname=teleo_clone_working_leo" in call["command"][1] for call in calls)
|
|
assert all("expected_database=teleo_clone_working_leo" in call["command"] for call in calls)
|
|
assert calls[0]["env"]["PGOPTIONS"] == "-c default_transaction_read_only=on"
|
|
assert [call["env"]["PGOPTIONS"] for call in calls[1:]] == [
|
|
"-c default_transaction_read_only=off",
|
|
"-c default_transaction_read_only=off",
|
|
"-c default_transaction_read_only=off",
|
|
]
|
|
assert all("refusing database fallback" in str(call["input_text"]) for call in calls)
|
|
assert [receipt["purpose"] for receipt in executor.receipts] == ["read", "stage", "review", "apply"]
|
|
assert [receipt["capability_receipt"]["session_user"] for receipt in executor.receipts] == [
|
|
"kb_read",
|
|
"kb_stage_writer",
|
|
"kb_review",
|
|
"kb_apply",
|
|
]
|
|
|
|
|
|
def test_executor_refuses_unflagged_or_mislabeled_write_authority(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(suite.gcp, "run", lambda *_args, **_kwargs: "secret\n")
|
|
executor = suite.CloudSqlExecutor(executor_args(allow_apply=False))
|
|
|
|
with pytest.raises(RuntimeError, match="not explicitly authorized"):
|
|
executor.apply("select 1;")
|
|
with pytest.raises(RuntimeError, match="write mode is allowed only"):
|
|
executor.execute(
|
|
"select 1;",
|
|
role="postgres",
|
|
password_secret="read-secret",
|
|
purpose="read",
|
|
writable=True,
|
|
)
|
|
with pytest.raises(RuntimeError, match="safe identifier"):
|
|
executor.execute(
|
|
"select 1;",
|
|
role="postgres dbname=teleo",
|
|
password_secret="read-secret",
|
|
)
|
|
postgres_writer = suite.CloudSqlExecutor(executor_args(stage_role="postgres"))
|
|
with pytest.raises(RuntimeError, match="postgres write authority"):
|
|
postgres_writer.stage("select 1;")
|
|
|
|
|
|
def test_sha_pinned_target_and_source_receipts_bind_physical_context(tmp_path: Path) -> None:
|
|
target_path, target_sha = write_pinned_json(tmp_path, "target.json", target_identity_payload())
|
|
source_path, source_sha = write_pinned_json(tmp_path, "source.json", source_baseline_payload())
|
|
args = argparse.Namespace(
|
|
validation_only=True,
|
|
execute=False,
|
|
target_db="teleo_clone_working_leo",
|
|
host="10.61.0.3",
|
|
project="teleo-501523",
|
|
instance="teleo-pgvector-standby",
|
|
cloudsql_tool=Path("hermes-agent/leoclean-bin/cloudsql_memory_tool.py"),
|
|
manifest_sql=Path("ops/postgres_parity_manifest.sql"),
|
|
parity_receipt=write_parity(tmp_path),
|
|
target_identity_receipt=target_path,
|
|
target_identity_sha256=target_sha,
|
|
source_baseline_receipt=source_path,
|
|
source_baseline_sha256=source_sha,
|
|
)
|
|
|
|
result = suite.validate_local_inputs(args)
|
|
|
|
assert result["execute_input_contract"] == {"ready": True, "missing": []}
|
|
assert result["target_identity_receipt"]["system_identifier"] == "target-system-123"
|
|
assert result["source_baseline_receipt"]["source_system_identifier"] == "source-system-456"
|
|
suite.assert_live_target_identity(
|
|
args,
|
|
result["target_identity_receipt"],
|
|
{
|
|
"current_database": args.target_db,
|
|
"system_identifier": "target-system-123",
|
|
"server_address": args.host,
|
|
"ssl": True,
|
|
"default_transaction_read_only": "on",
|
|
},
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match="system_identifier"):
|
|
suite.assert_live_target_identity(
|
|
args,
|
|
result["target_identity_receipt"],
|
|
{
|
|
"current_database": args.target_db,
|
|
"system_identifier": "different-physical-cluster",
|
|
"server_address": args.host,
|
|
"ssl": True,
|
|
"default_transaction_read_only": "on",
|
|
},
|
|
)
|
|
with pytest.raises(RuntimeError, match="SHA-256 pin"):
|
|
suite.validate_target_identity_receipt(args, target_path, "0" * 64)
|
|
|
|
|
|
def test_source_snapshots_are_receipt_backed_and_inherited_source_checks_are_not_claimed() -> None:
|
|
args = executor_args()
|
|
source_payload = source_baseline_payload()
|
|
source = {
|
|
"guard_snapshot": source_payload["guard_snapshot"],
|
|
"gate_schema": source_payload["gate_schema"],
|
|
}
|
|
adapter = suite.RuntimeAdapter(args, suite.CloudSqlExecutor(args), source, {"system_identifier": "target"})
|
|
|
|
assert adapter._guard_snapshot(
|
|
suite.bound.PRODUCTION_CONTAINER,
|
|
suite.bound.PRODUCTION_DB,
|
|
tables=suite.composition.PRODUCTION_TABLES,
|
|
) == source_payload["guard_snapshot"]
|
|
assert adapter._gate_schema(suite.bound.PRODUCTION_CONTAINER, suite.bound.PRODUCTION_DB) == source_payload[
|
|
"gate_schema"
|
|
]
|
|
contract = suite.composition_check_contract()
|
|
required = [name for name in contract if name not in suite.UNSUPPORTED_INHERITED_COMPOSITION_CHECKS]
|
|
assert len(contract) == 34
|
|
assert len(required) == 32
|
|
assert "production_database_unchanged" not in required
|
|
assert "production_gate_schema_unchanged" not in required
|
|
|
|
|
|
def test_duplicate_credential_values_rejected_without_retaining_values(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
executor = suite.CloudSqlExecutor(executor_args())
|
|
dummy = "DUMMY-DUPLICATE-CREDENTIAL"
|
|
monkeypatch.setattr(executor, "_password", lambda _name: dummy)
|
|
|
|
with pytest.raises(RuntimeError, match="credential values must be distinct") as captured:
|
|
suite.verify_distinct_credential_values(executor)
|
|
|
|
assert dummy not in str(captured.value)
|
|
|
|
|
|
def test_catalog_top_level_gate_requires_status_and_every_safety_check() -> None:
|
|
assert suite.catalog_gate_pass({"status": "pass", "checks": {"score": True, "no_send": True}}) is True
|
|
assert suite.catalog_gate_pass({"status": "fail", "checks": {"score": True, "no_send": True}}) is False
|
|
assert suite.catalog_gate_pass({"status": "pass", "checks": {"score": True, "no_send": False}}) is False
|
|
assert suite.catalog_gate_pass({"status": "pass", "checks": {}}) is False
|
|
|
|
|
|
def test_prompt_scoped_evidence_requires_matching_nonce_target_read_and_role(tmp_path: Path) -> None:
|
|
path = tmp_path / "tool.jsonl"
|
|
identity = {
|
|
"current_database": "teleo_clone_working_leo",
|
|
"system_identifier": "target-system-123",
|
|
"transaction_read_only": "on",
|
|
"default_transaction_read_only": "on",
|
|
"session_user": "kb_read",
|
|
"current_user": "kb_read",
|
|
}
|
|
events = [
|
|
{
|
|
"phase": "start",
|
|
"prompt_id": "OE-01",
|
|
"run_nonce": "nonce-1",
|
|
"invocation_id": "invocation-1",
|
|
"container": suite.gcp.SOURCE_COMPUTE,
|
|
"database": "teleo_clone_working_leo",
|
|
"database_identity": identity,
|
|
"argv": ["search", "working leo"],
|
|
},
|
|
{
|
|
"phase": "end",
|
|
"prompt_id": "OE-01",
|
|
"run_nonce": "nonce-1",
|
|
"invocation_id": "invocation-1",
|
|
"container": suite.gcp.SOURCE_COMPUTE,
|
|
"database": "teleo_clone_working_leo",
|
|
"returncode": 0,
|
|
},
|
|
]
|
|
path.write_text("".join(json.dumps(event) + "\n" for event in events), encoding="utf-8")
|
|
|
|
proof = suite.prompt_scoped_tool_evidence(
|
|
path,
|
|
prompt_id="OE-01",
|
|
run_nonce="nonce-1",
|
|
target_db="teleo_clone_working_leo",
|
|
db_identity=identity,
|
|
read_role="kb_read",
|
|
)
|
|
|
|
assert proof["pass"] is True
|
|
mismatched = suite.prompt_scoped_tool_evidence(
|
|
path,
|
|
prompt_id="OE-02",
|
|
run_nonce="nonce-1",
|
|
target_db="teleo_clone_working_leo",
|
|
db_identity=identity,
|
|
read_role="kb_read",
|
|
)
|
|
assert mismatched["pass"] is False
|
|
assert mismatched["checks"]["prompt_and_run_nonce_match"] is False
|
|
|
|
|
|
def test_dc_sessions_must_use_unique_fresh_profiles_and_sessions() -> None:
|
|
results = [
|
|
{
|
|
"prompt_id": f"DC-{index:02d}",
|
|
"private_profile_id": f"profile-{index}",
|
|
"persisted_session_id": f"session-{index}",
|
|
"fresh_private_profile": True,
|
|
"prior_prompt_ids": [],
|
|
}
|
|
for index in range(1, 7)
|
|
]
|
|
|
|
assert suite.catalog_session_isolation(results) == {
|
|
"all_profiles_unique": True,
|
|
"dc_isolated": True,
|
|
}
|
|
results[-1]["persisted_session_id"] = results[0]["persisted_session_id"]
|
|
assert suite.catalog_session_isolation(results)["dc_isolated"] is False
|
|
|
|
|
|
def test_dummy_secret_never_leaks_from_executor_error(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
dummy = "DUMMY-SECRET-DO-NOT-LEAK"
|
|
|
|
def fake_run(command: list[str], **_kwargs: object) -> str:
|
|
if command[0] == "gcloud":
|
|
return dummy + "\n"
|
|
raise RuntimeError(f"driver echoed {dummy}")
|
|
|
|
monkeypatch.setattr(suite.gcp, "run", fake_run)
|
|
executor = suite.CloudSqlExecutor(executor_args())
|
|
|
|
with pytest.raises(RuntimeError) as captured:
|
|
executor.read("select 1;")
|
|
|
|
assert dummy not in str(captured.value)
|
|
assert "REDACTED_SECRET" in str(captured.value)
|
|
|
|
|
|
def test_runtime_adapter_refuses_fallback_and_unclassified_writes() -> None:
|
|
args = executor_args()
|
|
executor = suite.CloudSqlExecutor(args)
|
|
adapter = suite.RuntimeAdapter(
|
|
args,
|
|
executor,
|
|
{
|
|
"guard_snapshot": source_baseline_payload()["guard_snapshot"],
|
|
"gate_schema": source_baseline_payload()["gate_schema"],
|
|
},
|
|
{"system_identifier": "system-1"},
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match="rebind"):
|
|
adapter._psql_json("wrong-target", args.target_db, "select '{}'::jsonb;")
|
|
with pytest.raises(RuntimeError, match="unclassified SQL write"):
|
|
adapter._psql_json(adapter.target["id"], args.target_db, "delete from public.claims;")
|
|
|
|
|
|
def test_runtime_adapter_restores_helpers_when_enter_fails(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
args = executor_args()
|
|
executor = suite.CloudSqlExecutor(args)
|
|
original = suite.bound._psql_json
|
|
monkeypatch.setattr(executor, "json", lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("gate failed")))
|
|
adapter = suite.RuntimeAdapter(
|
|
args,
|
|
executor,
|
|
{
|
|
"guard_snapshot": source_baseline_payload()["guard_snapshot"],
|
|
"gate_schema": source_baseline_payload()["gate_schema"],
|
|
},
|
|
{"system_identifier": "system-1"},
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match="gate failed"):
|
|
adapter.__enter__()
|
|
|
|
assert suite.bound._psql_json is original
|
|
assert adapter.originals == []
|
|
|
|
|
|
def test_lifecycle_wrapper_exposes_only_read_or_exact_stage_and_never_delivery(tmp_path: Path) -> None:
|
|
identity_path, identity_sha = write_pinned_json(tmp_path, "identity.json", target_identity_payload())
|
|
args = executor_args(
|
|
cloudsql_tool=Path("hermes-agent/leoclean-bin/cloudsql_memory_tool.py").resolve(),
|
|
target_identity_receipt=identity_path,
|
|
target_identity_sha256=identity_sha,
|
|
)
|
|
|
|
wrapper = suite._cloudsql_lifecycle_wrapper(
|
|
args,
|
|
tool_log=tmp_path / "calls.jsonl",
|
|
run_marker="gcp-lifecycle-marker-1234",
|
|
run_nonce="nonce",
|
|
)
|
|
|
|
assert suite.lifecycle.STAGE_COMMAND in wrapper
|
|
assert "accepts no model-supplied arguments" in wrapper
|
|
assert "--stage-password-secret" in wrapper
|
|
assert "--read-role" in wrapper
|
|
assert 'user=$READ_ROLE' in wrapper
|
|
assert "default_transaction_read_only=on" in wrapper
|
|
assert "systemctl restart" not in wrapper
|
|
assert "send_message" not in wrapper
|
|
assert "approve-proposal" not in wrapper
|
|
assert "apply-proposal" not in wrapper
|
|
|
|
catalog_wrapper = suite._build_gcp_read_wrapper(
|
|
args,
|
|
cloudsql_tool=args.cloudsql_tool,
|
|
tool_log=tmp_path / "catalog.jsonl",
|
|
run_nonce="catalog-nonce",
|
|
prompt_id="DC-01",
|
|
)
|
|
assert catalog_wrapper.count('"prompt_id": "DC-01"') == 2
|
|
assert '--user "$READ_ROLE"' in catalog_wrapper
|
|
|
|
|
|
def test_capability_receipt_rejects_superuser_or_wrong_phase_acl() -> None:
|
|
args = executor_args()
|
|
receipt = json.loads(capability_receipt("kb_stage_writer", writable=True)[len(suite.ROLE_RECEIPT_PREFIX) :])
|
|
suite.validate_capability_receipt(
|
|
receipt,
|
|
args=args,
|
|
purpose="stage",
|
|
role="kb_stage_writer",
|
|
writable=True,
|
|
)
|
|
|
|
receipt["is_superuser"] = True
|
|
with pytest.raises(RuntimeError, match="not_superuser"):
|
|
suite.validate_capability_receipt(
|
|
receipt,
|
|
args=args,
|
|
purpose="stage",
|
|
role="kb_stage_writer",
|
|
writable=True,
|
|
)
|
|
receipt["is_superuser"] = False
|
|
receipt["can_approve"] = True
|
|
with pytest.raises(RuntimeError, match="can_approve"):
|
|
suite.validate_capability_receipt(
|
|
receipt,
|
|
args=args,
|
|
purpose="stage",
|
|
role="kb_stage_writer",
|
|
writable=True,
|
|
)
|
|
|
|
|
|
def test_validation_only_never_connects_or_calls_model_and_retains_report(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
output = tmp_path / "report.json"
|
|
args = argparse.Namespace(
|
|
validation_only=True,
|
|
target_db="teleo_clone_working_leo",
|
|
cloudsql_tool=Path("hermes-agent/leoclean-bin/cloudsql_memory_tool.py"),
|
|
manifest_sql=Path("ops/postgres_parity_manifest.sql"),
|
|
parity_receipt=write_parity(tmp_path),
|
|
output=output,
|
|
)
|
|
monkeypatch.setattr(suite.gcp, "database_identity", lambda *_args: pytest.fail("must not connect"))
|
|
monkeypatch.setattr(suite.bound, "invoke_gateway_subprocess", lambda *_args: pytest.fail("must not call model"))
|
|
|
|
report = asyncio.run(suite.run_orchestrator(args))
|
|
|
|
assert report["status"] == "pass"
|
|
assert report["database_connection_attempted"] is False
|
|
assert report["model_call_attempted"] is False
|
|
assert json.loads(output.read_text())["status"] == "pass"
|
|
|
|
|
|
def test_failure_always_writes_structured_report(tmp_path: Path) -> None:
|
|
output = tmp_path / "failure.json"
|
|
bad_tool = tmp_path / "bad-tool.py"
|
|
bad_tool.write_text("print('not reviewed')\n", encoding="utf-8")
|
|
args = argparse.Namespace(
|
|
validation_only=True,
|
|
target_db="teleo_clone_working_leo",
|
|
cloudsql_tool=bad_tool,
|
|
manifest_sql=Path("ops/postgres_parity_manifest.sql"),
|
|
parity_receipt=write_parity(tmp_path),
|
|
output=output,
|
|
)
|
|
|
|
report = asyncio.run(suite.run_orchestrator(args))
|
|
|
|
retained = json.loads(output.read_text())
|
|
assert report["status"] == "fail"
|
|
assert retained["status"] == "fail"
|
|
assert retained["target_database"] == "teleo_clone_working_leo"
|
|
assert retained["posted_to_telegram"] is False
|
|
assert retained["systemd_restart_attempted"] is False
|
|
assert retained["errors"][0]["phase"] == "orchestrator"
|
|
|
|
|
|
def test_validation_only_cli_executes_end_to_end(tmp_path: Path) -> None:
|
|
output = tmp_path / "cli-report.json"
|
|
completed = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"scripts/run_gcp_generated_db_working_leo_suite.py",
|
|
"--validation-only",
|
|
"--target-db",
|
|
"teleo_clone_working_leo",
|
|
"--cloudsql-tool",
|
|
"hermes-agent/leoclean-bin/cloudsql_memory_tool.py",
|
|
"--manifest-sql",
|
|
"ops/postgres_parity_manifest.sql",
|
|
"--parity-receipt",
|
|
str(write_parity(tmp_path)),
|
|
"--output",
|
|
str(output),
|
|
],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
|
|
assert completed.returncode == 0, completed.stderr
|
|
assert json.loads(output.read_text(encoding="utf-8"))["status"] == "pass"
|
|
assert '"mode": "validation_only"' in completed.stdout
|