320 lines
11 KiB
Python
320 lines
11 KiB
Python
"""Unit-Tests für monorepo.knowledge.ingestion.config.
|
||
|
||
Testet SourceConfig, PipelineSettings, Env-Var-Auflösung,
|
||
Validierung und sources.yaml-Parsing.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from pathlib import Path
|
||
from textwrap import dedent
|
||
|
||
import pytest
|
||
|
||
from monorepo.knowledge.ingestion.config import (
|
||
ConfigValidationError,
|
||
EnrichmentSettings,
|
||
GitSettings,
|
||
OCRSettings,
|
||
PipelineSettings,
|
||
SourceConfig,
|
||
load_sources_config,
|
||
resolve_env_vars,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# SourceConfig Dataclass
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestSourceConfig:
|
||
"""Tests für die SourceConfig-Dataclass."""
|
||
|
||
def test_required_fields(self) -> None:
|
||
cfg = SourceConfig(type="confluence", name="Team Wiki")
|
||
assert cfg.type == "confluence"
|
||
assert cfg.name == "Team Wiki"
|
||
|
||
def test_defaults(self) -> None:
|
||
cfg = SourceConfig(type="jira", name="Sprint Issues")
|
||
assert cfg.params == {}
|
||
assert cfg.target_folder == ""
|
||
assert cfg.sync_frequency == "manual"
|
||
assert cfg.enabled is True
|
||
|
||
def test_full_config(self) -> None:
|
||
cfg = SourceConfig(
|
||
type="confluence",
|
||
name="DB Confluence",
|
||
params={"base_url": "https://confluence.bahn.de", "spaces": ["ACV2"]},
|
||
target_folder="references",
|
||
sync_frequency="daily",
|
||
enabled=True,
|
||
)
|
||
assert cfg.params["base_url"] == "https://confluence.bahn.de"
|
||
assert cfg.target_folder == "references"
|
||
assert cfg.sync_frequency == "daily"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# PipelineSettings Dataclass
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestPipelineSettings:
|
||
"""Tests für PipelineSettings und Unter-Dataclasses."""
|
||
|
||
def test_defaults(self) -> None:
|
||
settings = PipelineSettings()
|
||
assert settings.create_tasks is True
|
||
assert settings.enrichment.provider == "kiro"
|
||
assert settings.enrichment.confidence_threshold == 0.7
|
||
assert settings.ocr.provider == "kiro"
|
||
assert settings.ocr.languages == "deu+eng"
|
||
assert settings.git.auto_commit is True
|
||
|
||
def test_custom_settings(self) -> None:
|
||
settings = PipelineSettings(
|
||
create_tasks=False,
|
||
enrichment=EnrichmentSettings(provider="litellm", model="gpt-4"),
|
||
ocr=OCRSettings(provider="tesseract", languages="deu"),
|
||
git=GitSettings(auto_commit=False),
|
||
)
|
||
assert settings.create_tasks is False
|
||
assert settings.enrichment.provider == "litellm"
|
||
assert settings.enrichment.model == "gpt-4"
|
||
assert settings.ocr.provider == "tesseract"
|
||
assert settings.git.auto_commit is False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# resolve_env_vars
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestResolveEnvVars:
|
||
"""Tests für die Umgebungsvariablen-Auflösung."""
|
||
|
||
def test_no_vars(self) -> None:
|
||
assert resolve_env_vars("hello world") == "hello world"
|
||
|
||
def test_single_var(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
monkeypatch.setenv("MY_TOKEN", "secret123")
|
||
assert resolve_env_vars("${MY_TOKEN}") == "secret123"
|
||
|
||
def test_var_in_string(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
monkeypatch.setenv("HOST", "example.com")
|
||
assert resolve_env_vars("https://${HOST}/api") == "https://example.com/api"
|
||
|
||
def test_multiple_vars(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
monkeypatch.setenv("USER", "admin")
|
||
monkeypatch.setenv("PASS", "secret")
|
||
result = resolve_env_vars("${USER}:${PASS}")
|
||
assert result == "admin:secret"
|
||
|
||
def test_unset_var_raises(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
monkeypatch.delenv("NONEXISTENT_VAR_XYZ", raising=False)
|
||
with pytest.raises(ConfigValidationError) as exc_info:
|
||
resolve_env_vars("${NONEXISTENT_VAR_XYZ}")
|
||
assert "NONEXISTENT_VAR_XYZ" in str(exc_info.value)
|
||
|
||
def test_empty_string(self) -> None:
|
||
assert resolve_env_vars("") == ""
|
||
|
||
def test_var_with_empty_value(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
monkeypatch.setenv("EMPTY_VAR", "")
|
||
assert resolve_env_vars("prefix_${EMPTY_VAR}_suffix") == "prefix__suffix"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# load_sources_config – Erfolgsfälle
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestLoadSourcesConfig:
|
||
"""Tests für load_sources_config."""
|
||
|
||
def test_basic_config(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
monkeypatch.setenv("CONF_TOKEN", "tok123")
|
||
config_file = tmp_path / "sources.yaml"
|
||
config_file.write_text(dedent("""\
|
||
version: "1.0"
|
||
sources:
|
||
- name: "Team Confluence"
|
||
type: confluence
|
||
enabled: true
|
||
sync_frequency: daily
|
||
params:
|
||
base_url: "https://confluence.bahn.de"
|
||
api_key: "${CONF_TOKEN}"
|
||
spaces: ["ACV2"]
|
||
target_folder: references
|
||
settings:
|
||
create_tasks: true
|
||
enrichment:
|
||
provider: "kiro"
|
||
confidence_threshold: 0.8
|
||
ocr:
|
||
provider: "kiro"
|
||
git:
|
||
auto_commit: false
|
||
"""), encoding="utf-8")
|
||
|
||
sources, settings = load_sources_config(config_file)
|
||
|
||
assert len(sources) == 1
|
||
src = sources[0]
|
||
assert src.type == "confluence"
|
||
assert src.name == "Team Confluence"
|
||
assert src.params["api_key"] == "tok123"
|
||
assert src.params["spaces"] == ["ACV2"]
|
||
assert src.target_folder == "references"
|
||
assert src.sync_frequency == "daily"
|
||
assert src.enabled is True
|
||
|
||
assert settings.create_tasks is True
|
||
assert settings.enrichment.provider == "kiro"
|
||
assert settings.enrichment.confidence_threshold == 0.8
|
||
assert settings.git.auto_commit is False
|
||
|
||
def test_multiple_sources(self, tmp_path: Path) -> None:
|
||
config_file = tmp_path / "sources.yaml"
|
||
config_file.write_text(dedent("""\
|
||
version: "1.0"
|
||
sources:
|
||
- name: "Source A"
|
||
type: file
|
||
params:
|
||
path: "/data"
|
||
- name: "Source B"
|
||
type: jira
|
||
enabled: false
|
||
params:
|
||
jql: "project = X"
|
||
"""), encoding="utf-8")
|
||
|
||
sources, settings = load_sources_config(config_file)
|
||
assert len(sources) == 2
|
||
assert sources[0].name == "Source A"
|
||
assert sources[1].name == "Source B"
|
||
assert sources[1].enabled is False
|
||
|
||
def test_empty_sources_list(self, tmp_path: Path) -> None:
|
||
config_file = tmp_path / "sources.yaml"
|
||
config_file.write_text(dedent("""\
|
||
version: "1.0"
|
||
sources: []
|
||
"""), encoding="utf-8")
|
||
|
||
sources, settings = load_sources_config(config_file)
|
||
assert sources == []
|
||
assert isinstance(settings, PipelineSettings)
|
||
|
||
def test_no_settings_block(self, tmp_path: Path) -> None:
|
||
config_file = tmp_path / "sources.yaml"
|
||
config_file.write_text(dedent("""\
|
||
version: "1.0"
|
||
sources:
|
||
- name: "Minimal"
|
||
type: file
|
||
"""), encoding="utf-8")
|
||
|
||
sources, settings = load_sources_config(config_file)
|
||
assert len(sources) == 1
|
||
# Should use defaults
|
||
assert settings.create_tasks is True
|
||
assert settings.enrichment.provider == "kiro"
|
||
|
||
def test_file_not_found(self, tmp_path: Path) -> None:
|
||
with pytest.raises(FileNotFoundError):
|
||
load_sources_config(tmp_path / "nonexistent.yaml")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# load_sources_config – Validierungsfehler
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestLoadSourcesConfigValidation:
|
||
"""Tests für Validierungsfehler in load_sources_config."""
|
||
|
||
def test_missing_type(self, tmp_path: Path) -> None:
|
||
config_file = tmp_path / "sources.yaml"
|
||
config_file.write_text(dedent("""\
|
||
version: "1.0"
|
||
sources:
|
||
- name: "No Type"
|
||
params: {}
|
||
"""), encoding="utf-8")
|
||
|
||
with pytest.raises(ConfigValidationError) as exc_info:
|
||
load_sources_config(config_file)
|
||
assert "type" in str(exc_info.value).lower()
|
||
|
||
def test_missing_name(self, tmp_path: Path) -> None:
|
||
config_file = tmp_path / "sources.yaml"
|
||
config_file.write_text(dedent("""\
|
||
version: "1.0"
|
||
sources:
|
||
- type: confluence
|
||
params: {}
|
||
"""), encoding="utf-8")
|
||
|
||
with pytest.raises(ConfigValidationError) as exc_info:
|
||
load_sources_config(config_file)
|
||
assert "name" in str(exc_info.value).lower()
|
||
|
||
def test_unresolvable_env_var(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
monkeypatch.delenv("MISSING_SECRET_XYZ", raising=False)
|
||
config_file = tmp_path / "sources.yaml"
|
||
config_file.write_text(dedent("""\
|
||
version: "1.0"
|
||
sources:
|
||
- name: "Bad Source"
|
||
type: confluence
|
||
params:
|
||
api_key: "${MISSING_SECRET_XYZ}"
|
||
"""), encoding="utf-8")
|
||
|
||
with pytest.raises(ConfigValidationError) as exc_info:
|
||
load_sources_config(config_file)
|
||
assert "MISSING_SECRET_XYZ" in str(exc_info.value)
|
||
|
||
def test_multiple_errors_collected(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
monkeypatch.delenv("TOKEN_A", raising=False)
|
||
monkeypatch.delenv("TOKEN_B", raising=False)
|
||
config_file = tmp_path / "sources.yaml"
|
||
config_file.write_text(dedent("""\
|
||
version: "1.0"
|
||
sources:
|
||
- name: "Source 1"
|
||
type: confluence
|
||
params:
|
||
key_a: "${TOKEN_A}"
|
||
key_b: "${TOKEN_B}"
|
||
"""), encoding="utf-8")
|
||
|
||
with pytest.raises(ConfigValidationError) as exc_info:
|
||
load_sources_config(config_file)
|
||
assert "TOKEN_A" in str(exc_info.value)
|
||
assert "TOKEN_B" in str(exc_info.value)
|
||
|
||
def test_missing_fields_and_env_vars(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
monkeypatch.delenv("GHOST_VAR", raising=False)
|
||
config_file = tmp_path / "sources.yaml"
|
||
config_file.write_text(dedent("""\
|
||
version: "1.0"
|
||
sources:
|
||
- type: jira
|
||
params:
|
||
token: "${GHOST_VAR}"
|
||
"""), encoding="utf-8")
|
||
|
||
with pytest.raises(ConfigValidationError) as exc_info:
|
||
load_sources_config(config_file)
|
||
# Both missing name and unresolvable env var
|
||
assert "name" in str(exc_info.value).lower()
|
||
assert "GHOST_VAR" in str(exc_info.value)
|