diff --git a/tests/test_smart_chunking_integration.py b/tests/test_smart_chunking_integration.py index d1f67bc..6e2adf1 100644 --- a/tests/test_smart_chunking_integration.py +++ b/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.")