fix(sre): verify DR scorecard RAG writeback

This commit is contained in:
Your Name
2026-07-19 01:12:32 +08:00
parent 0f30d1350f
commit 207f00522a
2 changed files with 85 additions and 3 deletions

View File

@@ -189,6 +189,66 @@ async def test_backup_mcp_ack_requires_dr_scorecard_binding() -> None:
assert receipt is None
@pytest.mark.asyncio
async def test_dr_scorecard_ack_requires_complete_safe_refs_and_rag_embedding(
monkeypatch,
) -> None:
identity = build_agent99_dispatch_identity(
project_id="awoooi",
incident_id="INC-20260719-DR-RAG",
source_fingerprint="backup-dr-rag-fingerprint",
route_id="agent99:backup_health:BackupCheck",
)
created = []
class Knowledge:
async def create_entry(self, request, **kwargs): # type: ignore[no-untyped-def]
created.append((request, kwargs))
return SimpleNamespace(
id="00000000-0000-0000-0000-0000000000dr",
title=request.title,
content=request.content,
)
async def ensure_entry_embedding(
self, *_args, **_kwargs
): # type: ignore[no-untyped-def]
return True
monkeypatch.setattr(job, "get_knowledge_service", lambda: Knowledge())
complete_refs = {
"agent99_outcome_receipt_id": "agent99:outcome:1",
"post_verifier_evidence_ref": "verifier:1",
"source_event_evidence_ref": "source:1",
"backup_status_evidence_ref": "backup:1",
"freshness_evidence_ref": "freshness:1",
"offsite_verify_evidence_ref": "offsite:1",
"escrow_evidence_ref": "escrow:1",
"restore_drill_evidence_ref": "restore:1",
"source_resolution_receipt_ref": "resolution:1",
}
missing = await job._ensure_dr_scorecard(
identity,
evidence_refs={"backup_status_evidence_ref": "backup:1"},
)
receipt = await job._ensure_dr_scorecard(
identity,
evidence_refs={**complete_refs, "unsafe": "secret=value"},
)
assert missing is None
assert receipt == (
"knowledge_entries:00000000-0000-0000-0000-0000000000dr:"
"dr_scorecard_embedding_verified"
)
assert len(created) == 1
request, kwargs = created[0]
assert kwargs["schedule_embedding"] is False
assert "unsafe" not in request.content
assert all(value in request.content for value in complete_refs.values())
@pytest.mark.asyncio
async def test_backup_telegram_ack_is_no_write_lifecycle_not_apply(
monkeypatch,