133 lines
5.1 KiB
Python
133 lines
5.1 KiB
Python
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() |