awoooi/apps/api/tests/test_knowledge_service_readback_degraded.py

import pytest

from src.models.knowledge import (
    EntrySource,
    EntryStatus,
    EntryType,
    KnowledgeEntry,
    KnowledgeListResponse,
)
from src.services import knowledge_service as knowledge_service_module
from src.services.knowledge_service import KnowledgeService


class _BrokenDbContext:
    async def __aenter__(self):
        raise RuntimeError("db pool exhausted")

    async def __aexit__(self, exc_type, exc, tb):
        return False


class _OkDbContext:
    async def __aenter__(self):
        return object()

    async def __aexit__(self, exc_type, exc, tb):
        return False


@pytest.mark.asyncio
async def test_knowledge_list_entries_fails_soft_when_readback_breaks(monkeypatch) -> None:
    monkeypatch.setattr(
        knowledge_service_module,
        "get_db_context",
        lambda: _BrokenDbContext(),
    )
    service = KnowledgeService.__new__(KnowledgeService)

    response = await service.list_entries(limit=50)

    assert response.total == 13
    assert len(response.items) == 13
    assert response.items[0].id == "source-backed-project-awoooi"
    assert response.items[0].source == "ai_extracted"
    assert response.items[0].status == "approved"
    assert [row.category for row in response.categories] == [
        "project",
        "product",
        "website",
        "service",
        "package",
        "tool",
        "log",
        "alert",
        "playbook",
        "rag",
        "mcp",
        "schedule",
        "general",
    ]
    assert all(row.count == 1 for row in response.categories)
    assert [row.key for row in response.asset_taxonomy] == [
        "project",
        "product",
        "website",
        "service",
        "package",
        "tool",
        "log",
        "alert",
        "playbook",
        "rag",
        "mcp",
        "schedule",
    ]
    assert all(row.count >= 1 for row in response.asset_taxonomy)
    assert response.readback_status == "source_backed_degraded"
    assert response.primary_readback_ready is False
    assert response.degraded_reason_code == "primary_km_db_timeout_or_pool_exhausted"
    assert response.operator_stage == "knowledge_readback_source_backed_ai_controlled_repair"
    assert response.next_step == "repair_primary_km_db_readback_then_promote_source_backed_receipts_to_persistent_km"
    assert response.writes_on_read is False
    assert response.manual_review_required is False


@pytest.mark.asyncio
async def test_knowledge_list_entries_source_backed_filter_and_search(monkeypatch) -> None:
    monkeypatch.setattr(
        knowledge_service_module,
        "get_db_context",
        lambda: _BrokenDbContext(),
    )
    service = KnowledgeService.__new__(KnowledgeService)

    response = await service.list_entries(category="alert", q="Telegram", limit=50)

    assert response.total == 1
    assert response.items[0].id == "source-backed-alert-telegram-monitoring-coverage"
    assert response.categories[[row.category for row in response.categories].index("alert")].count == 1
    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_list_entries_keeps_primary_items_when_taxonomy_degrades(monkeypatch) -> None:
    primary_entry = KnowledgeEntry(
        id="km-primary-1",
        title="Primary KM row",
        content="Persistent AI automation memory row",
        entry_type=EntryType.RUNBOOK,
        category="alert_handling",
        tags=["telegram", "ai_agent"],
        source=EntrySource.AI_EXTRACTED,
        status=EntryStatus.APPROVED,
    )

    class _TaxonomyBrokenRepo:
        def __init__(self, _db):
            pass

        async def list_entries(self, **_kwargs):
            return [primary_entry], 638

        async def get_categories(self):
            raise RuntimeError("db pool exhausted")

        async def get_asset_taxonomy_counts(self):
            raise RuntimeError("db pool exhausted")

    monkeypatch.setattr(
        knowledge_service_module,
        "get_db_context",
        lambda: _OkDbContext(),
    )
    monkeypatch.setattr(
        knowledge_service_module,
        "KnowledgeDBRepository",
        _TaxonomyBrokenRepo,
    )
    service = KnowledgeService.__new__(KnowledgeService)

    response = await service.list_entries(limit=50)

    assert response.total == 638
    assert response.items == [primary_entry]
    assert response.readback_status == "ready_partial_taxonomy_degraded"
    assert response.primary_readback_ready is True
    assert response.degraded_reason_code == "primary_km_taxonomy_readback_degraded"
    assert response.operator_stage == "knowledge_primary_entries_ready_taxonomy_degraded"
    assert response.categories[[row.category for row in response.categories].index("alert_handling")].count == 1
    assert response.asset_taxonomy[[row.key for row in response.asset_taxonomy].index("alert")].count == 1


@pytest.mark.asyncio
async def test_knowledge_search_fails_soft_to_source_backed_entries(monkeypatch) -> None:
    monkeypatch.setattr(
        knowledge_service_module,
        "get_db_context",
        lambda: _BrokenDbContext(),
    )
    service = KnowledgeService.__new__(KnowledgeService)

    entries = await service.search("Telegram", limit=10)

    entry_ids = {entry.id for entry in entries}
    assert all(entry.id.startswith("source-backed-") for entry in entries)
    assert {
        "source-backed-service-telegram-alert-receipts",
        "source-backed-alert-telegram-monitoring-coverage",
        "source-backed-schedule-report-monitoring",
    }.issubset(entry_ids)


@pytest.mark.asyncio
async def test_knowledge_categories_fails_soft_when_readback_breaks(monkeypatch) -> None:
    monkeypatch.setattr(
        knowledge_service_module,
        "get_db_context",
        lambda: _BrokenDbContext(),
    )
    service = KnowledgeService.__new__(KnowledgeService)

    categories = await service.get_categories()

    assert [row.category for row in categories] == [
        "project",
        "product",
        "website",
        "service",
        "package",
        "tool",
        "log",
        "alert",
        "playbook",
        "rag",
        "mcp",
        "schedule",
        "general",
    ]
    assert all(row.count == 1 for row in categories)


@pytest.mark.asyncio
async def test_knowledge_asset_taxonomy_fails_soft_when_readback_breaks(monkeypatch) -> None:
    monkeypatch.setattr(
        knowledge_service_module,
        "get_db_context",
        lambda: _BrokenDbContext(),
    )
    service = KnowledgeService.__new__(KnowledgeService)

    taxonomy = await service.get_asset_taxonomy()

    assert [row.key for row in taxonomy] == [
        "project",
        "product",
        "website",
        "service",
        "package",
        "tool",
        "log",
        "alert",
        "playbook",
        "rag",
        "mcp",
        "schedule",
    ]
    assert all(row.count >= 1 for row in taxonomy)