awoooi/apps/api/tests/test_knowledge_repository_read_model.py

from __future__ import annotations

from datetime import UTC, datetime
from types import SimpleNamespace

import pytest
from sqlalchemy.dialects import postgresql

from src.db.models import KnowledgeEntryRecord
from src.models.knowledge import EntrySource, EntryStatus, EntryType
from src.repositories.knowledge_repository import (
    _ASSET_TAXONOMY_KEYS,
    KnowledgeDBRepository,
    _enum_text,
)


def _record(**overrides):
    base = {
        "id": "km-1",
        "title": "KM readback",
        "content": "Production KM entry",
        "entry_type": "POSTMORTEM",
        "category": "postmortem",
        "tags": ["km", "readback"],
        "source": "AI_EXTRACTED",
        "status": "APPROVED",
        "related_incident_id": None,
        "related_playbook_id": None,
        "related_approval_id": None,
        "path_type": None,
        "symptoms_hash": None,
        "view_count": 7,
        "created_by": "ai-agent",
        "created_at": datetime(2026, 7, 2, tzinfo=UTC),
        "updated_at": datetime(2026, 7, 2, tzinfo=UTC),
    }
    base.update(overrides)
    return SimpleNamespace(**base)


def test_knowledge_read_model_normalizes_uppercase_enum_names() -> None:
    repo = KnowledgeDBRepository(db=object())  # type: ignore[arg-type]

    entry = repo._to_model(_record())

    assert entry.entry_type == EntryType.POSTMORTEM
    assert entry.source == EntrySource.AI_EXTRACTED
    assert entry.status == EntryStatus.APPROVED


def test_knowledge_read_model_accepts_lowercase_enum_values() -> None:
    repo = KnowledgeDBRepository(db=object())  # type: ignore[arg-type]

    entry = repo._to_model(
        _record(
            entry_type="auto_runbook",
            source="human",
            status="published",
        )
    )

    assert entry.entry_type == EntryType.AUTO_RUNBOOK
    assert entry.source == EntrySource.HUMAN
    assert entry.status == EntryStatus.PUBLISHED


def test_knowledge_read_model_falls_back_for_unknown_legacy_enum_values() -> None:
    repo = KnowledgeDBRepository(db=object())  # type: ignore[arg-type]

    entry = repo._to_model(
        _record(
            entry_type="analysis_note",
            source="system_import",
            status="needs_ai_review",
        )
    )

    assert entry.entry_type == EntryType.BEST_PRACTICE
    assert entry.source == EntrySource.AI_EXTRACTED
    assert entry.status == EntryStatus.REVIEW


def test_knowledge_read_model_normalizes_legacy_tag_shapes_and_null_counts() -> None:
    repo = KnowledgeDBRepository(db=object())  # type: ignore[arg-type]

    json_tag_entry = repo._to_model(_record(tags='["telegram", " ai_loop "]', view_count=None))
    scalar_tag_entry = repo._to_model(_record(tags="telegram"))
    bad_count_entry = repo._to_model(_record(view_count="not-a-number"))

    assert json_tag_entry.tags == ["telegram", "ai_loop"]
    assert json_tag_entry.view_count == 0
    assert scalar_tag_entry.tags == ["telegram"]
    assert bad_count_entry.view_count == 0


def test_knowledge_read_model_fills_missing_legacy_timestamps() -> None:
    repo = KnowledgeDBRepository(db=object())  # type: ignore[arg-type]

    entry = repo._to_model(_record(created_at=None, updated_at=None))

    assert entry.created_at is not None
    assert entry.updated_at is not None


def test_knowledge_read_model_defaults_null_or_blank_category() -> None:
    repo = KnowledgeDBRepository(db=object())  # type: ignore[arg-type]

    assert repo._to_model(_record(category=None)).category == "general"
    assert repo._to_model(_record(category="  ")).category == "general"


class _ScalarResult:
    def __init__(self, value: int):
        self.value = value

    def scalar(self) -> int:
        return self.value


class _MappingResult:
    def __init__(self, rows: list[dict]):
        self.rows = rows

    def mappings(self):
        return self

    def all(self):
        return self.rows


class _RowsResult:
    def __init__(self, rows: list[SimpleNamespace]):
        self.rows = rows

    def all(self):
        return self.rows


class _FakeDb:
    def __init__(self):
        self.statements = []
        self._results = [
            _ScalarResult(1),
            _MappingResult([_record().__dict__]),
        ]

    async def execute(self, statement):
        self.statements.append(statement)
        return self._results.pop(0)


@pytest.mark.asyncio
async def test_list_entries_uses_enum_safe_text_filters() -> None:
    db = _FakeDb()
    repo = KnowledgeDBRepository(db=db)  # type: ignore[arg-type]

    entries, total = await repo.list_entries(
        entry_type=EntryType.POSTMORTEM,
        status=EntryStatus.APPROVED,
        limit=5,
    )

    assert total == 1
    assert entries[0].status == EntryStatus.APPROVED
    compiled = "\n".join(
        str(stmt.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}))
        for stmt in db.statements
    ).lower()
    assert "lower(cast(knowledge_entries.status as varchar))" in compiled
    assert "lower(cast(knowledge_entries.entry_type as varchar))" in compiled
    assert "approved" in compiled
    assert "postmortem" in compiled


@pytest.mark.asyncio
async def test_list_entries_uses_normalized_category_filter() -> None:
    db = _FakeDb()
    repo = KnowledgeDBRepository(db=db)  # type: ignore[arg-type]

    await repo.list_entries(category="general", limit=5)

    compiled = "\n".join(
        str(stmt.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}))
        for stmt in db.statements
    ).lower()
    assert "coalesce(nullif(trim(knowledge_entries.category)" in compiled
    assert "general" in compiled


@pytest.mark.asyncio
async def test_get_categories_normalizes_null_or_blank_category() -> None:
    class _CategoryDb:
        def __init__(self):
            self.statements = []

        async def execute(self, statement):
            self.statements.append(statement)
            return _RowsResult(
                [
                    SimpleNamespace(category=None, cnt=2),
                    SimpleNamespace(category="", cnt=1),
                    SimpleNamespace(category="service", cnt=3),
                ]
            )

    db = _CategoryDb()
    repo = KnowledgeDBRepository(db=db)  # type: ignore[arg-type]

    categories = await repo.get_categories()

    assert categories == [("general", 2), ("general", 1), ("service", 3)]
    compiled = str(
        db.statements[0].compile(
            dialect=postgresql.dialect(),
            compile_kwargs={"literal_binds": True},
        )
    ).lower()
    assert "coalesce(nullif(trim(knowledge_entries.category)" in compiled


@pytest.mark.asyncio
async def test_get_asset_taxonomy_counts_uses_stable_ai_automation_lenses() -> None:
    class _TaxonomyDb:
        def __init__(self):
            self.statements = []
            self.values = list(range(10, 10 + len(_ASSET_TAXONOMY_KEYS)))

        async def execute(self, statement):
            self.statements.append(statement)
            return _ScalarResult(self.values.pop(0))

    db = _TaxonomyDb()
    repo = KnowledgeDBRepository(db=db)  # type: ignore[arg-type]

    counts = await repo.get_asset_taxonomy_counts()

    assert [key for key, _ in counts] == list(_ASSET_TAXONOMY_KEYS)
    assert [count for _, count in counts] == list(range(10, 10 + len(_ASSET_TAXONOMY_KEYS)))
    compiled = "\n".join(
        str(stmt.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}))
        for stmt in db.statements
    ).lower()
    assert "source_control" in compiled
    assert "awoooi" in compiled
    assert "related_playbook_id is not null" in compiled


def test_enum_text_helper_casts_status_to_lowercase_text() -> None:
    compiled = str(
        (_enum_text(KnowledgeEntryRecord.status) == EntryStatus.ARCHIVED.value).compile(
            dialect=postgresql.dialect(),
            compile_kwargs={"literal_binds": True},
        )
    ).lower()

    assert "lower(cast(knowledge_entries.status as varchar))" in compiled
    assert "archived" in compiled