testscript

tests/test_smart_chunking_integration.py

@@ -1,4 +1,4 @@
-# tests/test_smart_chunking_integration.py (Final für Stabilität)
+# tests/test_smart_chunking_integration.py
 
 import asyncio
 import unittest
@@ -6,7 +6,8 @@ import os
 import sys
 from pathlib import Path
 from typing import List, Dict
-import time # Nur für debug/sleep
+import time
+import threading # Import für die Thread-basierte Schließung
 
 # --- PFAD-KORREKTUR ---
 ROOT_DIR = Path(__file__).resolve().parent.parent
@@ -16,8 +17,9 @@ sys.path.insert(0, str(ROOT_DIR))
 # Import der Kernkomponenten
 from app.core import chunker 
 from app.core import derive_edges 
+# Import der benötigten Klasse SemanticAnalyzer
 from app.services.semantic_analyzer import SemanticAnalyzer 
-from app.core.chunker import extract_frontmatter_from_text # Wichtig für Status Check
+
 
 # 1. Definieren der Test-Note (Simuliert eine journal.md Datei)
 TEST_NOTE_ID = "20251211-journal-sem-test"
@@ -41,36 +43,37 @@ Am Nachmittag gab es einen Konflikt bei der Karate-Trainer-Ausbildung. Ein Schü
 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.
 """
 
-# Helper, um den Client im vorhandenen Loop zu schließen (Muss in der Klasse sein)
-async def _async_close_analyzer(analyzer):
-    await analyzer.close()
+# --- HILFSFUNKTION FÜR DAS ASYNCHRONE SCHLIESSEN IN SYNCHRONER UMGEBUNG ---
+# Dies löst den Event Loop is closed Fehler in Python 3.12+
+def run_async_in_new_loop(coro):
+    """Führt eine Koroutine in einem neuen, temporären asyncio-Loop aus."""
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    try:
+        return loop.run_until_complete(coro)
+    finally:
+        loop.close()
+        asyncio.set_event_loop(asyncio.new_event_loop()) # Setzt den Loop zurück
+
 
 class TestSemanticChunking(unittest.TestCase):
     
-    # 2. Ressourcen-Management 
     _analyzer_instance = None
     
     @classmethod
     def setUpClass(cls):
-        """Initialisiert den SemanticAnalyzer einmalig und asynchron."""
-        # Da LLMService async ist, nutzen wir die Singleton-Instanz der Klasse
+        """Initialisiert den SemanticAnalyzer einmalig."""
+        # WICHTIG: Die Instanz muss erzeugt werden, bevor der erste async-Call läuft.
         cls._analyzer_instance = SemanticAnalyzer()
-        # Stellt sicher, dass der Chunker diese Instanz verwenden kann.
+        # 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 (Korrektur des Event Loop Fehlers)."""
+        """Schließt den httpx.AsyncClient nach allen Tests (Löst Loop-Konflikt)."""
         if cls._analyzer_instance:
-             # Wir nutzen run_coroutine_threadsafe, da der Loop geschlossen ist
-             loop = asyncio.get_event_loop()
-             if loop.is_running():
-                 # Wenn Loop noch läuft, planen wir den Aufruf
-                 asyncio.run_coroutine_threadsafe(cls._analyzer_instance.close(), loop)
-             else:
-                 # Andernfalls starten wir einen neuen temporären Loop nur zum Schließen
-                 asyncio.run(cls._analyzer_instance.close())
-
+             # Wir führen die async close Methode in einem neuen Loop aus
+             run_async_in_new_loop(cls._analyzer_instance.close())
 
     def setUp(self):
         self.config = chunker.get_chunk_config(TEST_NOTE_TYPE)
@@ -83,7 +86,6 @@ class TestSemanticChunking(unittest.TestCase):
     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) ---
@@ -96,6 +98,7 @@ class TestSemanticChunking(unittest.TestCase):
         print(f"\n--- LLM Chunker Output: {len(chunks)} Chunks ---")
         
         # Assertion B1: Zerlegung (Das LLM muss mehr als 1 Chunk liefern)
+        # Die LLM-Stabilität ist das Problem. Wir prüfen auf erfolgreiche Zerlegung.
         self.assertTrue(len(chunks) > 1, 
                         "Assertion B1 Fehler: LLM hat nicht zerlegt (Fallback aktiv). Prüfe LLM-Stabilität.")
         
@@ -144,11 +147,9 @@ class TestSemanticChunking(unittest.TestCase):
         ))
         
         # 2. Prüfen der Chunker-IDs
-        # Assertion C1: LLM-Chunking muss verhindert werden (darf NICHT mit '#sem' starten)
         self.assertFalse(chunks[0].id.startswith(TEST_NOTE_ID + '#sem'),
                          "Assertion C1 Fehler: LLM-Chunking wurde für den Status 'draft' nicht verhindert.")
         
-        # Assertion C2: Fallback-Strategie sollte by_heading sein (ID muss mit '#c' starten)
         self.assertTrue(chunks[0].id.startswith(TEST_NOTE_ID + '#c'),
                          "Assertion C2 Fehler: Fallback-Strategie 'by_heading' wurde nicht korrekt ausgeführt.")