mindnet/tests/test_smart_chunking_integration.py

# tests/test_smart_chunking_integration.py (Final für Stabilität)

import asyncio
import unittest
import os
import sys
from pathlib import Path
from typing import List, Dict
import time # Nur für debug/sleep

# --- 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 
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"
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.
"""

# Helper, um den Client im vorhandenen Loop zu schließen (Muss in der Klasse sein)
async def _async_close_analyzer(analyzer):
    await analyzer.close()

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
        cls._analyzer_instance = SemanticAnalyzer()
        # Stellt sicher, dass der Chunker diese Instanz verwenden kann.
        chunker._semantic_analyzer_instance = cls._analyzer_instance 

    @classmethod
    def tearDownClass(cls):
        """Schließt den httpx.AsyncClient nach allen Tests (Korrektur des Event Loop Fehlers)."""
        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())


    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 (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.")
        
        # --- 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
        # 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.")

        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()