129 lines
3.6 KiB
Python
129 lines
3.6 KiB
Python
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 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
|
|
|
|
|
|
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 _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
|
|
|
|
|
|
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
|