fix(governance): fail soft stale km readbacks
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m11s
CD Pipeline / post-deploy-checks (push) Has been cancelled
CD Pipeline / build-and-deploy (push) Has been cancelled

This commit is contained in:
Your Name
2026-07-03 09:19:30 +08:00
parent 7085a94eb6
commit 43620a977f
2 changed files with 110 additions and 1 deletions

View File

@@ -742,6 +742,28 @@ class TestKmReviewDraftDedupe:
"signoz",
]
def test_km_stale_candidates_endpoint_readback_exception_returns_empty_queue(self, client):
"""stale candidates 背景讀取失敗時應 fail-soft不讓 Work Items 收到 500."""
with patch(
"src.api.v1.ai_governance.query_km_stale_candidates",
new=AsyncMock(side_effect=RuntimeError("temporary stale candidates failure")),
):
r = client.get(
"/api/v1/ai/governance/km-stale-candidates"
"?project_id=awoooi&limit=20"
)
assert r.status_code == 200
data = r.json()
assert data["schema_version"] == "km_stale_candidates_v1"
assert data["project_id"] == "awoooi"
assert data["total_stale"] == 0
assert data["returned"] == 0
assert data["writes_on_read"] is False
assert data["manual_review_required"] is True
assert data["items"] == []
assert data["generated_at"]
def test_build_km_stale_candidate_prioritizes_linked_evidence(self):
"""有 Incident / PlayBook / Sentry / SigNoz 脈絡的 stale KM 應排前面。"""
record = KnowledgeEntryRecord(
@@ -1302,6 +1324,43 @@ class TestKmReviewDraftDedupe:
assert data["items"][0]["can_preview"] is True
assert data["items"][1]["writes_km_on_confirm"] is False
def test_owner_review_completion_queue_readback_exception_returns_empty_queue(
self,
client,
):
"""completion queue 背景讀取失敗時應 fail-soft不讓 Work Items 收到 500."""
with patch(
"src.api.v1.ai_governance.query_km_stale_owner_review_completion_queue",
new=AsyncMock(side_effect=RuntimeError("temporary completion queue failure")),
):
r = client.get(
"/api/v1/ai/governance/km-stale-owner-review-completion-queue"
"?project_id=awoooi&status_bucket=all&priority_tier=P0"
"&recommended_completion_outcome=refresh_with_evidence"
"&can_preview=true&limit=30"
)
assert r.status_code == 200
data = r.json()
assert data["schema_version"] == "km_stale_owner_review_completion_queue_v1"
assert data["project_id"] == "awoooi"
assert data["status_bucket"] == "all"
assert data["priority_tiers"] == ["P0"]
assert data["recommended_completion_outcome"] == "refresh_with_evidence"
assert data["can_preview"] is True
assert data["total"] == 0
assert data["returned"] == 0
assert data["pending_count"] == 0
assert data["ready_count"] == 0
assert data["blocked_count"] == 0
assert data["completed_count"] == 0
assert data["failed_count"] == 0
assert data["writes_on_read"] is False
assert data["manual_review_required"] is True
assert data["batch_writes_allowed"] is False
assert data["items"] == []
assert data["generated_at"]
def test_completion_queue_batch_preview_endpoint_is_dry_run_only(self, client):
fake = KnowledgeStaleOwnerReviewCompletionBatchPreviewResponse(
project_id="awoooi",