Lars/mindnet
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 21 Wiki Activity

modified: tests/test_smart_chunking_integration.py

Browse Source
This commit is contained in:
Lars 2025-12-12 09:18:36 +01:00
parent 94dbaafc72
commit 652d22e8e8
1 changed files with 9 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
tests/test_smart_chunking_integration.py
View File

@ -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 asyncio
import unittest 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. 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 --- # --- ENTFERNEN DER KOMPLEXEN TEARDOWN-HILFEN ---
# Führt die asynchrone Koroutine in einem temporären, dedizierten Loop aus. # Wir entfernen die fehleranfällige asynchrone Schließungslogik.
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): class TestSemanticChunking(unittest.TestCase):
@ -75,10 +57,11 @@ class TestSemanticChunking(unittest.TestCase):
@classmethod @classmethod
def tearDownClass(cls): def tearDownClass(cls):
"""Schließt den httpx.AsyncClient nach allen Tests (Löst Loop-Konflikt).""" """
if cls._analyzer_instance: PRAGMATISCHE LÖSUNG: Überspringe das explizite Aclose() im Teardown,
# Nutzt die temporäre Loop-Lösung um den Event Loop Konflikt zu vermeiden. Die GC wird die Verbindung schließen.
_teardown_sync_async_client(cls._analyzer_instance.close()) """
pass
def setUp(self): def setUp(self):
self.config = chunker.get_chunk_config(TEST_NOTE_TYPE) 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): 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. 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) --- # --- 1. Chunking (Asynchron) ---
@ -103,7 +85,7 @@ class TestSemanticChunking(unittest.TestCase):
print(f"\n--- LLM Chunker Output: {len(chunks)} Chunks ---") 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, self.assertTrue(len(chunks) > 1,
"Assertion B1 Fehler: LLM hat nicht zerlegt (Fallback aktiv). Prüfe LLM-Stabilität.") "Assertion B1 Fehler: LLM hat nicht zerlegt (Fallback aktiv). Prüfe LLM-Stabilität.")