feat(governance): bind km archive confirm to dry-run fingerprint
All checks were successful
Code Review / ai-code-review (push) Successful in 14s
Type Sync Check / check-type-sync (push) Successful in 31s
CD Pipeline / tests (push) Successful in 4m8s
CD Pipeline / build-and-deploy (push) Successful in 4m48s
CD Pipeline / post-deploy-checks (push) Successful in 2m13s

This commit is contained in:
Your Name
2026-05-20 09:19:32 +08:00
parent 83ca72e989
commit 584d2a77ff
8 changed files with 230 additions and 2 deletions

View File

@@ -38,8 +38,10 @@ from src.models.governance import (
)
from src.services.governance_km_review_service import (
KmReviewDraftArchiveError,
_build_dry_run_plan_fingerprint,
_build_stale_ratio_recheck_context,
_validate_archive_request_against_plan,
_validate_dry_run_plan_fingerprint,
)
from src.services.governance_query_service import (
_build_km_review_draft_dedupe_groups,
@@ -541,12 +543,14 @@ class TestKmReviewDraftDedupe:
"owner": "operator_console",
"owner_approved": True,
"dry_run": False,
"dry_run_plan_fingerprint": "sha256:" + "a" * 64,
},
)
assert r.status_code == 200
assert captured["governance_event_id"] == "event-001"
assert captured["request"].owner_approved is True
assert captured["request"].dry_run_plan_fingerprint == "sha256:" + "a" * 64
data = r.json()
assert data["status"] == "archived"
assert data["writes_km"] is True
@@ -602,6 +606,39 @@ class TestKmReviewDraftDedupe:
assert exc.value.status_code == 409
assert "latest dedupe plan" in exc.value.detail
def test_archive_plan_fingerprint_is_required_before_write(self):
group = KnowledgeReviewDraftDedupeGroup(
governance_event_id="event-001",
canonical_entry_id="km-canonical",
canonical_title="canonical",
preferred_source="dispatch_context",
duplicate_entry_ids=["km-dup-2", "km-dup-1"],
duplicate_count=2,
total_entries=3,
suggested_action="owner_review_canonical_then_archive_duplicates",
owner_action="review_canonical_and_archive_duplicate_drafts",
writes_on_read=False,
can_archive_without_owner_approval=False,
)
fingerprint = _build_dry_run_plan_fingerprint("event-001", group)
assert fingerprint.startswith("sha256:")
assert fingerprint == _build_dry_run_plan_fingerprint("event-001", group)
_validate_dry_run_plan_fingerprint(fingerprint, fingerprint)
with pytest.raises(KmReviewDraftArchiveError) as missing:
_validate_dry_run_plan_fingerprint(None, fingerprint)
assert missing.value.status_code == 403
assert "dry_run_plan_fingerprint" in missing.value.detail
with pytest.raises(KmReviewDraftArchiveError) as mismatch:
_validate_dry_run_plan_fingerprint("sha256:" + "0" * 64, fingerprint)
assert mismatch.value.status_code == 409
assert "latest dedupe plan" in mismatch.value.detail
def test_stale_ratio_recheck_context_is_operator_visible(self):
snapshot = KnowledgeReviewDraftStaleRatioSnapshot(
stale_count=120,
@@ -615,6 +652,7 @@ class TestKmReviewDraftDedupe:
duplicate_entry_ids=["km-dup-1", "km-dup-2"],
owner="operator_console",
owner_approved=True,
dry_run_plan_fingerprint="sha256:" + "a" * 64,
)
ctx = _build_stale_ratio_recheck_context(
@@ -630,6 +668,7 @@ class TestKmReviewDraftDedupe:
assert ctx["workflow"]["current_stage"] == "stale_ratio_recheck"
assert ctx["workflow"]["stage_by_dispatch_status"]["pending"] == "stale_ratio_recheck"
assert ctx["workflow"]["writes_km"] is False
assert ctx["workflow"]["dry_run_plan_fingerprint"] == "sha256:" + "a" * 64
assert ctx["workflow"]["stale_ratio_snapshot"]["stale_ratio"] == pytest.approx(0.6)
assert ctx["worker_result"]["status"] == "stale_ratio_rechecked"
assert ctx["worker_result"]["above_threshold"] is True