575 lines
20 KiB
Python
575 lines
20 KiB
Python
"""Unit-Tests für ConfigMerger (Shared Tooling und MCP-Konfiguration).
|
|
|
|
Testet:
|
|
- MCP-Konfiguration-Merge mit Kontext-Vorrang (Req 8.4, 8.6)
|
|
- Konflikt-Protokollierung bei Überschreibungen
|
|
- Shared-Tool-Versionierung (Req 8.2)
|
|
- Agent-Erweiterungen: Listing und harness-agnostisches Format (Req 8.3, 8.8)
|
|
- Adapter-Mechanismus für verschiedene Harnesses (Req 8.9)
|
|
- Provisionierung eines Kontexts
|
|
|
|
Requirements: 8.2, 8.3, 8.4, 8.5, 8.6, 8.7, 8.8, 8.9
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from monorepo.shared_config import (
|
|
AgentExtension,
|
|
ConfigMerger,
|
|
HarnessAdapter,
|
|
MCPServerConfig,
|
|
MergeConflict,
|
|
MergedMCPConfig,
|
|
SharedToolStatus,
|
|
ToolReference,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def monorepo_root(tmp_path: Path) -> Path:
|
|
"""Erstellt eine Monorepo-Basisstruktur für Tests."""
|
|
# Shared-Bereich
|
|
(tmp_path / "shared" / "mcp-servers").mkdir(parents=True)
|
|
(tmp_path / "shared" / "tools").mkdir(parents=True)
|
|
(tmp_path / "shared" / "powers").mkdir(parents=True)
|
|
(tmp_path / "shared" / "config").mkdir(parents=True)
|
|
|
|
# Kontexte
|
|
for ctx in ("privat", "dhive", "bahn"):
|
|
(tmp_path / ctx).mkdir()
|
|
|
|
return tmp_path
|
|
|
|
|
|
@pytest.fixture
|
|
def merger(monorepo_root: Path) -> ConfigMerger:
|
|
"""Erstellt einen ConfigMerger mit der Test-Monorepo-Struktur."""
|
|
return ConfigMerger(monorepo_root)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# MCP-Konfiguration-Merge Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMCPConfigMerge:
|
|
"""Tests für MCP-Server-Konfiguration-Merge (Req 8.4, 8.6)."""
|
|
|
|
def test_merge_with_only_shared_config(
|
|
self, monorepo_root: Path, merger: ConfigMerger
|
|
) -> None:
|
|
"""Shared-Konfiguration wird vollständig übernommen wenn kein Override."""
|
|
shared_config = {
|
|
"servers": {
|
|
"db-context": {
|
|
"command": "npx",
|
|
"args": ["-y", "@db/mcp-server"],
|
|
"env": {"DB_TOKEN": "shared-token"},
|
|
}
|
|
}
|
|
}
|
|
config_file = monorepo_root / "shared" / "mcp-servers" / "mcp-servers.yaml"
|
|
config_file.write_text(yaml.dump(shared_config), encoding="utf-8")
|
|
|
|
result = merger.merge_mcp_config("dhive")
|
|
|
|
assert "db-context" in result.servers
|
|
assert result.servers["db-context"].command == "npx"
|
|
assert result.servers["db-context"].args == ["-y", "@db/mcp-server"]
|
|
assert result.conflicts == []
|
|
|
|
def test_context_override_takes_precedence(
|
|
self, monorepo_root: Path, merger: ConfigMerger
|
|
) -> None:
|
|
"""Kontextspezifische Konfiguration überschreibt shared (Req 8.4)."""
|
|
# Shared config
|
|
shared_config = {
|
|
"servers": {
|
|
"db-context": {
|
|
"command": "npx",
|
|
"args": ["-y", "@db/mcp-server"],
|
|
"env": {"DB_TOKEN": "shared-token"},
|
|
}
|
|
}
|
|
}
|
|
config_file = monorepo_root / "shared" / "mcp-servers" / "mcp-servers.yaml"
|
|
config_file.write_text(yaml.dump(shared_config), encoding="utf-8")
|
|
|
|
# Context override
|
|
ctx_dir = monorepo_root / "dhive" / ".mcp-servers"
|
|
ctx_dir.mkdir()
|
|
ctx_config = {
|
|
"servers": {
|
|
"db-context": {
|
|
"command": "node",
|
|
"args": ["./custom-server.js"],
|
|
"env": {"DB_TOKEN": "dhive-token"},
|
|
}
|
|
}
|
|
}
|
|
(ctx_dir / "mcp-servers.yaml").write_text(
|
|
yaml.dump(ctx_config), encoding="utf-8"
|
|
)
|
|
|
|
result = merger.merge_mcp_config("dhive")
|
|
|
|
# Context values must win
|
|
assert result.servers["db-context"].command == "node"
|
|
assert result.servers["db-context"].args == ["./custom-server.js"]
|
|
assert result.servers["db-context"].env["DB_TOKEN"] == "dhive-token"
|
|
|
|
def test_conflicts_are_logged(
|
|
self, monorepo_root: Path, merger: ConfigMerger
|
|
) -> None:
|
|
"""Konflikte werden protokolliert (Req 8.6)."""
|
|
shared_config = {
|
|
"servers": {
|
|
"my-server": {
|
|
"command": "npx",
|
|
"env": {"API_KEY": "shared-key"},
|
|
}
|
|
}
|
|
}
|
|
(monorepo_root / "shared" / "mcp-servers" / "mcp-servers.yaml").write_text(
|
|
yaml.dump(shared_config), encoding="utf-8"
|
|
)
|
|
|
|
ctx_dir = monorepo_root / "bahn" / ".mcp-servers"
|
|
ctx_dir.mkdir()
|
|
ctx_config = {
|
|
"servers": {
|
|
"my-server": {
|
|
"command": "python",
|
|
"env": {"API_KEY": "bahn-key"},
|
|
}
|
|
}
|
|
}
|
|
(ctx_dir / "mcp-servers.yaml").write_text(
|
|
yaml.dump(ctx_config), encoding="utf-8"
|
|
)
|
|
|
|
result = merger.merge_mcp_config("bahn")
|
|
|
|
assert len(result.conflicts) >= 2 # command + env.API_KEY
|
|
conflict_keys = [c.key for c in result.conflicts]
|
|
assert "command" in conflict_keys
|
|
assert "env.API_KEY" in conflict_keys
|
|
|
|
# Check conflict details
|
|
cmd_conflict = next(c for c in result.conflicts if c.key == "command")
|
|
assert cmd_conflict.shared_value == "npx"
|
|
assert cmd_conflict.context_value == "python"
|
|
assert cmd_conflict.resolution == "context-wins"
|
|
assert cmd_conflict.context == "bahn"
|
|
|
|
def test_context_only_servers_are_added(
|
|
self, monorepo_root: Path, merger: ConfigMerger
|
|
) -> None:
|
|
"""Server die nur im Kontext existieren werden hinzugefügt."""
|
|
shared_config = {"servers": {"shared-only": {"command": "npx"}}}
|
|
(monorepo_root / "shared" / "mcp-servers" / "mcp-servers.yaml").write_text(
|
|
yaml.dump(shared_config), encoding="utf-8"
|
|
)
|
|
|
|
ctx_dir = monorepo_root / "privat" / ".mcp-servers"
|
|
ctx_dir.mkdir()
|
|
ctx_config = {"servers": {"ctx-only": {"command": "python"}}}
|
|
(ctx_dir / "mcp-servers.yaml").write_text(
|
|
yaml.dump(ctx_config), encoding="utf-8"
|
|
)
|
|
|
|
result = merger.merge_mcp_config("privat")
|
|
|
|
assert "shared-only" in result.servers
|
|
assert "ctx-only" in result.servers
|
|
assert result.servers["ctx-only"].command == "python"
|
|
|
|
def test_no_config_returns_empty(self, merger: ConfigMerger) -> None:
|
|
"""Keine Konfigurationsdateien → leeres Ergebnis."""
|
|
result = merger.merge_mcp_config("dhive")
|
|
assert result.servers == {}
|
|
assert result.conflicts == []
|
|
|
|
def test_env_merge_combines_keys(
|
|
self, monorepo_root: Path, merger: ConfigMerger
|
|
) -> None:
|
|
"""Env-Variablen aus shared und Kontext werden kombiniert."""
|
|
shared_config = {
|
|
"servers": {
|
|
"s1": {"command": "x", "env": {"A": "1", "B": "2"}}
|
|
}
|
|
}
|
|
(monorepo_root / "shared" / "mcp-servers" / "mcp-servers.yaml").write_text(
|
|
yaml.dump(shared_config), encoding="utf-8"
|
|
)
|
|
|
|
ctx_dir = monorepo_root / "dhive" / ".mcp-servers"
|
|
ctx_dir.mkdir()
|
|
ctx_config = {
|
|
"servers": {
|
|
"s1": {"command": "x", "env": {"B": "override", "C": "3"}}
|
|
}
|
|
}
|
|
(ctx_dir / "mcp-servers.yaml").write_text(
|
|
yaml.dump(ctx_config), encoding="utf-8"
|
|
)
|
|
|
|
result = merger.merge_mcp_config("dhive")
|
|
env = result.servers["s1"].env
|
|
|
|
assert env["A"] == "1" # from shared
|
|
assert env["B"] == "override" # overridden by context
|
|
assert env["C"] == "3" # only in context
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Shared-Tool-Versionierung Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSharedToolVersioning:
|
|
"""Tests für Shared-Tool-Versionierung (Req 8.2)."""
|
|
|
|
def test_ensure_tool_links_creates_symlinks(
|
|
self, monorepo_root: Path, merger: ConfigMerger
|
|
) -> None:
|
|
"""Tool-Symlinks werden korrekt erstellt."""
|
|
# Create a fake shared tool
|
|
tool_dir = monorepo_root / "shared" / "tools" / "my-tool"
|
|
tool_dir.mkdir()
|
|
(tool_dir / "VERSION").write_text("1.2.3", encoding="utf-8")
|
|
|
|
refs = merger.ensure_tool_links("dhive")
|
|
|
|
assert len(refs) == 1
|
|
assert refs[0].name == "my-tool"
|
|
assert refs[0].version == "1.2.3"
|
|
|
|
# Verify the link/reference exists
|
|
link_path = monorepo_root / "dhive" / ".tools" / "my-tool"
|
|
assert link_path.exists() or (
|
|
monorepo_root / "dhive" / ".tools" / "my-tool.ref.yaml"
|
|
).exists()
|
|
|
|
def test_tool_version_from_pyproject(
|
|
self, monorepo_root: Path, merger: ConfigMerger
|
|
) -> None:
|
|
"""Version wird aus pyproject.toml gelesen."""
|
|
tool_dir = monorepo_root / "shared" / "tools" / "cli-tool"
|
|
tool_dir.mkdir()
|
|
pyproject_content = '[project]\nname = "cli-tool"\nversion = "0.5.0"\n'
|
|
(tool_dir / "pyproject.toml").write_text(pyproject_content, encoding="utf-8")
|
|
|
|
refs = merger.ensure_tool_links("bahn")
|
|
assert refs[0].version == "0.5.0"
|
|
|
|
def test_check_tool_status_reports_missing(
|
|
self, monorepo_root: Path, merger: ConfigMerger
|
|
) -> None:
|
|
"""Status-Check erkennt fehlende Tool-Links."""
|
|
tool_dir = monorepo_root / "shared" / "tools" / "some-tool"
|
|
tool_dir.mkdir()
|
|
(tool_dir / "VERSION").write_text("2.0.0", encoding="utf-8")
|
|
|
|
statuses = merger.check_tool_status("privat")
|
|
assert len(statuses) == 1
|
|
assert statuses[0].tool_name == "some-tool"
|
|
assert statuses[0].is_current is False
|
|
assert statuses[0].local_version == "missing"
|
|
|
|
def test_ensure_tool_links_is_idempotent(
|
|
self, monorepo_root: Path, merger: ConfigMerger
|
|
) -> None:
|
|
"""Wiederholter Aufruf verändert nichts (Links schon aktuell)."""
|
|
tool_dir = monorepo_root / "shared" / "tools" / "my-tool"
|
|
tool_dir.mkdir()
|
|
(tool_dir / "VERSION").write_text("1.0.0", encoding="utf-8")
|
|
|
|
refs1 = merger.ensure_tool_links("dhive")
|
|
refs2 = merger.ensure_tool_links("dhive")
|
|
|
|
assert len(refs1) == len(refs2)
|
|
assert refs1[0].name == refs2[0].name
|
|
|
|
def test_no_tools_returns_empty(self, merger: ConfigMerger) -> None:
|
|
"""Keine Shared-Tools → leere Liste."""
|
|
refs = merger.ensure_tool_links("privat")
|
|
assert refs == []
|
|
|
|
def test_hidden_dirs_are_skipped(
|
|
self, monorepo_root: Path, merger: ConfigMerger
|
|
) -> None:
|
|
"""Versteckte Verzeichnisse (mit Punkt) werden übersprungen."""
|
|
(monorepo_root / "shared" / "tools" / ".hidden").mkdir()
|
|
(monorepo_root / "shared" / "tools" / "visible").mkdir()
|
|
|
|
refs = merger.ensure_tool_links("bahn")
|
|
names = [r.name for r in refs]
|
|
assert ".hidden" not in names
|
|
assert "visible" in names
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Agent-Erweiterungen Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAgentExtensions:
|
|
"""Tests für Agent-Erweiterungen (Req 8.3, 8.8, 8.9)."""
|
|
|
|
def test_list_powers_from_shared(
|
|
self, monorepo_root: Path, merger: ConfigMerger
|
|
) -> None:
|
|
"""Powers aus shared/powers/ werden korrekt gelistet."""
|
|
power_dir = monorepo_root / "shared" / "powers" / "db-dxp-platform"
|
|
power_dir.mkdir()
|
|
(power_dir / "POWER.md").write_text(
|
|
"# DB DXP Platform\nHelps with platform stuff.", encoding="utf-8"
|
|
)
|
|
|
|
extensions = merger.list_extensions()
|
|
assert len(extensions) == 1
|
|
assert extensions[0].name == "db-dxp-platform"
|
|
assert extensions[0].type == "power"
|
|
assert extensions[0].description == "DB DXP Platform"
|
|
|
|
def test_list_extensions_with_metadata(
|
|
self, monorepo_root: Path, merger: ConfigMerger
|
|
) -> None:
|
|
"""Extension mit extension.yaml Metadaten werden korrekt geladen."""
|
|
power_dir = monorepo_root / "shared" / "powers" / "my-power"
|
|
power_dir.mkdir()
|
|
meta = {
|
|
"name": "My Custom Power",
|
|
"type": "power",
|
|
"description": "A custom power for testing",
|
|
"tags": ["testing", "custom"],
|
|
}
|
|
(power_dir / "extension.yaml").write_text(
|
|
yaml.dump(meta), encoding="utf-8"
|
|
)
|
|
|
|
extensions = merger.list_extensions()
|
|
assert len(extensions) == 1
|
|
assert extensions[0].name == "My Custom Power"
|
|
assert extensions[0].description == "A custom power for testing"
|
|
assert extensions[0].tags == ["testing", "custom"]
|
|
|
|
def test_filter_by_type(
|
|
self, monorepo_root: Path, merger: ConfigMerger
|
|
) -> None:
|
|
"""Filter nach Extension-Typ funktioniert."""
|
|
(monorepo_root / "shared" / "powers" / "power-1").mkdir()
|
|
ext_dir = monorepo_root / "shared" / "config" / "extensions"
|
|
ext_dir.mkdir()
|
|
(ext_dir / "steering.md").write_text("# Steering", encoding="utf-8")
|
|
|
|
powers = merger.list_extensions(ext_type="power")
|
|
steering = merger.list_extensions(ext_type="steering")
|
|
|
|
assert all(e.type == "power" for e in powers)
|
|
assert all(e.type == "steering" for e in steering)
|
|
|
|
def test_get_extensions_for_context_includes_shared(
|
|
self, monorepo_root: Path, merger: ConfigMerger
|
|
) -> None:
|
|
"""Kontext erhält shared-Extensions als Read-Only-Referenz (Req 8.3)."""
|
|
power_dir = monorepo_root / "shared" / "powers" / "shared-power"
|
|
power_dir.mkdir()
|
|
(power_dir / "POWER.md").write_text("# Shared Power", encoding="utf-8")
|
|
|
|
extensions = merger.get_extensions_for_context("dhive")
|
|
assert any(e.name == "shared-power" for e in extensions)
|
|
|
|
def test_get_extensions_includes_context_specific(
|
|
self, monorepo_root: Path, merger: ConfigMerger
|
|
) -> None:
|
|
"""Kontextspezifische Erweiterungen werden zusätzlich geladen."""
|
|
# Shared
|
|
(monorepo_root / "shared" / "powers" / "shared-power").mkdir()
|
|
|
|
# Context-specific
|
|
ctx_ext = monorepo_root / "dhive" / ".extensions"
|
|
ctx_ext.mkdir()
|
|
(ctx_ext / "local-skill").mkdir()
|
|
(ctx_ext / "local-skill" / "README.md").write_text(
|
|
"# Local Skill", encoding="utf-8"
|
|
)
|
|
|
|
extensions = merger.get_extensions_for_context("dhive")
|
|
names = [e.name for e in extensions]
|
|
assert "shared-power" in names
|
|
assert "local-skill" in names
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Harness-Adapter Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestHarnessAdapter:
|
|
"""Tests für den Adapter-Mechanismus (Req 8.7, 8.9)."""
|
|
|
|
def test_load_adapters_from_config(self, monorepo_root: Path) -> None:
|
|
"""Adapter werden aus harness-adapters.yaml geladen."""
|
|
adapters_config = {
|
|
"adapters": {
|
|
"kiro": {
|
|
"config_format": "kiro-powers",
|
|
"output_path": ".kiro/powers",
|
|
},
|
|
"codex": {
|
|
"config_format": "codex-agent",
|
|
"output_path": ".codex/agents",
|
|
},
|
|
}
|
|
}
|
|
(monorepo_root / "shared" / "config" / "harness-adapters.yaml").write_text(
|
|
yaml.dump(adapters_config), encoding="utf-8"
|
|
)
|
|
|
|
merger = ConfigMerger(monorepo_root)
|
|
adapters = merger.list_adapters()
|
|
|
|
assert len(adapters) == 2
|
|
names = [a.harness_name for a in adapters]
|
|
assert "kiro" in names
|
|
assert "codex" in names
|
|
|
|
def test_get_adapter_returns_none_for_unknown(
|
|
self, merger: ConfigMerger
|
|
) -> None:
|
|
"""Unbekannter Harness → None."""
|
|
assert merger.get_adapter("unknown-harness") is None
|
|
|
|
def test_transform_extension_for_codex(
|
|
self, monorepo_root: Path
|
|
) -> None:
|
|
"""Transformation für Codex erzeugt AGENTS.md Referenz."""
|
|
adapters_config = {
|
|
"adapters": {
|
|
"codex": {"config_format": "codex-agent"},
|
|
}
|
|
}
|
|
(monorepo_root / "shared" / "config" / "harness-adapters.yaml").write_text(
|
|
yaml.dump(adapters_config), encoding="utf-8"
|
|
)
|
|
|
|
# Create a power to transform
|
|
power_dir = monorepo_root / "shared" / "powers" / "test-power"
|
|
power_dir.mkdir()
|
|
(power_dir / "POWER.md").write_text("# Test Power\nDoes stuff.", encoding="utf-8")
|
|
|
|
merger = ConfigMerger(monorepo_root)
|
|
ext = AgentExtension(
|
|
name="test-power",
|
|
type="power",
|
|
path=power_dir,
|
|
description="Test power for testing",
|
|
)
|
|
|
|
result = merger.transform_extension_for_harness(ext, "codex")
|
|
assert result is not None
|
|
assert result.exists()
|
|
content = result.read_text(encoding="utf-8")
|
|
assert "test-power" in content
|
|
|
|
def test_transform_returns_none_without_adapter(
|
|
self, merger: ConfigMerger, monorepo_root: Path
|
|
) -> None:
|
|
"""Transformation ohne konfigurierten Adapter → None."""
|
|
ext = AgentExtension(
|
|
name="test",
|
|
type="power",
|
|
path=monorepo_root / "shared" / "powers",
|
|
)
|
|
result = merger.transform_extension_for_harness(ext, "nonexistent")
|
|
assert result is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Harness-Konfiguration pro Kontext Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestHarnessConfig:
|
|
"""Tests für Harness-Konfiguration pro Kontext (Req 8.7)."""
|
|
|
|
def test_get_harness_config_returns_context_config(
|
|
self, monorepo_root: Path, merger: ConfigMerger
|
|
) -> None:
|
|
"""Harness-Konfiguration pro Kontext wird korrekt zurückgegeben."""
|
|
config = {
|
|
"contexts": {
|
|
"dhive": {
|
|
"harness": "codex",
|
|
"config": {"model": "o3-pro"},
|
|
},
|
|
"bahn": {
|
|
"harness": "kiro",
|
|
"config": {"powers_path": "shared/powers"},
|
|
},
|
|
}
|
|
}
|
|
(monorepo_root / "shared" / "config" / "harness-config.yaml").write_text(
|
|
yaml.dump(config), encoding="utf-8"
|
|
)
|
|
|
|
dhive_config = merger.get_harness_config("dhive")
|
|
assert dhive_config["harness"] == "codex"
|
|
|
|
bahn_config = merger.get_harness_config("bahn")
|
|
assert bahn_config["harness"] == "kiro"
|
|
|
|
def test_get_harness_config_returns_empty_for_unknown(
|
|
self, merger: ConfigMerger
|
|
) -> None:
|
|
"""Unbekannter Kontext → leeres Dict."""
|
|
assert merger.get_harness_config("unknown") == {}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Provisionierung Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestProvisionContext:
|
|
"""Tests für die Kontext-Provisionierung."""
|
|
|
|
def test_provision_context_integrates_all(
|
|
self, monorepo_root: Path, merger: ConfigMerger
|
|
) -> None:
|
|
"""provision_context liefert Tools, MCP-Config und Extensions."""
|
|
# Setup shared tool
|
|
tool_dir = monorepo_root / "shared" / "tools" / "test-tool"
|
|
tool_dir.mkdir()
|
|
(tool_dir / "VERSION").write_text("1.0.0", encoding="utf-8")
|
|
|
|
# Setup shared MCP config
|
|
mcp_config = {"servers": {"s1": {"command": "npx"}}}
|
|
(monorepo_root / "shared" / "mcp-servers" / "mcp-servers.yaml").write_text(
|
|
yaml.dump(mcp_config), encoding="utf-8"
|
|
)
|
|
|
|
# Setup shared power
|
|
(monorepo_root / "shared" / "powers" / "test-power").mkdir()
|
|
|
|
result = merger.provision_context("dhive")
|
|
|
|
assert result["context"] == "dhive"
|
|
assert len(result["tools"]) == 1
|
|
assert result["tools"][0]["name"] == "test-tool"
|
|
assert "s1" in result["mcp_config"]["servers"]
|
|
assert any(e["name"] == "test-power" for e in result["extensions"])
|