diff --git a/tests/test_WP22_intelligence.py b/tests/test_WP22_intelligence.py new file mode 100644 index 0000000..f94d27c --- /dev/null +++ b/tests/test_WP22_intelligence.py @@ -0,0 +1,133 @@ +import unittest +import os +import shutil +import json +from unittest.mock import MagicMock, patch + +# Importiere die neuen Module +from app.services.edge_registry import EdgeRegistry +from app.core.retriever import _compute_total_score_v2, _get_status_multiplier + +# Wir mocken Teile, um DB-Abhängigkeit zu vermeiden +class TestWP22Intelligence(unittest.TestCase): + + def setUp(self): + # 1. Setup Dummy Vocabulary + self.test_vocab_path = "tests/fixtures/01_edge_vocabulary.md" + self.test_log_path = "tests/logs/unknown_edges.jsonl" + os.makedirs("tests/fixtures", exist_ok=True) + os.makedirs("tests/logs", exist_ok=True) + + with open(self.test_vocab_path, "w") as f: + f.write(""" +| **canonical** | Aliases | +| :--- | :--- | +| **caused_by** | ursache_ist, wegen | +| **next** | danach, folgt | + """) + + # Reset Registry Singleton for Test + EdgeRegistry._instance = None + self.registry = EdgeRegistry() + self.registry.vocab_path = self.test_vocab_path + self.registry.unknown_log_path = self.test_log_path + self.registry._load_vocabulary() + + def tearDown(self): + # Cleanup + if os.path.exists("tests/fixtures"): + shutil.rmtree("tests/fixtures") + if os.path.exists("tests/logs"): + shutil.rmtree("tests/logs") + + # --- TEIL A: EDGE REGISTRY --- + def test_registry_resolution(self): + print("\n--- Test A: Registry & Alias Resolution ---") + + # 1. Canonical Check + self.assertEqual(self.registry.resolve("caused_by"), "caused_by") + + # 2. Alias Check + resolved = self.registry.resolve("ursache_ist") + print(f"Resolving 'ursache_ist' -> '{resolved}'") + self.assertEqual(resolved, "caused_by") + + # 3. Unknown Check & Logging + unknown = self.registry.resolve("mystery_link") + print(f"Resolving 'mystery_link' -> '{unknown}' (sollte durchgereicht werden)") + self.assertEqual(unknown, "mystery_link") + + # Check Logfile + with open(self.test_log_path, "r") as f: + log_content = f.read() + self.assertIn("mystery_link", log_content) + print("✅ Unknown edge correctly logged.") + + # --- TEIL B: LIFECYCLE SCORING --- + def test_lifecycle_scoring(self): + print("\n--- Test B: Lifecycle Scoring Math ---") + + # Baseline: Semantic Score 0.9, keine Edges + base_sem = 0.9 + + payload_draft = {"status": "draft", "retriever_weight": 1.0} + payload_stable = {"status": "stable", "retriever_weight": 1.0} + + # Mock Settings + with patch("app.core.retriever._get_scoring_weights", return_value=(1.0, 0.5, 0.0)): + # Achtung: Hier rufen wir die Logik auf, die wir im Retriever implementiert haben + # Da wir die Funktion _compute_total_score_v2 im Chat-Prompt definiert haben, + # nutzen wir hier die Logik aus der _get_status_multiplier Helper Funktion + + mult_draft = _get_status_multiplier(payload_draft) + mult_stable = _get_status_multiplier(payload_stable) + + score_draft = base_sem * mult_draft + score_stable = base_sem * mult_stable + + print(f"Score Draft (0.8x): {score_draft:.2f}") + print(f"Score Stable (1.2x): {score_stable:.2f}") + + self.assertLess(score_draft, base_sem) + self.assertGreater(score_stable, base_sem) + print("✅ Stable notes scored higher than drafts.") + + # --- TEIL C: DYNAMIC EDGE BOOSTING --- + def test_dynamic_boosting(self): + print("\n--- Test C: Dynamic Edge Boosting ---") + + # Szenario: Wir simulieren, dass der Graph-Adapter einen Edge-Bonus von 1.0 berechnet hat + # Wir wollen prüfen, ob der Intent "WHY" diesen Bonus verstärkt. + + semantic_score = 0.5 + raw_edge_bonus = 1.0 # Stark vernetzt + + with patch("app.core.retriever._get_scoring_weights", return_value=(1.0, 1.0, 0.0)): + # Fall 1: Normale Suche (Kein Boost) + # Formel ca: (1.0 * 0.5) + (1.0 * 1.0) = 1.5 + from app.core.retriever import _compute_total_score + + score_normal, _, _ = _compute_total_score( + semantic_score, + {"status": "active"}, + edge_bonus=raw_edge_bonus, + dynamic_edge_boosts=None + ) + + # Fall 2: "WHY" Frage (Boost auf caused_by -> simuliert im Request) + boost_map = {"caused_by": 2.0} + score_boosted, _, _ = _compute_total_score( + semantic_score, + {"status": "active"}, + edge_bonus=raw_edge_bonus, + dynamic_edge_boosts=boost_map + ) + + print(f"Normal Score: {score_normal}") + print(f"Boosted Score: {score_boosted}") + + self.assertGreater(score_boosted, score_normal) + print("✅ Dynamic Boosting increased score successfully.") + +if __name__ == '__main__': + unittest.main() \ No newline at end of file