Files
2026-06-30 20:37:40 +02:00

557 lines
21 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Tests für Volltextsuche und Scope-basierte Filterung im KnowledgeStore.
Validiert Task 6.4:
- SearchResult Dataclass mit entry, relevance_score, partial_results
- Relevanz-Scoring: title > tags > summary > id
- Sortierung nach Relevanz (höchster Score zuerst)
- Scope-Filterung: Nur Ergebnisse aus autorisierten Scopes
- Keine Offenlegung nicht-autorisierter Artefakte
- Timeout-Handling: Abbruch nach 5 Sekunden mit Teilergebnis-Warnung
Requirements: 3.3, 3.7, 3.9
"""
from __future__ import annotations
import time
from pathlib import Path
from unittest.mock import patch
import pytest
from monorepo.knowledge.index import IndexEntry, YAMLIndex
from monorepo.knowledge.store import (
SEARCH_TIMEOUT_SECONDS,
KnowledgeStore,
SearchResult,
_compute_relevance,
)
from monorepo.models import ScopeConfig
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def scope_config() -> ScopeConfig:
"""Beispiel-ScopeConfig."""
return ScopeConfig(scopes={
"privat": {"scope": "privat", "paths": ["privat/"]},
"dhive": {"scope": "dhive", "paths": ["dhive/"]},
"bahn": {"scope": "bahn", "paths": ["bahn/"]},
"shared": {"scope": "shared", "paths": ["shared/"]},
})
@pytest.fixture
def sample_entries() -> list[IndexEntry]:
"""Vielfältige Einträge zum Testen der Relevanz-Sortierung."""
return [
IndexEntry(
id="bahn/decisions/api-design",
title="API-Designprinzipien für Microservices",
type="decision",
tags=["api", "microservices", "architecture"],
scope="bahn",
summary="REST-API-Konventionen für DB-InfraGO-Dienste",
path="bahn/decisions/api-design.md",
content_hash="sha256:abc123",
links=[],
),
IndexEntry(
id="privat/notes/rest-patterns",
title="REST-Patterns Notizen",
type="note",
tags=["api", "rest", "patterns"],
scope="privat",
summary="Sammlung bewährter REST-Patterns",
path="privat/notes/rest-patterns.md",
content_hash="sha256:def456",
links=[],
),
IndexEntry(
id="dhive/reference/cloud-native",
title="Cloud-Native Architektur Guide",
type="reference",
tags=["cloud", "kubernetes", "architecture"],
scope="dhive",
summary="Leitfaden für cloud-native Anwendungen bei dhive",
path="dhive/reference/cloud-native.md",
content_hash="sha256:ghi789",
links=[],
),
IndexEntry(
id="shared/patterns/api-versioning",
title="API Versioning",
type="pattern",
tags=["api", "versioning", "best-practices"],
scope="shared",
summary="Best Practices für API-Versionierung",
path="shared/patterns/api-versioning.md",
content_hash="sha256:jkl012",
links=[],
),
IndexEntry(
id="bahn/notes/kubernetes-deployment",
title="Kubernetes Deployment Notizen",
type="note",
tags=["kubernetes", "deployment", "infrastructure"],
scope="bahn",
summary="Notizen zum Kubernetes-Deployment bei DB-InfraGO",
path="bahn/notes/kubernetes-deployment.md",
content_hash="sha256:mno345",
links=[],
),
]
@pytest.fixture
def store_with_entries(
tmp_path: Path, scope_config: ScopeConfig, sample_entries: list[IndexEntry]
) -> KnowledgeStore:
"""KnowledgeStore mit vorgeladenen Beispiel-Einträgen."""
store = KnowledgeStore(tmp_path, scope_config)
for entry in sample_entries:
store.index.update_entry(entry)
return store
# ---------------------------------------------------------------------------
# SearchResult Dataclass Tests
# ---------------------------------------------------------------------------
class TestSearchResult:
"""Tests für die SearchResult Dataclass."""
def test_default_values(self) -> None:
"""SearchResult hat korrekte Standardwerte."""
entry = IndexEntry(id="test", title="Test", type="note", scope="privat")
result = SearchResult(entry=entry)
assert result.entry is entry
assert result.relevance_score == 0.0
assert result.partial_results is False
def test_custom_values(self) -> None:
"""SearchResult akzeptiert benutzerdefinierte Werte."""
entry = IndexEntry(id="test", title="Test", type="note", scope="privat")
result = SearchResult(entry=entry, relevance_score=0.85, partial_results=True)
assert result.relevance_score == 0.85
assert result.partial_results is True
def test_relevance_score_range(self) -> None:
"""relevance_score liegt zwischen 0.0 und 1.0."""
entry = IndexEntry(id="test", title="Test", type="note", scope="privat")
# Grenzwerte
result_zero = SearchResult(entry=entry, relevance_score=0.0)
result_one = SearchResult(entry=entry, relevance_score=1.0)
assert result_zero.relevance_score == 0.0
assert result_one.relevance_score == 1.0
# ---------------------------------------------------------------------------
# Relevanz-Scoring Tests
# ---------------------------------------------------------------------------
class TestRelevanceScoring:
"""Tests für die Relevanz-Score-Berechnung."""
def test_title_match_highest_score(self) -> None:
"""Title-Match ergibt höchsten Score (>= 0.8)."""
entry = IndexEntry(
id="test/doc",
title="API Design",
type="note",
tags=["documentation"],
scope="bahn",
summary="Allgemeine Informationen",
)
score = _compute_relevance("api design", entry)
assert score >= 0.8
def test_exact_title_match_maximum_score(self) -> None:
"""Exakter Title-Match ergibt Score 1.0."""
entry = IndexEntry(
id="test/doc",
title="API Design",
type="note",
tags=["misc"],
scope="bahn",
summary="Irgendwas",
)
score = _compute_relevance("api design", entry)
assert score == 1.0
def test_tag_match_medium_score(self) -> None:
"""Tag-Match ergibt mittleren Score (0.6 - 0.79)."""
entry = IndexEntry(
id="test/doc",
title="Allgemeines Dokument",
type="note",
tags=["kubernetes", "deployment"],
scope="bahn",
summary="Nicht relevanter Text",
)
score = _compute_relevance("kubernetes", entry)
assert 0.6 <= score <= 0.79
def test_summary_match_lower_score(self) -> None:
"""Summary-Match ergibt niedrigeren Score (0.4 - 0.59)."""
entry = IndexEntry(
id="test/doc",
title="Allgemeines Dokument",
type="note",
tags=["misc"],
scope="bahn",
summary="Informationen über Kubernetes-Cluster",
)
score = _compute_relevance("kubernetes", entry)
assert 0.4 <= score <= 0.59
def test_id_match_lowest_score(self) -> None:
"""ID-Match ergibt niedrigsten Score (0.2 - 0.39)."""
entry = IndexEntry(
id="bahn/notes/kubernetes-setup",
title="Allgemeines Dokument",
type="note",
tags=["misc"],
scope="bahn",
summary="Nicht relevanter Text",
)
score = _compute_relevance("kubernetes", entry)
assert 0.2 <= score <= 0.39
def test_no_match_zero_score(self) -> None:
"""Kein Match ergibt Score 0.0."""
entry = IndexEntry(
id="test/doc",
title="Allgemeines Dokument",
type="note",
tags=["misc"],
scope="bahn",
summary="Nicht relevanter Text",
)
score = _compute_relevance("zyxwv", entry)
assert score == 0.0
def test_title_beats_tags(self) -> None:
"""Title-Match hat Vorrang vor Tag-Match."""
entry_title = IndexEntry(
id="test/a", title="Kubernetes", type="note",
tags=["misc"], scope="bahn", summary="Anderes",
)
entry_tag = IndexEntry(
id="test/b", title="Anderes Dokument", type="note",
tags=["kubernetes"], scope="bahn", summary="Anderes",
)
score_title = _compute_relevance("kubernetes", entry_title)
score_tag = _compute_relevance("kubernetes", entry_tag)
assert score_title > score_tag
def test_tags_beat_summary(self) -> None:
"""Tag-Match hat Vorrang vor Summary-Match."""
entry_tag = IndexEntry(
id="test/a", title="Anderes", type="note",
tags=["kubernetes"], scope="bahn", summary="Anderer Text",
)
entry_summary = IndexEntry(
id="test/b", title="Anderes", type="note",
tags=["misc"], scope="bahn", summary="Kubernetes ist wichtig",
)
score_tag = _compute_relevance("kubernetes", entry_tag)
score_summary = _compute_relevance("kubernetes", entry_summary)
assert score_tag > score_summary
def test_summary_beats_id(self) -> None:
"""Summary-Match hat Vorrang vor ID-Match."""
entry_summary = IndexEntry(
id="test/other", title="Anderes", type="note",
tags=["misc"], scope="bahn", summary="Kubernetes Deployment Guide",
)
entry_id = IndexEntry(
id="bahn/kubernetes/setup", title="Anderes", type="note",
tags=["misc"], scope="bahn", summary="Anderer Text",
)
score_summary = _compute_relevance("kubernetes", entry_summary)
score_id = _compute_relevance("kubernetes", entry_id)
assert score_summary > score_id
def test_case_insensitive(self) -> None:
"""Relevanz-Berechnung ist case-insensitive bezüglich Entry-Feldern.
Die Funktion erwartet query_lower als bereits lowercase Input.
Die Entry-Felder werden intern lowercase verglichen.
"""
entry_upper = IndexEntry(
id="test/doc", title="API DESIGN", type="note",
tags=[], scope="bahn", summary="",
)
entry_lower = IndexEntry(
id="test/doc", title="api design", type="note",
tags=[], scope="bahn", summary="",
)
# Gleicher Query (lowercase) gegen verschiedene Entry-Schreibweisen
score_upper_entry = _compute_relevance("api design", entry_upper)
score_lower_entry = _compute_relevance("api design", entry_lower)
assert score_upper_entry == score_lower_entry
# ---------------------------------------------------------------------------
# KnowledgeStore.search() Tests
# ---------------------------------------------------------------------------
class TestKnowledgeStoreSearch:
"""Tests für KnowledgeStore.search() mit Relevanz-Sortierung."""
def test_returns_search_results(self, store_with_entries: KnowledgeStore) -> None:
"""search() liefert SearchResult-Objekte."""
results = store_with_entries.search("api", ["bahn", "privat", "shared"])
assert len(results) > 0
for r in results:
assert isinstance(r, SearchResult)
assert isinstance(r.entry, IndexEntry)
assert isinstance(r.relevance_score, float)
assert isinstance(r.partial_results, bool)
def test_sorted_by_relevance_descending(self, store_with_entries: KnowledgeStore) -> None:
"""Ergebnisse sind nach Relevanz absteigend sortiert."""
results = store_with_entries.search("api", ["bahn", "privat", "shared", "dhive"])
scores = [r.relevance_score for r in results]
assert scores == sorted(scores, reverse=True)
def test_title_match_ranked_higher(self, store_with_entries: KnowledgeStore) -> None:
"""Einträge mit Title-Match werden höher gerankt als Tag-Matches."""
results = store_with_entries.search(
"api versioning", ["bahn", "privat", "shared", "dhive"]
)
# "API Versioning" hat exakten Title-Match
if results:
assert results[0].entry.id == "shared/patterns/api-versioning"
def test_empty_query_returns_empty(self, store_with_entries: KnowledgeStore) -> None:
"""Leerer Query liefert leere Ergebnisliste."""
results = store_with_entries.search("", ["bahn", "privat", "shared", "dhive"])
assert results == []
def test_no_match_returns_empty(self, store_with_entries: KnowledgeStore) -> None:
"""Query ohne Treffer liefert leere Liste."""
results = store_with_entries.search("zyxwvutsrq", ["bahn", "privat", "shared", "dhive"])
assert results == []
def test_relevance_scores_within_range(self, store_with_entries: KnowledgeStore) -> None:
"""Alle Relevanz-Scores liegen zwischen 0.0 und 1.0."""
results = store_with_entries.search("api", ["bahn", "privat", "shared", "dhive"])
for r in results:
assert 0.0 <= r.relevance_score <= 1.0
# ---------------------------------------------------------------------------
# Scope-Filterung Tests
# ---------------------------------------------------------------------------
class TestSearchScopeFiltering:
"""Tests für Scope-basierte Filterung in der Suche."""
def test_only_authorized_scopes_returned(self, store_with_entries: KnowledgeStore) -> None:
"""Nur Ergebnisse aus autorisierten Scopes werden zurückgegeben."""
results = store_with_entries.search("api", ["bahn"])
for r in results:
assert r.entry.scope == "bahn"
def test_unauthorized_scope_not_in_results(
self, store_with_entries: KnowledgeStore
) -> None:
"""Artefakte aus nicht-autorisierten Scopes erscheinen nicht."""
# dhive-Scope nicht autorisiert, aber "cloud" existiert dort
results = store_with_entries.search("cloud", ["bahn", "privat"])
for r in results:
assert r.entry.scope != "dhive"
def test_existence_not_revealed(self, store_with_entries: KnowledgeStore) -> None:
"""Existenz nicht-autorisierter Artefakte wird nicht offengelegt.
Requirements: 3.9 unberechtigte Artefakte aus Suchergebnissen
ausschließen, ohne deren Existenz offenzulegen.
"""
# "cloud-native" existiert im dhive-Scope
results_restricted = store_with_entries.search("cloud-native", ["bahn", "privat"])
results_all = store_with_entries.search("cloud-native", ["bahn", "privat", "dhive"])
# Ohne dhive-Berechtigung: Kein Hinweis auf Existenz
assert len(results_restricted) == 0
# Mit dhive-Berechtigung: Ergebnis vorhanden
assert len(results_all) >= 1
def test_empty_scopes_returns_empty(self, store_with_entries: KnowledgeStore) -> None:
"""Leere Scope-Liste liefert keine Ergebnisse."""
results = store_with_entries.search("api", [])
assert results == []
def test_multiple_scopes(self, store_with_entries: KnowledgeStore) -> None:
"""Suche über mehrere Scopes liefert kombinierte Ergebnisse."""
results = store_with_entries.search("api", ["bahn", "shared"])
scopes_found = {r.entry.scope for r in results}
# Beide Scopes sollten vertreten sein
assert "bahn" in scopes_found or "shared" in scopes_found
# ---------------------------------------------------------------------------
# Timeout-Handling Tests
# ---------------------------------------------------------------------------
class TestSearchTimeout:
"""Tests für das Timeout-Handling der Suche."""
def test_default_timeout_is_5_seconds(self) -> None:
"""Standard-Timeout ist 5 Sekunden."""
assert SEARCH_TIMEOUT_SECONDS == 5.0
def test_no_timeout_under_normal_conditions(
self, store_with_entries: KnowledgeStore
) -> None:
"""Unter normalen Bedingungen kein Timeout (partial_results=False)."""
results = store_with_entries.search("api", ["bahn", "privat", "shared", "dhive"])
for r in results:
assert r.partial_results is False
def test_timeout_marks_partial_results(
self, tmp_path: Path, scope_config: ScopeConfig
) -> None:
"""Bei Timeout werden alle Ergebnisse als partial markiert."""
store = KnowledgeStore(tmp_path, scope_config)
# Viele Einträge erstellen
for i in range(100):
entry = IndexEntry(
id=f"bahn/test/entry-{i}",
title=f"Testartikel {i} über API",
type="note",
tags=["api", "test"],
scope="bahn",
summary=f"Zusammenfassung für Entry {i}",
path=f"bahn/test/entry-{i}.md",
content_hash=f"sha256:hash{i}",
)
store.index.update_entry(entry)
# Timeout auf 0 Sekunden simulieren
results = store.search("api", ["bahn"], timeout=0.0)
# Entweder keine Ergebnisse (sofort Timeout) oder alle partial
if results:
for r in results:
assert r.partial_results is True
def test_custom_timeout_parameter(
self, tmp_path: Path, scope_config: ScopeConfig
) -> None:
"""Benutzerdefinierter Timeout wird respektiert."""
store = KnowledgeStore(tmp_path, scope_config)
entry = IndexEntry(
id="bahn/test/entry",
title="API Test",
type="note",
tags=["api"],
scope="bahn",
summary="Test",
path="bahn/test/entry.md",
content_hash="sha256:test",
)
store.index.update_entry(entry)
# Großzügiger Timeout: Normale Ergebnisse
results = store.search("api", ["bahn"], timeout=10.0)
assert len(results) == 1
assert results[0].partial_results is False
def test_timeout_preserves_found_results(
self, tmp_path: Path, scope_config: ScopeConfig
) -> None:
"""Bei Timeout werden bereits gefundene Teilergebnisse beibehalten."""
store = KnowledgeStore(tmp_path, scope_config)
# Viele Einträge erstellen
for i in range(1000):
entry = IndexEntry(
id=f"bahn/test/entry-{i}",
title=f"Testartikel {i} über API",
type="note",
tags=["api", "test"],
scope="bahn",
summary=f"Zusammenfassung für Entry {i}",
path=f"bahn/test/entry-{i}.md",
content_hash=f"sha256:hash{i}",
)
store.index.update_entry(entry)
# Minimaler Timeout: Sollte Teilergebnisse liefern
# Da die Iteration auf einer modernen Maschine schnell ist,
# mocken wir time.monotonic für einen kontrollierten Test
call_count = 0
real_monotonic = time.monotonic
def fake_monotonic() -> float:
nonlocal call_count
call_count += 1
# Nach 5 Aufrufen "Timeout" simulieren
if call_count > 5:
return real_monotonic() + 10.0
return real_monotonic()
with patch("monorepo.knowledge.store.time.monotonic", side_effect=fake_monotonic):
results = store.search("api", ["bahn"], timeout=5.0)
# Sollte Teilergebnisse haben (nicht alle 1000)
if results:
assert all(r.partial_results is True for r in results)
assert len(results) < 1000
# ---------------------------------------------------------------------------
# Integration Tests
# ---------------------------------------------------------------------------
class TestSearchIntegration:
"""Integrationstests für die Volltextsuche."""
def test_search_finds_by_title(self, store_with_entries: KnowledgeStore) -> None:
"""Findet Einträge über Titel-Suche."""
results = store_with_entries.search(
"Cloud-Native Architektur", ["dhive"]
)
assert len(results) >= 1
assert results[0].entry.id == "dhive/reference/cloud-native"
assert results[0].relevance_score >= 0.8 # Title match
def test_search_finds_by_tag(self, store_with_entries: KnowledgeStore) -> None:
"""Findet Einträge über Tag-Suche."""
results = store_with_entries.search("deployment", ["bahn"])
assert len(results) >= 1
assert any(r.entry.id == "bahn/notes/kubernetes-deployment" for r in results)
def test_search_finds_by_summary(self, store_with_entries: KnowledgeStore) -> None:
"""Findet Einträge über Summary-Suche."""
results = store_with_entries.search("bewährter", ["privat"])
assert len(results) == 1
assert results[0].entry.id == "privat/notes/rest-patterns"
def test_case_insensitive_search(self, store_with_entries: KnowledgeStore) -> None:
"""Suche ist case-insensitive."""
results_lower = store_with_entries.search("kubernetes", ["bahn", "dhive"])
results_upper = store_with_entries.search("KUBERNETES", ["bahn", "dhive"])
assert len(results_lower) == len(results_upper)
def test_search_with_all_scopes(self, store_with_entries: KnowledgeStore) -> None:
"""Suche über alle Scopes liefert alle Treffer."""
all_scopes = ["bahn", "privat", "dhive", "shared"]
results = store_with_entries.search("api", all_scopes)
# api kommt in mehreren Einträgen vor (title, tags)
assert len(results) >= 3