def6435140
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 2m33s
CD Pipeline / build-and-deploy (push) Successful in 4m46s
CD Pipeline / post-deploy-checks (push) Has been cancelled
266 lines
8.0 KiB
Python
266 lines
8.0 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 (
|
|
_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
|
|
|
|
def one(self):
|
|
return self.rows[0]
|
|
|
|
|
|
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 = {
|
|
key: count
|
|
for count, key in enumerate(_ASSET_TAXONOMY_KEYS, start=10)
|
|
}
|
|
|
|
async def execute(self, statement):
|
|
self.statements.append(statement)
|
|
return _MappingResult([self.values])
|
|
|
|
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)))
|
|
assert len(db.statements) == 1
|
|
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
|