feat(H): H-EVT/H-DATA-LIFE/H-RET/H-ASK/H-REV — knowledge lifecycle (event-driven extraction, derived-data propagation, retention policy, ask-knowledge with evidence, review queue), 42 tests passing

This commit is contained in:
Agent Zero
2026-08-18 23:17:33 +02:00
parent a29dc58bcd
commit a6e593dc40
2 changed files with 547 additions and 0 deletions
+100
View File
@@ -375,3 +375,103 @@ class TestKnowledgeExtraction:
result = await extract_knowledge("Hi", uuid.uuid4())
assert result.entities == []
assert result.relationships == []
# ─── H-EVT/H-DATA-LIFE/H-RET/H-ASK/H-REV: Knowledge Lifecycle ────────────────
class TestKnowledgeLifecycle:
"""Test the knowledge lifecycle module (H-EVT, H-DATA-LIFE, H-RET, H-ASK, H-REV)."""
def test_should_extract_known_events(self):
"""should_extract returns True for configured extraction triggers."""
from app.ai.knowledge_lifecycle import should_extract
assert should_extract("mail.received") is True
assert should_extract("dms.file_uploaded") is True
assert should_extract("wiki.article_published") is True
assert should_extract("communication.message_created") is True
def test_should_extract_unknown_events(self):
"""should_extract returns False for non-configured events."""
from app.ai.knowledge_lifecycle import should_extract
assert should_extract("contact.created") is False
assert should_extract("random.event") is False
assert should_extract("") is False
def test_extraction_triggers_count(self):
"""4 extraction triggers are configured."""
from app.ai.knowledge_lifecycle import EXTRACTION_TRIGGERS
assert len(EXTRACTION_TRIGGERS) == 4
def test_retention_days_wiki_unlimited(self):
"""Wiki retention is 0 (unlimited)."""
from app.ai.knowledge_lifecycle import get_retention_days
assert get_retention_days("wiki") == 0
def test_retention_days_dms_one_year(self):
"""DMS retention is 365 days."""
from app.ai.knowledge_lifecycle import get_retention_days
assert get_retention_days("dms") == 365
def test_retention_days_mail_180(self):
"""Mail retention is 180 days."""
from app.ai.knowledge_lifecycle import get_retention_days
assert get_retention_days("mail") == 180
def test_retention_days_communication_90(self):
"""Communication retention is 90 days."""
from app.ai.knowledge_lifecycle import get_retention_days
assert get_retention_days("communication") == 90
def test_retention_days_unknown_default(self):
"""Unknown source type gets default retention of 180 days."""
from app.ai.knowledge_lifecycle import get_retention_days
assert get_retention_days("unknown") == 180
@pytest.mark.asyncio
async def test_handle_extraction_event_unknown_event(self):
"""handle_extraction_event returns None for unknown events."""
from app.ai.knowledge_lifecycle import handle_extraction_event
result = await handle_extraction_event(
db=MagicMock(),
tenant_id=uuid.uuid4(),
event_name="unknown.event",
payload={"entity_id": str(uuid.uuid4())},
)
assert result is None
@pytest.mark.asyncio
async def test_handle_extraction_event_no_entity_id(self):
"""handle_extraction_event returns None when no entity_id in payload."""
from app.ai.knowledge_lifecycle import handle_extraction_event
result = await handle_extraction_event(
db=MagicMock(),
tenant_id=uuid.uuid4(),
event_name="mail.received",
payload={},
)
assert result is None
@pytest.mark.asyncio
async def test_ask_knowledge_empty_query(self):
"""ask_knowledge returns empty result for empty query."""
from app.ai.knowledge_lifecycle import ask_knowledge
result = await ask_knowledge(
db=MagicMock(),
tenant_id=uuid.uuid4(),
user_id=uuid.uuid4(),
query="",
)
assert result["answer"] == ""
assert result["evidence"] == []