fix(km): retry transient list readbacks
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 1s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / build-and-deploy (push) Has been cancelled
CD Pipeline / post-deploy-checks (push) Has been cancelled
CD Pipeline / tests (push) Has been cancelled

This commit is contained in:
ogt
2026-07-09 18:22:12 +08:00
parent 709724bd89
commit c94e08e995
2 changed files with 101 additions and 32 deletions

View File

@@ -1,5 +1,6 @@
import pytest
from src.models.knowledge import KnowledgeListResponse
from src.services import knowledge_service as knowledge_service_module
from src.services.knowledge_service import KnowledgeService
@@ -85,6 +86,28 @@ async def test_knowledge_list_entries_source_backed_filter_and_search(monkeypatc
assert response.asset_taxonomy[[row.key for row in response.asset_taxonomy].index("alert")].count == 1
@pytest.mark.asyncio
async def test_knowledge_list_entries_retries_transient_pool_timeout(monkeypatch) -> None:
service = KnowledgeService.__new__(KnowledgeService)
calls = 0
async def fake_primary(**_kwargs):
nonlocal calls
calls += 1
if calls == 1:
raise RuntimeError("db pool exhausted")
return KnowledgeListResponse(items=[], total=1, readback_status="ready")
monkeypatch.setattr(service, "_list_entries_from_primary", fake_primary)
response = await service.list_entries(limit=50)
assert calls == 2
assert response.readback_status == "ready"
assert response.primary_readback_ready is True
assert response.operator_stage == "knowledge_readback_primary_retry_recovered"
@pytest.mark.asyncio
async def test_knowledge_search_fails_soft_to_source_backed_entries(monkeypatch) -> None:
monkeypatch.setattr(