mindnet/tests/test_WP22_intelligence.py

"""
FILE: tests/test_WP22_intelligence.py
DESCRIPTION: Integrationstest für WP-22. 
             FIX: Erzwingt Pfad-Synchronisation für Registry & Router. Behebt Pydantic Validation Errors.
"""
import unittest
import os
import shutil
import yaml
import asyncio
from unittest.mock import MagicMock, patch, AsyncMock

# --- Modul-Caching Fix: Wir müssen Caches leeren ---
import app.routers.chat
from app.models.dto import ChatRequest, QueryHit, QueryRequest
from app.services.edge_registry import EdgeRegistry
from app.core.retrieval.retriever_scoring import compute_wp22_score, get_status_multiplier
from app.routers.chat import _classify_intent, get_decision_strategy, chat_endpoint

class TestWP22Integration(unittest.IsolatedAsyncioTestCase):

    async def asyncSetUp(self):
        """Bereitet eine isolierte Test-Umgebung vor."""
        # Wir simulieren hier 'vault_master' (oder venv_master) als Verzeichnis
        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")
        
        # 1. Pfade erstellen
        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 (MOCK CONFIG)
        self.decision_path = os.path.join(self.test_config_dir, "decision_engine.yaml")
        self.decision_config = {
            "strategies": {
                "FACT": {
                    "trigger_keywords": ["was ist"],
                    "edge_boosts": {"part_of": 2.0}
                },
                "CAUSAL": {
                    "trigger_keywords": ["warum"],
                    "edge_boosts": {"caused_by": 3.0}
                }
            }
        }
        with open(self.decision_path, "w", encoding="utf-8") as f:
            yaml.dump(self.decision_config, f)

        # 3. Vocabulary File am RICHTIGEN Ort relativ zum test_vault
        self.vocab_path = os.path.join(self.test_vault, "01_User_Manual/01_edge_vocabulary.md")
        with open(self.vocab_path, "w", encoding="utf-8") as f:
            f.write("| System-Typ | Aliases |\n| :--- | :--- |\n| **caused_by** | ursache_ist |\n| **part_of** | teil_von |")

        # 4. MOCKING / RESETTING GLOBAL STATE
        # Zwinge get_settings, unsere Test-Pfade zurückzugeben
        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"
        
        # Patching get_settings in allen relevanten Modulen
        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()
        
        # Caches zwingend leeren
        app.routers.chat._DECISION_CONFIG_CACHE = None
        
        # Registry Singleton Reset & Force Init mit Test-Pfad
        EdgeRegistry._instance = None
        self.registry = EdgeRegistry(vault_root=self.test_vault)
        self.registry.unknown_log_path = os.path.join(self.test_root, "data/logs/unknown.jsonl")

    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 Pfad & Alias Resolution")
        # Prüfen ob die Datei gefunden wurde
        self.assertTrue(len(self.registry.valid_types) > 0, f"Registry leer! Root: {self.registry.vault_root}")
        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.retrieval.retriever_scoring.get_weights", return_value=(1.0, 1.0, 0.0)):
            # Stable (1.2)
            self.assertEqual(get_status_multiplier({"status": "stable"}), 1.2)
            # Draft (0.5)
            self.assertEqual(get_status_multiplier({"status": "draft"}), 0.5)
            
            # Scoring Formel Test: BaseScore * (1 + ConfigWeight + DynamicBoost)
            # BaseScore = 0.5 (sem) * 1.2 (stable) = 0.6
            # ConfigWeight = 1.0 (neutral) - 1.0 = 0.0
            # DynamicBoost = (1.0 * 0.5) = 0.5
            # Total = 0.6 * (1 + 0 + 0.5) = 0.9
            result = compute_wp22_score(0.5, {"status": "stable", "retriever_weight": 1.0}, edge_bonus_raw=0.5)
            self.assertAlmostEqual(result["total"], 0.9)
        print("✅ Scoring OK.")

    async def test_router_intent(self):
        print("\n🔵 TEST 3: Intent Classification")
        mock_llm = MagicMock()
        intent, _ = await _classify_intent("Warum ist das so?", mock_llm)
        self.assertEqual(intent, "CAUSAL")
        print("✅ Routing OK.")

    async def test_full_flow(self):
        print("\n🔵 TEST 4: End-to-End Pipeline & Dynamic Boosting")
        mock_llm = AsyncMock()
        mock_llm.prompts = {}
        mock_llm.generate_raw_response.return_value = "Test Antwort"
        
        mock_retriever = AsyncMock()
        # Fix note_id für Pydantic Validation
        mock_hit = QueryHit(
            node_id="c1", note_id="test_note_n1", semantic_score=0.8, edge_bonus=0.0,
            centrality_bonus=0.0, total_score=0.8, source={"text": "t"},
            payload={"status": "active", "type": "concept"}
        )
        mock_retriever.search.return_value.results = [mock_hit]
        
        req = ChatRequest(message="Warum ist das passiert?", top_k=1)
        resp = await chat_endpoint(req, llm=mock_llm, retriever=mock_retriever)
        
        # Verify Intent
        self.assertEqual(resp.intent, "CAUSAL")
        
        # Verify Boosts Reached Retriever
        called_req = mock_retriever.search.call_args[0][0]
        self.assertEqual(called_req.boost_edges.get("caused_by"), 3.0)
        print("✅ Full Flow & Boosting OK.")

if __name__ == '__main__':
    unittest.main()