97 lines
4.4 KiB
Python
97 lines
4.4 KiB
Python
"""
|
|
FILE: tests/test_WP22_intelligence.py
|
|
DESCRIPTION: Integrationstest für WP-22.
|
|
FIX: Erzwingt Pfad-Synchronisation für Registry & Router.
|
|
"""
|
|
import unittest
|
|
import os
|
|
import shutil
|
|
import yaml
|
|
import asyncio
|
|
from unittest.mock import MagicMock, patch, AsyncMock
|
|
|
|
import app.routers.chat
|
|
from app.models.dto import ChatRequest, QueryHit, QueryRequest
|
|
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, chat_endpoint
|
|
|
|
class TestWP22Integration(unittest.IsolatedAsyncioTestCase):
|
|
|
|
async def asyncSetUp(self):
|
|
"""Bereitet eine isolierte Test-Umgebung vor."""
|
|
self.test_root = os.path.abspath("tests/temp_wp22")
|
|
self.test_vault = os.path.join(self.test_root, "vault_master")
|
|
self.test_config_dir = os.path.join(self.test_root, "config")
|
|
|
|
os.makedirs(os.path.join(self.test_vault, "01_User_Manual"), exist_ok=True)
|
|
os.makedirs(self.test_config_dir, exist_ok=True)
|
|
os.makedirs(os.path.join(self.test_root, "data/logs"), exist_ok=True)
|
|
|
|
# 2. Config Files schreiben
|
|
self.decision_path = os.path.join(self.test_config_dir, "decision_engine.yaml")
|
|
self.decision_config = {
|
|
"strategies": {
|
|
"FACT": {"trigger_keywords": ["was"], "edge_boosts": {"part_of": 2.0}},
|
|
"CAUSAL": {"trigger_keywords": ["warum"], "edge_boosts": {"caused_by": 3.0}}
|
|
}
|
|
}
|
|
with open(self.decision_path, "w") as f: yaml.dump(self.decision_config, f)
|
|
|
|
# 3. Vocabulary File am RICHTIGEN Ort
|
|
self.vocab_path = os.path.join(self.test_vault, "01_User_Manual/01_edge_vocabulary.md")
|
|
with open(self.vocab_path, "w") as f:
|
|
f.write("| System-Typ | Aliases |\n| :--- | :--- |\n| **caused_by** | ursache_ist |\n| **part_of** | teil_von |")
|
|
|
|
# 4. MOCKING / RESETTING GLOBAL STATE
|
|
self.mock_settings = MagicMock()
|
|
self.mock_settings.DECISION_CONFIG_PATH = self.decision_path
|
|
self.mock_settings.MINDNET_VAULT_ROOT = self.test_vault
|
|
self.mock_settings.RETRIEVER_TOP_K = 5
|
|
self.mock_settings.MODEL_NAME = "test-model"
|
|
|
|
self.patch_settings_chat = patch('app.routers.chat.get_settings', return_value=self.mock_settings)
|
|
self.patch_settings_registry = patch('app.services.edge_registry.get_settings', return_value=self.mock_settings)
|
|
|
|
self.patch_settings_chat.start()
|
|
self.patch_settings_registry.start()
|
|
|
|
app.routers.chat._DECISION_CONFIG_CACHE = None
|
|
EdgeRegistry._instance = None
|
|
self.registry = EdgeRegistry(vault_root=self.test_vault)
|
|
|
|
async def asyncTearDown(self):
|
|
self.patch_settings_chat.stop()
|
|
self.patch_settings_registry.stop()
|
|
if os.path.exists(self.test_root): shutil.rmtree(self.test_root)
|
|
EdgeRegistry._instance = None
|
|
app.routers.chat._DECISION_CONFIG_CACHE = None
|
|
|
|
def test_registry_resolution(self):
|
|
print("\n🔵 TEST 1: Registry Resolution")
|
|
self.assertTrue(len(self.registry.valid_types) > 0)
|
|
self.assertEqual(self.registry.resolve("ursache_ist"), "caused_by")
|
|
print("✅ Registry OK.")
|
|
|
|
def test_scoring_math(self):
|
|
print("\n🔵 TEST 2: Scoring Math (Lifecycle)")
|
|
with patch("app.core.retriever._get_scoring_weights", return_value=(1.0, 1.0, 0.0)):
|
|
self.assertEqual(_get_status_multiplier({"status": "stable"}), 1.2)
|
|
self.assertEqual(_get_status_multiplier({"status": "draft"}), 0.8)
|
|
print("✅ Scoring OK.")
|
|
|
|
async def test_full_flow(self):
|
|
print("\n🔵 TEST 3: Pipeline flow")
|
|
mock_llm = AsyncMock(); mock_llm.prompts = {}; mock_llm.generate_raw_response.return_value = "Ok"
|
|
mock_ret = AsyncMock()
|
|
mock_hit = QueryHit(node_id="c1", note_id="n1", semantic_score=0.8, edge_bonus=0.0, centrality_bonus=0.0, total_score=0.8, source={"text": "t"}, payload={"status": "active"})
|
|
mock_ret.search.return_value.results = [mock_hit]
|
|
|
|
resp = await chat_endpoint(ChatRequest(message="Warum?"), llm=mock_llm, retriever=mock_ret)
|
|
self.assertEqual(resp.intent, "CAUSAL")
|
|
called_req = mock_ret.search.call_args[0][0]
|
|
self.assertEqual(called_req.boost_edges.get("caused_by"), 3.0)
|
|
print("✅ Full Flow OK.")
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main() |