"""Phase H Knowledge Plugin Tests — integration tests with real DB.""" from __future__ import annotations import uuid import pytest import pytest_asyncio from unittest.mock import AsyncMock, patch from sqlalchemy import select from app.core.db import Base from app.models.tenant import Tenant from app.models.user import User from app.core.auth import hash_password from app.plugins.builtins.knowledge.models import KnowledgeExtraction from app.plugins.builtins.knowledge.services import extract_knowledge, ask_knowledge, get_review_queue, review_extraction pytestmark = pytest.mark.asyncio @pytest_asyncio.fixture async def seed(db_session): """Create tenant + admin user for tests.""" tenant = Tenant(name="Test Tenant", slug="test-tenant") db_session.add(tenant) await db_session.flush() admin = User( email="admin@test.local", name="Admin", password_hash=hash_password("Test123!"), is_active=True, ) db_session.add(admin) await db_session.flush() return {"tenant": tenant, "admin": admin} def _mock_llm_response(content: str = '{"entities": [], "relationships": [], "confidence": 0.5}'): """Create a mock LLM response.""" return AsyncMock(return_value={"content": content, "cost_usd": 0.001}) class TestKnowledgeExtraction: """Test extract_knowledge service.""" async def test_extract_creates_extraction_record(self, db_session, seed): """extract_knowledge creates a KnowledgeExtraction record in DB.""" source_id = uuid.uuid4() mock_response = {"content": '{"entities": [{"type": "person", "name": "John Doe"}], "relationships": [], "confidence": 0.7}', "cost_usd": 0.001} with patch("app.plugins.builtins.knowledge.services.llm_complete", new_callable=AsyncMock, return_value=mock_response): result = await extract_knowledge( db=db_session, tenant_id=seed["tenant"].id, source_type="wiki_article", source_id=source_id, source_title="Test Article", source_text="John Doe works at ACME Corp. He is the CEO.", ) assert "id" in result assert "entities" in result assert "relationships" in result assert "confidence" in result assert "status" in result # Verify DB record ext = await db_session.execute( select(KnowledgeExtraction).where( KnowledgeExtraction.tenant_id == seed["tenant"].id ) ) records = ext.scalars().all() assert len(records) >= 1 assert records[0].source_type == "wiki_article" assert records[0].source_title == "Test Article" async def test_extract_with_empty_text(self, db_session, seed): """extract_knowledge handles empty text gracefully.""" source_id = uuid.uuid4() mock_response = {"content": '{"entities": [], "relationships": [], "confidence": 0.0}', "cost_usd": 0.001} with patch("app.plugins.builtins.knowledge.services.llm_complete", new_callable=AsyncMock, return_value=mock_response): result = await extract_knowledge( db=db_session, tenant_id=seed["tenant"].id, source_type="wiki_article", source_id=source_id, source_title="Empty Article", source_text="", ) assert "id" in result assert result["entities"] == [] assert result["relationships"] == [] class TestReviewQueue: """Test review queue service.""" async def test_review_queue_returns_pending(self, db_session, seed): """get_review_queue returns pending extractions.""" ext = KnowledgeExtraction( tenant_id=seed["tenant"].id, source_type="wiki_article", source_id=uuid.uuid4(), source_title="Test", extracted_entities=[{"type": "person", "name": "John"}], extracted_relationships=[], confidence=0.5, status="pending", ) db_session.add(ext) await db_session.flush() result = await get_review_queue(db=db_session, tenant_id=seed["tenant"].id) assert result["total"] >= 1 assert any(item["id"] == str(ext.id) for item in result["items"]) async def test_review_queue_excludes_approved(self, db_session, seed): """get_review_queue excludes approved extractions.""" ext = KnowledgeExtraction( tenant_id=seed["tenant"].id, source_type="wiki_article", source_id=uuid.uuid4(), source_title="Approved", extracted_entities=[], extracted_relationships=[], confidence=0.9, status="approved", ) db_session.add(ext) await db_session.flush() result = await get_review_queue(db=db_session, tenant_id=seed["tenant"].id) assert not any(item["id"] == str(ext.id) for item in result["items"]) async def test_review_approve(self, db_session, seed): """review_extraction approves a pending extraction.""" ext = KnowledgeExtraction( tenant_id=seed["tenant"].id, source_type="wiki_article", source_id=uuid.uuid4(), source_title="To Approve", extracted_entities=[{"type": "person", "name": "Jane"}], extracted_relationships=[], confidence=0.6, status="pending", ) db_session.add(ext) await db_session.flush() result = await review_extraction( db=db_session, tenant_id=seed["tenant"].id, extraction_id=ext.id, approved=True, user_id=seed["admin"].id, notes="Looks good", ) assert result["status"] == "approved" await db_session.refresh(ext) assert ext.status == "approved" assert ext.reviewed_by == seed["admin"].id assert ext.review_notes == "Looks good" async def test_review_reject(self, db_session, seed): """review_extraction rejects a pending extraction.""" ext = KnowledgeExtraction( tenant_id=seed["tenant"].id, source_type="wiki_article", source_id=uuid.uuid4(), source_title="To Reject", extracted_entities=[], extracted_relationships=[], confidence=0.3, status="pending", ) db_session.add(ext) await db_session.flush() result = await review_extraction( db=db_session, tenant_id=seed["tenant"].id, extraction_id=ext.id, approved=False, user_id=seed["admin"].id, ) assert result["status"] == "rejected" await db_session.refresh(ext) assert ext.status == "rejected" async def test_review_nonexistent(self, db_session, seed): """review_extraction returns error for nonexistent extraction.""" result = await review_extraction( db=db_session, tenant_id=seed["tenant"].id, extraction_id=uuid.uuid4(), approved=True, user_id=seed["admin"].id, ) assert "error" in result class TestKnowledgeModel: """Test KnowledgeExtraction model fields.""" async def test_model_fields(self, db_session, seed): """KnowledgeExtraction has all required fields.""" ext = KnowledgeExtraction( tenant_id=seed["tenant"].id, source_type="dms_file", source_id=uuid.uuid4(), source_title="Test File", extracted_entities=[{"type": "company", "name": "ACME"}], extracted_relationships=[{"source": "John", "target": "ACME", "type": "works_for"}], confidence=0.85, status="auto_created", llm_model="openai/gpt-4o-mini", llm_cost_usd=0.001, ) db_session.add(ext) await db_session.flush() assert ext.id is not None assert ext.source_type == "dms_file" assert ext.confidence == 0.85 assert ext.status == "auto_created" assert ext.llm_model == "openai/gpt-4o-mini" assert len(ext.extracted_entities) == 1 assert len(ext.extracted_relationships) == 1 async def test_model_defaults(self, db_session, seed): """KnowledgeExtraction has correct defaults.""" ext = KnowledgeExtraction( tenant_id=seed["tenant"].id, source_type="mail", source_id=uuid.uuid4(), ) db_session.add(ext) await db_session.flush() assert ext.confidence == 0.0 assert ext.status == "pending" assert ext.llm_cost_usd == 0.0 assert ext.extracted_entities == [] assert ext.extracted_relationships == []