188 lines
7.6 KiB
Python
188 lines
7.6 KiB
Python
"""
|
|
FILE: tests/test_WP22_integration.py
|
|
DESCRIPTION: Integrationstest für WP-22 (Graph Intelligence).
|
|
FIXES: Pydantic Validation & Config Caching Issues.
|
|
"""
|
|
import unittest
|
|
import os
|
|
import shutil
|
|
import json
|
|
import yaml
|
|
import asyncio
|
|
from unittest.mock import MagicMock, patch, AsyncMock
|
|
|
|
# Wir importieren das Modul direkt, um auf den Cache zuzugreifen
|
|
import app.routers.chat
|
|
|
|
# DTOs und Logik
|
|
from app.models.dto import ChatRequest, QueryRequest, QueryHit
|
|
from app.services.edge_registry import EdgeRegistry
|
|
from app.core.retriever import _compute_total_score, _get_status_multiplier
|
|
from app.routers.chat import _classify_intent, get_decision_strategy, chat_endpoint
|
|
|
|
class TestWP22Integration(unittest.IsolatedAsyncioTestCase):
|
|
|
|
def setUp(self):
|
|
"""Bereitet eine isolierte Test-Umgebung vor."""
|
|
self.test_dir = "tests/temp_integration"
|
|
|
|
# 1. Environment Patching
|
|
self.os_env_patch = patch.dict(os.environ, {
|
|
"MINDNET_VAULT_ROOT": self.test_dir,
|
|
"MINDNET_DECISION_CONFIG": os.path.join(self.test_dir, "config", "decision_engine.yaml"),
|
|
"MINDNET_TYPES_FILE": os.path.join(self.test_dir, "config", "types.yaml")
|
|
})
|
|
self.os_env_patch.start()
|
|
|
|
# 2. Verzeichnisse erstellen
|
|
os.makedirs(os.path.join(self.test_dir, "config"), exist_ok=True)
|
|
os.makedirs(os.path.join(self.test_dir, "01_User_Manual"), exist_ok=True)
|
|
os.makedirs(os.path.join(self.test_dir, "data", "logs"), exist_ok=True)
|
|
|
|
# 3. Config: decision_engine.yaml schreiben (Test-Definition)
|
|
self.decision_config = {
|
|
"strategies": {
|
|
"FACT": {
|
|
"trigger_keywords": ["was ist"],
|
|
"edge_boosts": {"part_of": 2.0} # Kein 'caused_by' hier!
|
|
},
|
|
"CAUSAL": {
|
|
"trigger_keywords": ["warum", "weshalb"],
|
|
"edge_boosts": {"caused_by": 3.0, "related_to": 0.5}
|
|
}
|
|
}
|
|
}
|
|
with open(os.environ["MINDNET_DECISION_CONFIG"], "w") as f:
|
|
yaml.dump(self.decision_config, f)
|
|
|
|
# 4. Config: Edge Vocabulary schreiben
|
|
vocab_path = os.path.join(self.test_dir, "01_User_Manual", "01_edge_vocabulary.md")
|
|
with open(vocab_path, "w") as f:
|
|
f.write("| **caused_by** | ursache_ist, wegen |\n| **part_of** | teil_von |")
|
|
|
|
# 5. CACHE RESET (WICHTIG!)
|
|
# Damit der Router die oben geschriebene YAML auch wirklich liest:
|
|
app.routers.chat._DECISION_CONFIG_CACHE = None
|
|
EdgeRegistry._instance = None
|
|
|
|
# Registry neu init
|
|
self.registry = EdgeRegistry(vault_root=self.test_dir)
|
|
|
|
def tearDown(self):
|
|
self.os_env_patch.stop()
|
|
if os.path.exists(self.test_dir):
|
|
shutil.rmtree(self.test_dir)
|
|
EdgeRegistry._instance = None
|
|
app.routers.chat._DECISION_CONFIG_CACHE = None
|
|
|
|
# ------------------------------------------------------------------------
|
|
# TEST 1: Edge Registry & Validation
|
|
# ------------------------------------------------------------------------
|
|
def test_edge_registry_aliases(self):
|
|
print("\n🔵 TEST 1: Edge Registry Resolution")
|
|
resolved = self.registry.resolve("ursache_ist")
|
|
self.assertEqual(resolved, "caused_by")
|
|
|
|
unknown = self.registry.resolve("foobar_link")
|
|
self.assertEqual(unknown, "foobar_link")
|
|
|
|
log_path = self.registry.unknown_log_path
|
|
self.assertTrue(os.path.exists(log_path))
|
|
print("✅ Registry funktioniert.")
|
|
|
|
# ------------------------------------------------------------------------
|
|
# TEST 2: Lifecycle Scoring
|
|
# ------------------------------------------------------------------------
|
|
def test_lifecycle_scoring_logic(self):
|
|
print("\n🔵 TEST 2: Lifecycle Scoring")
|
|
with patch("app.core.retriever._get_scoring_weights", return_value=(1.0, 0.5, 0.0)):
|
|
base_sem = 0.9
|
|
|
|
payload_draft = {"status": "draft", "retriever_weight": 1.0}
|
|
mult_draft = _get_status_multiplier(payload_draft)
|
|
self.assertEqual(mult_draft, 0.8)
|
|
|
|
payload_stable = {"status": "stable", "retriever_weight": 1.0}
|
|
mult_stable = _get_status_multiplier(payload_stable)
|
|
self.assertEqual(mult_stable, 1.2)
|
|
print("✅ Lifecycle Scoring korrekt.")
|
|
|
|
# ------------------------------------------------------------------------
|
|
# TEST 3: Semantic Router & Boosting
|
|
# ------------------------------------------------------------------------
|
|
async def test_router_integration(self):
|
|
print("\n🔵 TEST 3: Semantic Router Integration")
|
|
|
|
mock_llm = MagicMock()
|
|
mock_llm.prompts = {}
|
|
|
|
# Da der Cache im setUp gelöscht wurde, sollte er jetzt CAUSAL finden
|
|
query_causal = "Warum ist das Projekt gescheitert?"
|
|
intent, source = await _classify_intent(query_causal, mock_llm)
|
|
|
|
self.assertEqual(intent, "CAUSAL", f"Erwartete CAUSAL, bekam {intent} via {source}")
|
|
|
|
strategy = get_decision_strategy(intent)
|
|
boosts = strategy.get("edge_boosts", {})
|
|
self.assertEqual(boosts.get("caused_by"), 3.0)
|
|
print("✅ Router lädt Config korrekt.")
|
|
|
|
# ------------------------------------------------------------------------
|
|
# TEST 4: Full Pipeline
|
|
# ------------------------------------------------------------------------
|
|
async def test_full_pipeline_flow(self):
|
|
print("\n🔵 TEST 4: Full Chat Pipeline")
|
|
|
|
mock_llm = AsyncMock()
|
|
mock_llm.prompts = {}
|
|
mock_llm.generate_raw_response.return_value = "Antwort."
|
|
|
|
mock_retriever = AsyncMock()
|
|
# FIX: note_id hinzugefügt für Pydantic
|
|
mock_hit = QueryHit(
|
|
node_id="123",
|
|
note_id="test_note_123", # <--- WICHTIG
|
|
semantic_score=0.9,
|
|
edge_bonus=0.5,
|
|
centrality_bonus=0.0,
|
|
total_score=1.0,
|
|
source={"text": "Inhalt"},
|
|
payload={"type": "concept"}
|
|
)
|
|
mock_retriever.search.return_value.results = [mock_hit]
|
|
|
|
req = ChatRequest(message="Warum ist das passiert?", top_k=3)
|
|
|
|
response = await chat_endpoint(req, llm=mock_llm, retriever=mock_retriever)
|
|
|
|
called_query_req = mock_retriever.search.call_args[0][0]
|
|
self.assertEqual(called_query_req.boost_edges.get("caused_by"), 3.0)
|
|
self.assertEqual(response.intent, "CAUSAL")
|
|
print("✅ Pipeline reicht Boosts weiter.")
|
|
|
|
# ------------------------------------------------------------------------
|
|
# TEST 5: Regression Check
|
|
# ------------------------------------------------------------------------
|
|
async def test_regression_standard_query(self):
|
|
print("\n🔵 TEST 5: Regression")
|
|
|
|
mock_llm = AsyncMock()
|
|
mock_llm.prompts = {}
|
|
mock_llm.generate_raw_response.return_value = "Antwort."
|
|
|
|
mock_retriever = AsyncMock()
|
|
mock_retriever.search.return_value.results = []
|
|
|
|
req = ChatRequest(message="Was ist das?", top_k=3)
|
|
|
|
response = await chat_endpoint(req, llm=mock_llm, retriever=mock_retriever)
|
|
|
|
called_query_req = mock_retriever.search.call_args[0][0]
|
|
|
|
# FACT strategy hat in unserem Test Setup NUR 'part_of', KEIN 'caused_by'
|
|
self.assertEqual(response.intent, "FACT")
|
|
self.assertNotIn("caused_by", called_query_req.boost_edges or {})
|
|
print("✅ Regression Test bestanden.")
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main() |