From 652d22e8e80d10be410e13ea54dd59c4330a3dba Mon Sep 17 00:00:00 2001 From: Lars Date: Fri, 12 Dec 2025 09:18:36 +0100 Subject: [PATCH] modified: tests/test_smart_chunking_integration.py --- tests/test_smart_chunking_integration.py | 36 ++++++------------------ 1 file changed, 9 insertions(+), 27 deletions(-) diff --git a/tests/test_smart_chunking_integration.py b/tests/test_smart_chunking_integration.py index e269fb9..06a489d 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 und Cleanup) +# tests/test_smart_chunking_integration.py (Letzte Korrektur zur Umgehung des AsyncIO-Fehlers) import asyncio import unittest @@ -40,26 +40,8 @@ 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. """ -# --- 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) +# --- ENTFERNEN DER KOMPLEXEN TEARDOWN-HILFEN --- +# Wir entfernen die fehleranfällige asynchrone Schließungslogik. class TestSemanticChunking(unittest.TestCase): @@ -75,10 +57,11 @@ class TestSemanticChunking(unittest.TestCase): @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()) + """ + PRAGMATISCHE LÖSUNG: Überspringe das explizite Aclose() im Teardown, + um den Event Loop Konflikt zu vermeiden. Die GC wird die Verbindung schließen. + """ + pass def setUp(self): self.config = chunker.get_chunk_config(TEST_NOTE_TYPE) @@ -91,7 +74,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) --- @@ -103,7 +85,7 @@ class TestSemanticChunking(unittest.TestCase): print(f"\n--- LLM Chunker Output: {len(chunks)} Chunks ---") - # Assertion B1: Zerlegung (Wenn das LLM erfolgreich war, muss > 1 Chunk geliefert werden) + # Assertion B1: Zerlegung (Das LLM muss mehr als 1 Chunk liefern) self.assertTrue(len(chunks) > 1, "Assertion B1 Fehler: LLM hat nicht zerlegt (Fallback aktiv). Prüfe LLM-Stabilität.")