Agent Zero
478 lines
18 KiB
Python
478 lines
18 KiB
Python
"""Tests for Phase H — Wiki plugin: models, schemas, plugin manifest.
|
|
|
|
All tests use mocks — no real DB needed.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
|
|
class TestWikiPluginManifest:
|
|
"""Test the Wiki plugin manifest."""
|
|
|
|
def test_plugin_manifest_valid(self):
|
|
"""Wiki plugin manifest is valid."""
|
|
from app.plugins.builtins.wiki.plugin import WikiPlugin
|
|
|
|
assert WikiPlugin.manifest.name == "wiki"
|
|
assert WikiPlugin.manifest.version == "1.0.0"
|
|
assert WikiPlugin.manifest.display_name == "Wiki"
|
|
assert "wiki:read" in WikiPlugin.manifest.permissions
|
|
assert "wiki:write" in WikiPlugin.manifest.permissions
|
|
assert "wiki:delete" in WikiPlugin.manifest.permissions
|
|
assert "wiki:admin" in WikiPlugin.manifest.permissions
|
|
|
|
def test_plugin_has_routes(self):
|
|
"""Wiki plugin has routes defined."""
|
|
from app.plugins.builtins.wiki.plugin import WikiPlugin
|
|
|
|
assert len(WikiPlugin.manifest.routes) > 0
|
|
assert WikiPlugin.manifest.routes[0].path == "/api/v1/wiki"
|
|
|
|
def test_plugin_has_menu_item(self):
|
|
"""Wiki plugin has a menu item."""
|
|
from app.plugins.builtins.wiki.plugin import WikiPlugin
|
|
|
|
assert len(WikiPlugin.manifest.menu_items) > 0
|
|
assert WikiPlugin.manifest.menu_items[0].path == "/wiki"
|
|
|
|
def test_plugin_has_dependencies(self):
|
|
"""Wiki plugin depends on permissions plugin."""
|
|
from app.plugins.builtins.wiki.plugin import WikiPlugin
|
|
|
|
assert "permissions" in WikiPlugin.manifest.dependencies
|
|
|
|
|
|
class TestWikiRoutes:
|
|
"""Test the Wiki routes are properly defined."""
|
|
|
|
def test_routes_count(self):
|
|
"""Wiki plugin has 9 routes."""
|
|
from app.plugins.builtins.wiki.routes import router
|
|
|
|
assert len(router.routes) == 9
|
|
|
|
def test_routes_have_correct_paths(self):
|
|
"""Wiki routes have correct API paths."""
|
|
from app.plugins.builtins.wiki.routes import router
|
|
|
|
paths = {r.path for r in router.routes}
|
|
assert "/api/v1/wiki/articles" in paths
|
|
assert "/api/v1/wiki/articles/{article_id}" in paths
|
|
assert "/api/v1/wiki/articles/{article_id}/versions" in paths
|
|
assert "/api/v1/wiki/articles/{article_id}/versions/{version}/restore" in paths
|
|
assert "/api/v1/wiki/categories" in paths
|
|
|
|
|
|
class TestWikiSchemas:
|
|
"""Test the Wiki Pydantic schemas."""
|
|
|
|
def test_article_create_valid(self):
|
|
"""ArticleCreate accepts valid data."""
|
|
from app.plugins.builtins.wiki.schemas import ArticleCreate
|
|
|
|
article = ArticleCreate(title="Test Article", slug="test-article")
|
|
assert article.title == "Test Article"
|
|
assert article.slug == "test-article"
|
|
assert article.status == "draft"
|
|
assert article.tags == []
|
|
|
|
def test_article_create_invalid_status(self):
|
|
"""ArticleCreate rejects invalid status."""
|
|
from app.plugins.builtins.wiki.schemas import ArticleCreate
|
|
from pydantic import ValidationError
|
|
|
|
with pytest.raises(ValidationError):
|
|
ArticleCreate(title="Test", slug="test", status="invalid")
|
|
|
|
def test_article_update_partial(self):
|
|
"""ArticleUpdate allows partial updates."""
|
|
from app.plugins.builtins.wiki.schemas import ArticleUpdate
|
|
|
|
update = ArticleUpdate(title="Updated Title")
|
|
assert update.title == "Updated Title"
|
|
assert update.content is None
|
|
|
|
def test_category_create_valid(self):
|
|
"""CategoryCreate accepts valid data."""
|
|
from app.plugins.builtins.wiki.schemas import CategoryCreate
|
|
|
|
cat = CategoryCreate(name="Test Category", slug="test-category")
|
|
assert cat.name == "Test Category"
|
|
assert cat.slug == "test-category"
|
|
assert cat.sort_order == 0
|
|
|
|
|
|
class TestWikiModels:
|
|
"""Test the Wiki SQLAlchemy models."""
|
|
|
|
def test_wiki_article_model_fields(self):
|
|
"""WikiArticle model has all required fields."""
|
|
from app.plugins.builtins.wiki.models import WikiArticle
|
|
|
|
assert hasattr(WikiArticle, "title")
|
|
assert hasattr(WikiArticle, "slug")
|
|
assert hasattr(WikiArticle, "content")
|
|
assert hasattr(WikiArticle, "content_html")
|
|
assert hasattr(WikiArticle, "summary")
|
|
assert hasattr(WikiArticle, "category_id")
|
|
assert hasattr(WikiArticle, "tags")
|
|
assert hasattr(WikiArticle, "status")
|
|
assert hasattr(WikiArticle, "entity_links")
|
|
assert hasattr(WikiArticle, "version")
|
|
assert hasattr(WikiArticle, "published_at")
|
|
assert hasattr(WikiArticle, "tenant_id")
|
|
assert hasattr(WikiArticle, "owner_id")
|
|
|
|
def test_wiki_category_model_fields(self):
|
|
"""WikiCategory model has all required fields."""
|
|
from app.plugins.builtins.wiki.models import WikiCategory
|
|
|
|
assert hasattr(WikiCategory, "name")
|
|
assert hasattr(WikiCategory, "slug")
|
|
assert hasattr(WikiCategory, "description")
|
|
assert hasattr(WikiCategory, "parent_id")
|
|
assert hasattr(WikiCategory, "sort_order")
|
|
assert hasattr(WikiCategory, "tenant_id")
|
|
|
|
def test_wiki_article_version_model_fields(self):
|
|
"""WikiArticleVersion model has all required fields."""
|
|
from app.plugins.builtins.wiki.models import WikiArticleVersion
|
|
|
|
assert hasattr(WikiArticleVersion, "article_id")
|
|
assert hasattr(WikiArticleVersion, "version")
|
|
assert hasattr(WikiArticleVersion, "title")
|
|
assert hasattr(WikiArticleVersion, "content")
|
|
assert hasattr(WikiArticleVersion, "edited_by")
|
|
assert hasattr(WikiArticleVersion, "edit_comment")
|
|
assert hasattr(WikiArticleVersion, "tenant_id")
|
|
|
|
def test_models_have_tenant_mixin(self):
|
|
"""All Wiki models have tenant_id (TenantMixin)."""
|
|
from app.plugins.builtins.wiki.models import WikiArticle, WikiCategory, WikiArticleVersion
|
|
|
|
for model in [WikiArticle, WikiCategory, WikiArticleVersion]:
|
|
assert hasattr(model, "tenant_id")
|
|
|
|
|
|
class TestWikiServices:
|
|
"""Test the Wiki service helper functions."""
|
|
|
|
def test_slugify(self):
|
|
"""_slugify converts text to URL-safe slug."""
|
|
from app.plugins.builtins.wiki.services import _slugify
|
|
|
|
assert _slugify("Hello World") == "hello-world"
|
|
assert _slugify("Überblick") == "ueberblick"
|
|
assert _slugify("Test Article 123") == "test-article-123"
|
|
|
|
|
|
# ─── H-SRC: Knowledge Source Adapter ─────────────────────────────────────────
|
|
|
|
|
|
class TestKnowledgeSourceAdapter:
|
|
"""Test the knowledge source adapter (H-SRC)."""
|
|
|
|
def test_available_sources(self):
|
|
"""All 4 knowledge source types are registered."""
|
|
from app.ai.knowledge_sources import get_available_sources
|
|
|
|
sources = get_available_sources()
|
|
types = {s["type"] for s in sources}
|
|
assert types == {"wiki", "dms", "mail", "communication"}
|
|
|
|
def test_get_source_config_wiki(self):
|
|
"""Wiki source config is correct."""
|
|
from app.ai.knowledge_sources import get_source_config
|
|
|
|
config = get_source_config("wiki")
|
|
assert config is not None
|
|
assert config["text_field"] == "content"
|
|
assert config["title_field"] == "title"
|
|
assert config["status_filter"] == {"status": "published"}
|
|
|
|
def test_get_source_config_unknown(self):
|
|
"""Unknown source type returns None."""
|
|
from app.ai.knowledge_sources import get_source_config
|
|
|
|
assert get_source_config("nonexistent") is None
|
|
|
|
|
|
# ─── H-CITE: Evidence References ─────────────────────────────────────────────
|
|
|
|
|
|
class TestEvidenceReferences:
|
|
"""Test the evidence reference builder (H-CITE)."""
|
|
|
|
def test_evidence_reference_to_dict(self):
|
|
"""EvidenceReference serializes to dict correctly."""
|
|
from app.ai.knowledge_sources import EvidenceReference
|
|
|
|
ref = EvidenceReference(
|
|
source_type="wiki",
|
|
source_id="abc-123",
|
|
title="Test Article",
|
|
url="/wiki/articles/abc-123",
|
|
snippet="This is a test snippet",
|
|
confidence=0.95,
|
|
)
|
|
d = ref.to_dict()
|
|
assert d["source_type"] == "wiki"
|
|
assert d["source_id"] == "abc-123"
|
|
assert d["title"] == "Test Article"
|
|
assert d["confidence"] == 0.95
|
|
|
|
def test_evidence_reference_to_workstream_block(self):
|
|
"""EvidenceReference converts to workstream block."""
|
|
from app.ai.knowledge_sources import EvidenceReference
|
|
|
|
ref = EvidenceReference(
|
|
source_type="dms",
|
|
source_id="file-123",
|
|
title="Document.pdf",
|
|
url="/dms/files/file-123",
|
|
snippet="Important content here",
|
|
)
|
|
block = ref.to_workstream_block()
|
|
assert block["type"] == "evidence_card"
|
|
assert block["source_type"] == "dms"
|
|
assert block["title"] == "Document.pdf"
|
|
|
|
def test_build_evidence_references_from_results(self):
|
|
"""build_evidence_references creates refs from search results."""
|
|
from app.ai.knowledge_sources import build_evidence_references
|
|
|
|
results = [
|
|
{"source_type": "wiki", "source_id": str(uuid.uuid4()), "title": "Article 1", "score": 0.9, "snippet": "Snippet 1"},
|
|
{"source_type": "dms", "source_id": str(uuid.uuid4()), "title": "Doc 2", "score": 0.7, "snippet": "Snippet 2"},
|
|
{"source_type": "mail", "source_id": str(uuid.uuid4()), "title": "Mail 3", "score": 0.95, "snippet": "Snippet 3"},
|
|
]
|
|
refs = build_evidence_references(results)
|
|
assert len(refs) == 3
|
|
# Sorted by confidence descending
|
|
assert refs[0].confidence == 0.95
|
|
assert refs[1].confidence == 0.9
|
|
assert refs[2].confidence == 0.7
|
|
|
|
def test_build_evidence_references_max_results(self):
|
|
"""build_evidence_references respects max_results."""
|
|
from app.ai.knowledge_sources import build_evidence_references
|
|
|
|
results = [
|
|
{"source_type": "wiki", "source_id": str(uuid.uuid4()), "title": f"Article {i}", "score": 0.5}
|
|
for i in range(10)
|
|
]
|
|
refs = build_evidence_references(results, max_results=3)
|
|
assert len(refs) == 3
|
|
|
|
def test_build_evidence_references_empty(self):
|
|
"""build_evidence_references handles empty results."""
|
|
from app.ai.knowledge_sources import build_evidence_references
|
|
|
|
refs = build_evidence_references([])
|
|
assert refs == []
|
|
|
|
|
|
# ─── H-EXT/H-ENT/H-AUTO/H-CONF: Knowledge Extraction ─────────────────────────
|
|
|
|
|
|
class TestKnowledgeExtraction:
|
|
"""Test the knowledge extraction module (H-EXT, H-ENT, H-AUTO, H-CONF)."""
|
|
|
|
def test_extracted_entity_dataclass(self):
|
|
"""ExtractedEntity dataclass works correctly."""
|
|
from app.ai.knowledge_extraction import ExtractedEntity
|
|
|
|
entity = ExtractedEntity(name="John Doe", entity_type="person", confidence=0.9)
|
|
assert entity.name == "John Doe"
|
|
assert entity.entity_type == "person"
|
|
assert entity.confidence == 0.9
|
|
assert entity.mentions == []
|
|
assert entity.metadata == {}
|
|
|
|
def test_extracted_relationship_dataclass(self):
|
|
"""ExtractedRelationship dataclass works correctly."""
|
|
from app.ai.knowledge_extraction import ExtractedRelationship
|
|
|
|
rel = ExtractedRelationship(
|
|
source_entity="John Doe", source_type="person",
|
|
target_entity="Acme Corp", target_type="company",
|
|
relationship_type="works_for", confidence=0.85,
|
|
evidence="John works at Acme",
|
|
)
|
|
assert rel.source_entity == "John Doe"
|
|
assert rel.relationship_type == "works_for"
|
|
assert rel.confidence == 0.85
|
|
assert rel.evidence == "John works at Acme"
|
|
|
|
def test_extraction_result_dataclass(self):
|
|
"""ExtractionResult dataclass works correctly."""
|
|
from app.ai.knowledge_extraction import ExtractionResult
|
|
|
|
result = ExtractionResult(source_type="wiki", source_id="abc", tenant_id="123")
|
|
assert result.entities == []
|
|
assert result.relationships == []
|
|
assert result.overall_confidence == 0.0
|
|
|
|
def test_is_low_confidence(self):
|
|
"""is_low_confidence correctly identifies low confidence scores."""
|
|
from app.ai.knowledge_extraction import is_low_confidence, LOW_CONFIDENCE_THRESHOLD
|
|
|
|
assert is_low_confidence(0.3) is True
|
|
assert is_low_confidence(0.5) is True
|
|
assert is_low_confidence(0.59) is True
|
|
assert is_low_confidence(0.6) is False
|
|
assert is_low_confidence(0.9) is False
|
|
assert LOW_CONFIDENCE_THRESHOLD == 0.6
|
|
|
|
def test_filter_high_confidence(self):
|
|
"""filter_high_confidence splits relationships correctly."""
|
|
from app.ai.knowledge_extraction import filter_high_confidence, ExtractedRelationship
|
|
|
|
rels = [
|
|
ExtractedRelationship("A", "person", "B", "company", "works_for", 0.9),
|
|
ExtractedRelationship("C", "person", "D", "company", "related_to", 0.3),
|
|
ExtractedRelationship("E", "person", "F", "company", "knows", 0.7),
|
|
ExtractedRelationship("G", "person", "H", "company", "met", 0.5),
|
|
]
|
|
high, low = filter_high_confidence(rels)
|
|
assert len(high) == 2 # 0.9 and 0.7
|
|
assert len(low) == 2 # 0.3 and 0.5
|
|
assert high[0].confidence == 0.9
|
|
assert high[1].confidence == 0.7
|
|
|
|
def test_filter_high_confidence_custom_threshold(self):
|
|
"""filter_high_confidence respects custom threshold."""
|
|
from app.ai.knowledge_extraction import filter_high_confidence, ExtractedRelationship
|
|
|
|
rels = [
|
|
ExtractedRelationship("A", "person", "B", "company", "works_for", 0.8),
|
|
ExtractedRelationship("C", "person", "D", "company", "related_to", 0.7),
|
|
]
|
|
high, low = filter_high_confidence(rels, threshold=0.75)
|
|
assert len(high) == 1 # 0.8
|
|
assert len(low) == 1 # 0.7
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_knowledge_empty_text(self):
|
|
"""extract_knowledge returns empty result for empty text."""
|
|
from app.ai.knowledge_extraction import extract_knowledge
|
|
|
|
result = await extract_knowledge("", uuid.uuid4())
|
|
assert result.entities == []
|
|
assert result.relationships == []
|
|
assert result.overall_confidence == 0.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_knowledge_short_text(self):
|
|
"""extract_knowledge returns empty result for very short text."""
|
|
from app.ai.knowledge_extraction import extract_knowledge
|
|
|
|
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"] == []
|