79 lines
2.9 KiB
Python
79 lines
2.9 KiB
Python
"""Focused tests for the strict approved-change clone receipt."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "scripts"))
|
|
|
|
import run_approve_claim_clone_canary as canary # noqa: E402
|
|
|
|
|
|
def _passing_result(expected_rows: dict, expected_deltas: dict) -> dict:
|
|
proposal_id = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"
|
|
return {
|
|
"status": "pass",
|
|
"fixture_ids": {"proposal_id": proposal_id},
|
|
"apply_transition": {
|
|
"applied_readback": {
|
|
"id": proposal_id,
|
|
"status": "applied",
|
|
"applied_at": "2026-07-12 12:00:00+00",
|
|
"applied_by_handle": "kb-apply",
|
|
}
|
|
},
|
|
"row_readback": expected_rows,
|
|
"clone_counts_before_apply": {key: 0 for key in expected_deltas},
|
|
"clone_counts_after_apply": expected_deltas,
|
|
"clone_apply_table_deltas": expected_deltas,
|
|
"clone_dropped": True,
|
|
"cleanup_readback": {"leftover_clone_database_count": 0},
|
|
"source_database_mutated": False,
|
|
"checks": {"conflict_transaction_rolled_back_exactly": True},
|
|
}
|
|
|
|
|
|
def test_service_state_records_unavailable_systemctl(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(
|
|
canary,
|
|
"_run",
|
|
lambda *_args, **_kwargs: (_ for _ in ()).throw(FileNotFoundError()),
|
|
)
|
|
|
|
assert canary._service_state("unused.service") == {
|
|
"available": False,
|
|
"returncode": 127,
|
|
"reason": "systemctl_unavailable",
|
|
}
|
|
|
|
|
|
def test_strict_receipt_requires_exact_ids_counts_applied_at_and_cleanup() -> None:
|
|
expected_rows = canary._expected_bundle_rows(canary.build_fixture("unit")["apply_payload"])
|
|
expected_deltas = canary._expected_table_deltas(expected_rows)
|
|
result = _passing_result(expected_rows, expected_deltas)
|
|
|
|
receipt = canary._strict_receipt(result, expected_rows, expected_deltas)
|
|
|
|
assert receipt["pass"] is True
|
|
assert receipt["proposal"]["status"] == "applied"
|
|
assert receipt["proposal"]["applied_at"]
|
|
assert receipt["canonical_row_keys"]["actual"] == receipt["canonical_row_keys"]["expected"]
|
|
assert receipt["row_counts"]["actual_deltas"] == expected_deltas
|
|
assert receipt["rollback_cleanup"]["clone_database_removed"] is True
|
|
|
|
result["apply_transition"]["applied_readback"]["applied_at"] = None
|
|
result["row_readback"] = {**expected_rows, "claims": []}
|
|
result["clone_apply_table_deltas"] = {**expected_deltas, "public.claims": 0}
|
|
result["clone_dropped"] = False
|
|
|
|
failed = canary._strict_receipt(result, expected_rows, expected_deltas)
|
|
|
|
assert failed["pass"] is False
|
|
assert failed["checks"]["proposal_applied_at_present"] is False
|
|
assert failed["checks"]["canonical_row_keys_exact"] is False
|
|
assert failed["checks"]["table_deltas_exact"] is False
|
|
assert failed["checks"]["clone_database_removed"] is False
|