507 lines
18 KiB
Python
507 lines
18 KiB
Python
"""Tests für ETLPipeline und MarkdownSource.
|
|
|
|
Validiert:
|
|
- MarkdownSource: Rekursives Scannen, Fehlerbehandlung, Frontmatter-Parsing
|
|
- ETLPipeline: Inkrementelle Verarbeitung, Fehlerresilienz, Index-Update
|
|
- IngestResult: Korrekte Zähler und Fehlerlisten
|
|
- KnowledgeStore.ingest(): Integration mit ETLPipeline
|
|
|
|
Requirements: 3.2, 3.6, 3.10, 3.12, 3.20, 3.21
|
|
"""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from monorepo.knowledge.etl import ETLPipeline, IngestError, IngestResult
|
|
from monorepo.knowledge.index import IndexEntry, YAMLIndex
|
|
from monorepo.knowledge.sources.markdown import MarkdownSource, SourceError
|
|
from monorepo.knowledge.store import KnowledgeStore
|
|
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 source_dir_with_artifacts(tmp_path: Path) -> Path:
|
|
"""Verzeichnis mit gültigen Markdown-Dateien mit YAML-Frontmatter."""
|
|
source_dir = tmp_path / "source"
|
|
source_dir.mkdir()
|
|
|
|
# Datei 1: gültiges Artefakt
|
|
(source_dir / "api-design.md").write_text(
|
|
"---\n"
|
|
"type: decision\n"
|
|
"title: API-Designprinzipien\n"
|
|
"tags: [api, rest]\n"
|
|
"---\n"
|
|
"\n# API-Designprinzipien\n\nREST-Konventionen für Services.\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
# Datei 2: gültiges Artefakt
|
|
(source_dir / "cloud-patterns.md").write_text(
|
|
"---\n"
|
|
"type: pattern\n"
|
|
"title: Cloud-Native Patterns\n"
|
|
"tags: [cloud, kubernetes]\n"
|
|
"---\n"
|
|
"\n# Cloud-Native Patterns\n\nBest Practices für cloud-native Apps.\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
return source_dir
|
|
|
|
|
|
@pytest.fixture
|
|
def source_dir_with_subdirs(tmp_path: Path) -> Path:
|
|
"""Verzeichnis mit verschachtelten Markdown-Dateien."""
|
|
source_dir = tmp_path / "nested_source"
|
|
source_dir.mkdir()
|
|
|
|
# Top-level Datei
|
|
(source_dir / "top-level.md").write_text(
|
|
"---\ntype: note\ntitle: Top Level\ntags: []\n---\n\nTop level content.\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
# Unterverzeichnis
|
|
sub_dir = source_dir / "decisions"
|
|
sub_dir.mkdir()
|
|
(sub_dir / "decision-one.md").write_text(
|
|
"---\ntype: decision\ntitle: Decision One\ntags: [arch]\n---\n\nDecision content.\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
return source_dir
|
|
|
|
|
|
@pytest.fixture
|
|
def source_dir_with_invalid(tmp_path: Path) -> Path:
|
|
"""Verzeichnis mit gemischten gültigen und ungültigen Dateien."""
|
|
source_dir = tmp_path / "mixed"
|
|
source_dir.mkdir()
|
|
|
|
# Gültige Datei
|
|
(source_dir / "valid.md").write_text(
|
|
"---\ntype: note\ntitle: Valid Note\ntags: [test]\n---\n\nValid content.\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
# Ungültige Datei (kein Frontmatter)
|
|
(source_dir / "no-frontmatter.md").write_text(
|
|
"# Just a Heading\n\nNo YAML frontmatter here.\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
# Ungültige Datei (leeres Frontmatter)
|
|
(source_dir / "empty-frontmatter.md").write_text(
|
|
"---\n---\n\nEmpty frontmatter.\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
return source_dir
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# MarkdownSource Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMarkdownSource:
|
|
"""Tests für MarkdownSource."""
|
|
|
|
def test_extract_finds_all_md_files(
|
|
self, source_dir_with_artifacts: Path
|
|
) -> None:
|
|
"""extract() findet alle .md-Dateien im Verzeichnis."""
|
|
source = MarkdownSource(directory=source_dir_with_artifacts)
|
|
artifacts = source.extract()
|
|
|
|
assert len(artifacts) == 2
|
|
titles = {a.metadata.title for a in artifacts}
|
|
assert "API-Designprinzipien" in titles
|
|
assert "Cloud-Native Patterns" in titles
|
|
|
|
def test_extract_recursive(self, source_dir_with_subdirs: Path) -> None:
|
|
"""extract() findet Dateien in Unterverzeichnissen."""
|
|
source = MarkdownSource(directory=source_dir_with_subdirs)
|
|
artifacts = source.extract()
|
|
|
|
assert len(artifacts) == 2
|
|
titles = {a.metadata.title for a in artifacts}
|
|
assert "Top Level" in titles
|
|
assert "Decision One" in titles
|
|
|
|
def test_extract_nonexistent_directory(self, tmp_path: Path) -> None:
|
|
"""extract() bei nicht-existierendem Verzeichnis gibt leere Liste + Fehler."""
|
|
source = MarkdownSource(directory=tmp_path / "nonexistent")
|
|
artifacts = source.extract()
|
|
|
|
assert artifacts == []
|
|
assert len(source.errors) == 1
|
|
assert source.errors[0].retry is True
|
|
|
|
def test_extract_handles_invalid_frontmatter_gracefully(
|
|
self, source_dir_with_invalid: Path
|
|
) -> None:
|
|
"""extract() überspringt Dateien ohne gültiges Frontmatter und loggt Fehler."""
|
|
source = MarkdownSource(directory=source_dir_with_invalid)
|
|
artifacts = source.extract()
|
|
|
|
# valid.md und empty-frontmatter.md parse successfully
|
|
# (empty-frontmatter has valid --- delimiters, just empty metadata)
|
|
# Only no-frontmatter.md is rejected (no --- at start)
|
|
assert len(artifacts) == 2
|
|
|
|
# Fehler für die ungültige Datei ohne Frontmatter
|
|
assert len(source.errors) == 1
|
|
assert any("no-frontmatter" in e.file_path for e in source.errors)
|
|
|
|
def test_extract_empty_directory(self, tmp_path: Path) -> None:
|
|
"""extract() bei leerem Verzeichnis gibt leere Liste ohne Fehler."""
|
|
empty_dir = tmp_path / "empty"
|
|
empty_dir.mkdir()
|
|
|
|
source = MarkdownSource(directory=empty_dir)
|
|
artifacts = source.extract()
|
|
|
|
assert artifacts == []
|
|
assert source.errors == []
|
|
|
|
def test_extract_file_instead_of_directory(self, tmp_path: Path) -> None:
|
|
"""extract() bei Datei statt Verzeichnis gibt Fehler."""
|
|
file_path = tmp_path / "not-a-dir.md"
|
|
file_path.write_text("content", encoding="utf-8")
|
|
|
|
source = MarkdownSource(directory=file_path)
|
|
artifacts = source.extract()
|
|
|
|
assert artifacts == []
|
|
assert len(source.errors) == 1
|
|
assert source.errors[0].retry is False
|
|
|
|
def test_extract_resets_errors(self, source_dir_with_invalid: Path) -> None:
|
|
"""extract() setzt vorherige Fehler zurück bei erneutem Aufruf."""
|
|
source = MarkdownSource(directory=source_dir_with_invalid)
|
|
source.extract()
|
|
first_errors = len(source.errors)
|
|
|
|
source.extract()
|
|
# Fehler sollten gleich sein (kein Akkumulieren)
|
|
assert len(source.errors) == first_errors
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ETLPipeline Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestETLPipeline:
|
|
"""Tests für ETLPipeline."""
|
|
|
|
def test_ingest_processes_all_artifacts(
|
|
self, tmp_path: Path, source_dir_with_artifacts: Path
|
|
) -> None:
|
|
"""ingest() verarbeitet alle gültigen Artefakte aus der Quelle."""
|
|
store_path = tmp_path / "store"
|
|
store_path.mkdir()
|
|
index = YAMLIndex(store_path / "_index.yaml")
|
|
index.load()
|
|
|
|
pipeline = ETLPipeline(index=index, store_path=store_path)
|
|
source = MarkdownSource(directory=source_dir_with_artifacts)
|
|
|
|
result = pipeline.ingest(source, "bahn")
|
|
|
|
assert result.processed == 2
|
|
assert result.updated == 2
|
|
assert result.skipped == 0
|
|
assert result.errors == []
|
|
|
|
def test_ingest_creates_index_entries(
|
|
self, tmp_path: Path, source_dir_with_artifacts: Path
|
|
) -> None:
|
|
"""ingest() erstellt Index-Einträge für alle verarbeiteten Artefakte."""
|
|
store_path = tmp_path / "store"
|
|
store_path.mkdir()
|
|
index = YAMLIndex(store_path / "_index.yaml")
|
|
index.load()
|
|
|
|
pipeline = ETLPipeline(index=index, store_path=store_path)
|
|
source = MarkdownSource(directory=source_dir_with_artifacts)
|
|
pipeline.ingest(source, "bahn")
|
|
|
|
# Index enthält Einträge
|
|
assert len(index.entries) == 2
|
|
api_entry = index.get_entry("bahn/decision/api-design")
|
|
assert api_entry is not None
|
|
assert api_entry.title == "API-Designprinzipien"
|
|
assert api_entry.scope == "bahn"
|
|
assert api_entry.type == "decision"
|
|
assert "api" in api_entry.tags
|
|
assert api_entry.content_hash.startswith("sha256:")
|
|
|
|
def test_ingest_incremental_skips_unchanged(
|
|
self, tmp_path: Path, source_dir_with_artifacts: Path
|
|
) -> None:
|
|
"""ingest() überspringt Artefakte mit unverändertem Content-Hash."""
|
|
store_path = tmp_path / "store"
|
|
store_path.mkdir()
|
|
index = YAMLIndex(store_path / "_index.yaml")
|
|
index.load()
|
|
|
|
pipeline = ETLPipeline(index=index, store_path=store_path)
|
|
source = MarkdownSource(directory=source_dir_with_artifacts)
|
|
|
|
# Erster Durchlauf: alle werden verarbeitet
|
|
result1 = pipeline.ingest(source, "bahn")
|
|
assert result1.updated == 2
|
|
|
|
# Zweiter Durchlauf: alle werden übersprungen (unverändert)
|
|
source2 = MarkdownSource(directory=source_dir_with_artifacts)
|
|
result2 = pipeline.ingest(source2, "bahn")
|
|
assert result2.processed == 2
|
|
assert result2.updated == 0
|
|
assert result2.skipped == 2
|
|
|
|
def test_ingest_incremental_detects_changes(
|
|
self, tmp_path: Path, source_dir_with_artifacts: Path
|
|
) -> None:
|
|
"""ingest() erkennt geänderte Inhalte über Content-Hash."""
|
|
store_path = tmp_path / "store"
|
|
store_path.mkdir()
|
|
index = YAMLIndex(store_path / "_index.yaml")
|
|
index.load()
|
|
|
|
pipeline = ETLPipeline(index=index, store_path=store_path)
|
|
source = MarkdownSource(directory=source_dir_with_artifacts)
|
|
pipeline.ingest(source, "bahn")
|
|
|
|
# Inhalt einer Datei ändern
|
|
api_file = source_dir_with_artifacts / "api-design.md"
|
|
api_file.write_text(
|
|
"---\n"
|
|
"type: decision\n"
|
|
"title: API-Designprinzipien\n"
|
|
"tags: [api, rest, updated]\n"
|
|
"---\n"
|
|
"\n# API-Designprinzipien V2\n\nAktualisierte Konventionen.\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
# Zweiter Durchlauf: nur die geänderte Datei wird aktualisiert
|
|
source2 = MarkdownSource(directory=source_dir_with_artifacts)
|
|
result2 = pipeline.ingest(source2, "bahn")
|
|
assert result2.updated == 1
|
|
assert result2.skipped == 1
|
|
|
|
def test_ingest_error_resilience(
|
|
self, tmp_path: Path, source_dir_with_invalid: Path
|
|
) -> None:
|
|
"""ingest() bewahrt erfolgreiche Artefakte bei Quellfehlern."""
|
|
store_path = tmp_path / "store"
|
|
store_path.mkdir()
|
|
index = YAMLIndex(store_path / "_index.yaml")
|
|
index.load()
|
|
|
|
pipeline = ETLPipeline(index=index, store_path=store_path)
|
|
source = MarkdownSource(directory=source_dir_with_invalid)
|
|
|
|
result = pipeline.ingest(source, "privat")
|
|
|
|
# valid.md and empty-frontmatter.md are parseable (2 artifacts)
|
|
# no-frontmatter.md is rejected by source
|
|
assert result.processed == 2
|
|
assert result.updated == 2
|
|
# Fehler für ungültige Dateien wurden protokolliert
|
|
assert len(result.errors) == 1
|
|
# Erfolgreiche Artefakte im Index
|
|
assert len(index.entries) == 2
|
|
|
|
def test_ingest_saves_index(
|
|
self, tmp_path: Path, source_dir_with_artifacts: Path
|
|
) -> None:
|
|
"""ingest() speichert den Index im selben Durchlauf."""
|
|
store_path = tmp_path / "store"
|
|
store_path.mkdir()
|
|
index_path = store_path / "_index.yaml"
|
|
index = YAMLIndex(index_path)
|
|
index.load()
|
|
|
|
pipeline = ETLPipeline(index=index, store_path=store_path)
|
|
source = MarkdownSource(directory=source_dir_with_artifacts)
|
|
pipeline.ingest(source, "bahn")
|
|
|
|
# Index-Datei wurde geschrieben
|
|
assert index_path.exists()
|
|
with open(index_path, encoding="utf-8") as f:
|
|
data = yaml.safe_load(f)
|
|
assert len(data["artifacts"]) == 2
|
|
assert data["version"] == "1.0"
|
|
assert "last_updated" in data
|
|
|
|
def test_ingest_sets_metadata(
|
|
self, tmp_path: Path, source_dir_with_artifacts: Path
|
|
) -> None:
|
|
"""ingest() setzt source_context und Datum in den Metadaten."""
|
|
store_path = tmp_path / "store"
|
|
store_path.mkdir()
|
|
index = YAMLIndex(store_path / "_index.yaml")
|
|
index.load()
|
|
|
|
pipeline = ETLPipeline(index=index, store_path=store_path)
|
|
source = MarkdownSource(directory=source_dir_with_artifacts)
|
|
pipeline.ingest(source, "bahn")
|
|
|
|
entry = index.get_entry("bahn/decision/api-design")
|
|
assert entry is not None
|
|
assert entry.scope == "bahn"
|
|
# Content-Hash ist gesetzt
|
|
assert entry.content_hash.startswith("sha256:")
|
|
|
|
def test_ingest_writes_artifact_files(
|
|
self, tmp_path: Path, source_dir_with_artifacts: Path
|
|
) -> None:
|
|
"""ingest() schreibt Artefakt-Dateien in die Scope-basierte Ordnerstruktur."""
|
|
store_path = tmp_path / "store"
|
|
store_path.mkdir()
|
|
index = YAMLIndex(store_path / "_index.yaml")
|
|
index.load()
|
|
|
|
pipeline = ETLPipeline(index=index, store_path=store_path)
|
|
source = MarkdownSource(directory=source_dir_with_artifacts)
|
|
pipeline.ingest(source, "bahn")
|
|
|
|
# Artefakt-Datei existiert in der Scope-Struktur
|
|
expected_path = store_path / "bahn" / "decision" / "api-design.md"
|
|
assert expected_path.exists()
|
|
content = expected_path.read_text(encoding="utf-8")
|
|
assert "---" in content
|
|
assert "API-Designprinzipien" in content
|
|
|
|
def test_ingest_nonexistent_source_directory(self, tmp_path: Path) -> None:
|
|
"""ingest() bei nicht-existierendem Quellverzeichnis gibt Fehler zurück."""
|
|
store_path = tmp_path / "store"
|
|
store_path.mkdir()
|
|
index = YAMLIndex(store_path / "_index.yaml")
|
|
index.load()
|
|
|
|
pipeline = ETLPipeline(index=index, store_path=store_path)
|
|
source = MarkdownSource(directory=tmp_path / "nonexistent")
|
|
|
|
result = pipeline.ingest(source, "bahn")
|
|
|
|
assert result.processed == 0
|
|
assert len(result.errors) >= 1
|
|
|
|
def test_ingest_empty_source(self, tmp_path: Path) -> None:
|
|
"""ingest() bei leerem Quellverzeichnis verarbeitet null Artefakte."""
|
|
store_path = tmp_path / "store"
|
|
store_path.mkdir()
|
|
empty_dir = tmp_path / "empty"
|
|
empty_dir.mkdir()
|
|
index = YAMLIndex(store_path / "_index.yaml")
|
|
index.load()
|
|
|
|
pipeline = ETLPipeline(index=index, store_path=store_path)
|
|
source = MarkdownSource(directory=empty_dir)
|
|
|
|
result = pipeline.ingest(source, "bahn")
|
|
|
|
assert result.processed == 0
|
|
assert result.updated == 0
|
|
assert result.skipped == 0
|
|
assert result.errors == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# KnowledgeStore.ingest() Integration Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestKnowledgeStoreIngest:
|
|
"""Tests für KnowledgeStore.ingest() Integration mit ETLPipeline."""
|
|
|
|
def test_ingest_via_store(
|
|
self, tmp_path: Path, scope_config: ScopeConfig, source_dir_with_artifacts: Path
|
|
) -> None:
|
|
"""KnowledgeStore.ingest() delegiert an ETLPipeline."""
|
|
store = KnowledgeStore(tmp_path, scope_config)
|
|
source = MarkdownSource(directory=source_dir_with_artifacts)
|
|
|
|
result = store.ingest(source, "bahn")
|
|
|
|
assert result.processed == 2
|
|
assert result.updated == 2
|
|
# Artefakte sind im Index
|
|
assert len(store.index.entries) == 2
|
|
|
|
def test_ingest_via_store_incremental(
|
|
self, tmp_path: Path, scope_config: ScopeConfig, source_dir_with_artifacts: Path
|
|
) -> None:
|
|
"""KnowledgeStore.ingest() unterstützt inkrementelle Verarbeitung."""
|
|
store = KnowledgeStore(tmp_path, scope_config)
|
|
source1 = MarkdownSource(directory=source_dir_with_artifacts)
|
|
store.ingest(source1, "bahn")
|
|
|
|
# Zweiter Durchlauf: keine Änderungen
|
|
source2 = MarkdownSource(directory=source_dir_with_artifacts)
|
|
result2 = store.ingest(source2, "bahn")
|
|
assert result2.skipped == 2
|
|
assert result2.updated == 0
|
|
|
|
def test_ingest_via_store_search_after_ingest(
|
|
self, tmp_path: Path, scope_config: ScopeConfig, source_dir_with_artifacts: Path
|
|
) -> None:
|
|
"""Nach ingest() sind Artefakte über search() auffindbar."""
|
|
store = KnowledgeStore(tmp_path, scope_config)
|
|
source = MarkdownSource(directory=source_dir_with_artifacts)
|
|
store.ingest(source, "bahn")
|
|
|
|
results = store.search("API", ["bahn"])
|
|
assert len(results) >= 1
|
|
assert any(r.entry.title == "API-Designprinzipien" for r in results)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# IngestResult Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestIngestResult:
|
|
"""Tests für IngestResult Dataclass."""
|
|
|
|
def test_success_with_processed(self) -> None:
|
|
"""success ist True wenn Artefakte verarbeitet wurden."""
|
|
result = IngestResult(processed=3, updated=2, skipped=1)
|
|
assert result.success is True
|
|
|
|
def test_success_empty_no_errors(self) -> None:
|
|
"""success ist True bei leerem Ergebnis ohne Fehler."""
|
|
result = IngestResult()
|
|
assert result.success is True
|
|
|
|
def test_defaults(self) -> None:
|
|
"""Standardwerte sind korrekt."""
|
|
result = IngestResult()
|
|
assert result.processed == 0
|
|
assert result.updated == 0
|
|
assert result.skipped == 0
|
|
assert result.errors == []
|