# tests/test_smart_chunking_integration.py (Final für Stabilität und Cleanup) import asyncio import unittest import os import sys from pathlib import Path from typing import List, Dict # --- PFAD-KORREKTUR --- ROOT_DIR = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT_DIR)) # ---------------------- # Import der Kernkomponenten from app.core import chunker from app.core import derive_edges from app.services.semantic_analyzer import SemanticAnalyzer # 1. Definieren der Test-Note (Simuliert eine journal.md Datei) TEST_NOTE_ID = "20251211-journal-sem-test" TEST_NOTE_TYPE = "journal" TEST_MARKDOWN = """ --- id: 20251211-journal-sem-test title: Tägliches Log - Semantischer Test type: journal status: active created: 2025-12-11 tags: [test, daily-log] --- # Log-Eintrag 2025-12-11 Heute war ein guter Tag. Zuerst habe ich mit der R1 Meditation begonnen, um meinen Nordstern Fokus zu klären. Das Ritual [[leitbild-rituale-system]] hat mir geholfen, ruhig in den Tag zu starten. Ich habe gespürt, wie wichtig meine [[leitbild-werte#Integrität]] für meine Entscheidungen ist. Das ist das Fundament. Am Nachmittag gab es einen Konflikt bei der Karate-Trainer-Ausbildung. Ein Schüler war uneinsichtig. Ich habe die Situation nach [[leitbild-prinzipien#P4 Gerechtigkeit & Fairness]] behandelt und beide Seiten gehört (Steelman). Das war anstrengend, aber ich habe meine [[leitbild-rollen#Karate-Trainer]] Mission erfüllt. Die Konsequenz war klar und ruhig. Abends habe ich den wöchentlichen Load-Check mit meinem Partner gemacht. Das Paar-Ritual [[leitbild-rituale-system#R5]] hilft, das Ziel [[leitbild-ziele-portfolio#Nordstern Partner]] aktiv zu verfolgen. Es ist der operative Rhythmus für uns beide. """ # --- HILFSFUNKTION FÜR DAS ASYNCHRONE SCHLIESSEN --- # Führt die asynchrone Koroutine in einem temporären, dedizierten Loop aus. def _teardown_sync_async_client(coro): """Führt eine async Koroutine in einem eigenen, temporären Loop aus.""" try: loop = asyncio.get_running_loop() except RuntimeError: loop = asyncio.new_event_loop() if loop.is_running(): # Wenn der Loop bereits läuft (z.B. durch andere async-Tasks im Hintergrund), # nutzen wir run_coroutine_threadsafe. future = asyncio.run_coroutine_threadsafe(coro, loop) try: return future.result(timeout=5) except Exception: future.cancel() else: # Führe im aktuellen Thread aus, wenn kein Loop läuft (typischer teardown-Fall) return loop.run_until_complete(coro) class TestSemanticChunking(unittest.TestCase): _analyzer_instance = None @classmethod def setUpClass(cls): """Initialisiert den SemanticAnalyzer einmalig.""" cls._analyzer_instance = SemanticAnalyzer() # Stellt sicher, dass der Chunker diese Singleton-Instanz verwendet chunker._semantic_analyzer_instance = cls._analyzer_instance @classmethod def tearDownClass(cls): """Schließt den httpx.AsyncClient nach allen Tests (Löst Loop-Konflikt).""" if cls._analyzer_instance: # Nutzt die temporäre Loop-Lösung _teardown_sync_async_client(cls._analyzer_instance.close()) def setUp(self): self.config = chunker.get_chunk_config(TEST_NOTE_TYPE) def test_a_strategy_selection(self): """Prüft, ob die Strategie 'semantic_llm' für den Typ 'journal' gewählt wird.""" self.assertEqual(self.config['strategy'], 'semantic_llm', "Fehler: 'journal' sollte die Strategie 'semantic_llm' nutzen.") def test_b_llm_chunking_and_injection(self): """ Prüft den gesamten End-to-End-Flow: 1. LLM-Chunking, 2. Kanten-Injektion, 3. Kanten-Erkennung. (Diese Tests setzen voraus, dass das LLM JSON liefert) """ # --- 1. Chunking (Asynchron) --- chunks = asyncio.run(chunker.assemble_chunks( note_id=TEST_NOTE_ID, md_text=TEST_MARKDOWN, note_type=TEST_NOTE_TYPE )) print(f"\n--- LLM Chunker Output: {len(chunks)} Chunks ---") # Assertion B1: Zerlegung (Wenn das LLM erfolgreich war, muss > 1 Chunk geliefert werden) self.assertTrue(len(chunks) > 1, "Assertion B1 Fehler: LLM hat nicht zerlegt (Fallback aktiv). Prüfe LLM-Stabilität.") # --- 2. Injektion prüfen --- chunk_1_text = chunks[0].text self.assertIn("[[rel:", chunk_1_text, "Assertion B2 Fehler: Der Chunk-Text muss die injizierte [[rel: Kante enthalten.") # --- 3. Kanten-Derivation (Synchron) --- edges = derive_edges.build_edges_for_note( note_id=TEST_NOTE_ID, chunks=[c.__dict__ for c in chunks] ) print(f"--- Edge Derivation Output: {len(edges)} Kanten ---") # Assertion B3: Mindestens 3 LLM-Kanten (inline:rel) llm_generated_edges = [ e for e in edges if e.get('rule_id') == 'inline:rel' and e.get('source_id').startswith(TEST_NOTE_ID + '#sem') ] self.assertTrue(len(llm_generated_edges) >= 3, "Assertion B3 Fehler: Es wurden weniger als 3 semantische Kanten gefunden.") # Assertion B4: Check für die Matrix-Logik / Werte-Kante has_matrix_kante = any( e['target_id'].startswith('leitbild-werte') and e['kind'] in ['based_on', 'derived_from'] for e in llm_generated_edges ) self.assertTrue(has_matrix_kante, "Assertion B4 Fehler: Die Matrix-Logik / Werte-Kante wurde nicht erkannt.") print("\n✅ Integrationstest für Semantic Chunking erfolgreich.") def test_c_draft_status_prevention(self): """Prüft, ob 'draft' Status semantic_llm auf by_heading überschreibt.""" DRAFT_MARKDOWN = TEST_MARKDOWN.replace("status: active", "status: draft") # 1. Chunking mit Draft Status chunks = asyncio.run(chunker.assemble_chunks( note_id=TEST_NOTE_ID, md_text=DRAFT_MARKDOWN, note_type=TEST_NOTE_TYPE )) # 2. Prüfen der Chunker-IDs self.assertFalse(chunks[0].id.startswith(TEST_NOTE_ID + '#sem'), "Assertion C1 Fehler: LLM-Chunking wurde für den Status 'draft' nicht verhindert.") self.assertTrue(chunks[0].id.startswith(TEST_NOTE_ID + '#c'), "Assertion C2 Fehler: Fallback-Strategie 'by_heading' wurde nicht korrekt ausgeführt.") print(f"\n✅ Prevention Test: Draft-Status hat LLM-Chunking verhindert (Fallback ID: {chunks[0].id}).") if __name__ == '__main__': print("Starte den Semantic Chunking Integrationstest.") unittest.main()