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

575 lines
22 KiB
Python

"""Unit-Tests für Team-Isolation und Cross-Context-Leakage-Prüfung.
Testet die Implementierung von Task 15.2:
- verify_isolation: Erkennung von Pfad-Referenzen, Env-Variablen,
Config-Referenzen und Kommentaren zu anderen Kontexten
- prepare_team_repo: Erstellung eines Team-Repos mit nur eigenem Kontext
- Integration mit SecretEncryptionManager für kontextspezifische Secret-Filterung
Validates: Requirements 10.5, 10.9, 10.10
"""
from __future__ import annotations
from pathlib import Path
from unittest.mock import MagicMock
import pytest
import yaml
from monorepo.federation import FederationManager
from monorepo.models import IsolationLeak, IsolationReport, MachineContext
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def team_repos_config(tmp_path: Path) -> Path:
"""Erstellt eine temporäre team-repos.yaml mit Testdaten."""
config_path = tmp_path / "config" / "team-repos.yaml"
config_path.parent.mkdir(parents=True)
data = {
"version": "1.0",
"federation": {
"topology": "hub-and-spoke",
"hub_owner": "andre",
"conflict_strategy": "team-wins",
},
"team_repos": [
{
"context": "privat",
"url": "https://github.com/test/privat-team.git",
"branch": "main",
"sync_direction": "bidirectional",
"sync_frequency": "on-push",
},
{
"context": "dhive",
"url": "https://gitlab.dhive.io/team/dhive-mono.git",
"branch": "main",
"sync_direction": "bidirectional",
"sync_frequency": "on-push",
},
{
"context": "bahn",
"url": "https://gitlab.example.com/bahn-workspace.git",
"branch": "main",
"sync_direction": "bidirectional",
"sync_frequency": "daily",
},
],
}
with open(config_path, "w", encoding="utf-8") as f:
yaml.dump(data, f)
return config_path
@pytest.fixture
def monorepo_root(tmp_path: Path) -> Path:
"""Erstellt ein temporäres Monorepo-Verzeichnis mit Kontextordnern."""
root = tmp_path / "monorepo"
root.mkdir()
(root / "privat").mkdir()
(root / "dhive").mkdir()
(root / "bahn").mkdir()
(root / "shared").mkdir()
return root
@pytest.fixture
def federation_manager(team_repos_config: Path, monorepo_root: Path) -> FederationManager:
"""Erstellt einen FederationManager mit Testkonfiguration."""
return FederationManager(
config_path=team_repos_config,
monorepo_root=monorepo_root,
)
@pytest.fixture
def mock_encryption_manager() -> MagicMock:
"""Erstellt einen gemockten SecretEncryptionManager."""
mock = MagicMock()
mock.is_authorized.return_value = True
return mock
@pytest.fixture
def federation_manager_with_encryption(
team_repos_config: Path, monorepo_root: Path, mock_encryption_manager: MagicMock
) -> FederationManager:
"""Erstellt einen FederationManager mit SecretEncryptionManager."""
return FederationManager(
config_path=team_repos_config,
monorepo_root=monorepo_root,
encryption_manager=mock_encryption_manager,
)
# ---------------------------------------------------------------------------
# Tests: verify_isolation - Pfad-Referenzen
# ---------------------------------------------------------------------------
class TestVerifyIsolationPathReferences:
"""Tests für die Erkennung von Pfad-Referenzen auf andere Kontexte."""
def test_clean_repo_is_isolated(
self, federation_manager: FederationManager, tmp_path: Path
) -> None:
"""Prüft dass ein sauberes Repo als isoliert erkannt wird."""
team_repo = tmp_path / "team-repo"
team_repo.mkdir()
(team_repo / "src").mkdir()
(team_repo / "src" / "main.py").write_text(
"# My privat project\nprint('hello')\n", encoding="utf-8"
)
report = federation_manager.verify_isolation(team_repo, "privat")
assert report.is_isolated is True
assert report.leaks == []
assert report.context == "privat"
def test_detects_path_reference_to_dhive(
self, federation_manager: FederationManager, tmp_path: Path
) -> None:
"""Prüft Erkennung einer Pfad-Referenz auf dhive/."""
team_repo = tmp_path / "team-repo"
team_repo.mkdir()
(team_repo / "config.yaml").write_text(
"path: dhive/project-x/src\n", encoding="utf-8"
)
report = federation_manager.verify_isolation(team_repo, "privat")
assert report.is_isolated is False
assert len(report.leaks) >= 1
leak = report.leaks[0]
assert leak.leaked_context == "dhive"
assert leak.leak_type == "path"
assert leak.line_number == 1
def test_detects_path_reference_to_bahn(
self, federation_manager: FederationManager, tmp_path: Path
) -> None:
"""Prüft Erkennung einer Pfad-Referenz auf bahn/."""
team_repo = tmp_path / "team-repo"
team_repo.mkdir()
(team_repo / "script.sh").write_text(
"#!/bin/bash\ncp bahn/data.csv .\n", encoding="utf-8"
)
report = federation_manager.verify_isolation(team_repo, "privat")
assert report.is_isolated is False
path_leaks = [l for l in report.leaks if l.leak_type == "path"]
assert len(path_leaks) >= 1
assert path_leaks[0].leaked_context == "bahn"
def test_own_context_path_not_detected(
self, federation_manager: FederationManager, tmp_path: Path
) -> None:
"""Prüft dass Pfade des eigenen Kontexts NICHT als Leak erkannt werden."""
team_repo = tmp_path / "team-repo"
team_repo.mkdir()
(team_repo / "readme.md").write_text(
"## Project\nSee privat/notes for details.\n", encoding="utf-8"
)
report = federation_manager.verify_isolation(team_repo, "privat")
# privat/ should NOT be a leak when context is "privat"
assert report.is_isolated is True
def test_nonexistent_path_returns_isolated(
self, federation_manager: FederationManager, tmp_path: Path
) -> None:
"""Prüft dass nicht-existierende Pfade als isoliert behandelt werden."""
report = federation_manager.verify_isolation(
tmp_path / "nonexistent", "privat"
)
assert report.is_isolated is True
assert report.leaks == []
# ---------------------------------------------------------------------------
# Tests: verify_isolation - Env-Variablen
# ---------------------------------------------------------------------------
class TestVerifyIsolationEnvVars:
"""Tests für die Erkennung von Env-Variablen-Referenzen anderer Kontexte."""
def test_detects_dhive_env_var(
self, federation_manager: FederationManager, tmp_path: Path
) -> None:
"""Prüft Erkennung von DHIVE_* Umgebungsvariablen."""
team_repo = tmp_path / "team-repo"
team_repo.mkdir()
(team_repo / "app.py").write_text(
'import os\napi_key = os.environ["DHIVE_API_KEY"]\n',
encoding="utf-8",
)
report = federation_manager.verify_isolation(team_repo, "privat")
assert report.is_isolated is False
env_leaks = [l for l in report.leaks if l.leak_type == "env_var"]
assert len(env_leaks) >= 1
assert env_leaks[0].leaked_context == "dhive"
def test_detects_bahn_env_var(
self, federation_manager: FederationManager, tmp_path: Path
) -> None:
"""Prüft Erkennung von BAHN_* Umgebungsvariablen."""
team_repo = tmp_path / "team-repo"
team_repo.mkdir()
(team_repo / ".env.example").write_text(
"BAHN_TOKEN=your-token-here\n", encoding="utf-8"
)
report = federation_manager.verify_isolation(team_repo, "privat")
assert report.is_isolated is False
env_leaks = [l for l in report.leaks if l.leak_type == "env_var"]
assert len(env_leaks) >= 1
assert env_leaks[0].leaked_context == "bahn"
def test_own_context_env_var_not_detected(
self, federation_manager: FederationManager, tmp_path: Path
) -> None:
"""Prüft dass PRIVAT_* Env-Vars im privat-Kontext NICHT als Leak gelten."""
team_repo = tmp_path / "team-repo"
team_repo.mkdir()
(team_repo / "config.py").write_text(
'PRIVAT_SECRET = "xxx"\n', encoding="utf-8"
)
report = federation_manager.verify_isolation(team_repo, "privat")
# PRIVAT_ env vars should not be leaks when context is "privat"
assert report.is_isolated is True
# ---------------------------------------------------------------------------
# Tests: verify_isolation - Config-Referenzen
# ---------------------------------------------------------------------------
class TestVerifyIsolationConfigRefs:
"""Tests für die Erkennung von Config-Referenzen auf andere Kontexte."""
def test_detects_git_crypt_filter_reference(
self, federation_manager: FederationManager, tmp_path: Path
) -> None:
"""Prüft Erkennung von git-crypt-Filter-Referenzen anderer Kontexte."""
team_repo = tmp_path / "team-repo"
team_repo.mkdir()
(team_repo / ".gitattributes").write_text(
".env filter=git-crypt-dhive diff=git-crypt-dhive\n",
encoding="utf-8",
)
report = federation_manager.verify_isolation(team_repo, "privat")
assert report.is_isolated is False
config_leaks = [l for l in report.leaks if l.leak_type == "config_ref"]
assert len(config_leaks) >= 1
assert config_leaks[0].leaked_context == "dhive"
def test_detects_context_yaml_reference(
self, federation_manager: FederationManager, tmp_path: Path
) -> None:
"""Prüft Erkennung von context: <other> in YAML-Dateien."""
team_repo = tmp_path / "team-repo"
team_repo.mkdir()
(team_repo / "settings.yaml").write_text(
"name: project\ncontext: bahn\n", encoding="utf-8"
)
report = federation_manager.verify_isolation(team_repo, "privat")
assert report.is_isolated is False
config_leaks = [l for l in report.leaks if l.leak_type == "config_ref"]
assert len(config_leaks) >= 1
assert config_leaks[0].leaked_context == "bahn"
def test_own_git_crypt_filter_not_detected(
self, federation_manager: FederationManager, tmp_path: Path
) -> None:
"""Prüft dass der eigene git-crypt-Filter NICHT als Leak erkannt wird."""
team_repo = tmp_path / "team-repo"
team_repo.mkdir()
(team_repo / ".gitattributes").write_text(
".env filter=git-crypt-privat diff=git-crypt-privat\n",
encoding="utf-8",
)
report = federation_manager.verify_isolation(team_repo, "privat")
assert report.is_isolated is True
# ---------------------------------------------------------------------------
# Tests: verify_isolation - Kommentare
# ---------------------------------------------------------------------------
class TestVerifyIsolationComments:
"""Tests für die Erkennung von Kommentaren zu anderen Kontexten."""
def test_detects_comment_mentioning_other_context(
self, federation_manager: FederationManager, tmp_path: Path
) -> None:
"""Prüft Erkennung von Kommentaren die andere Kontexte erwähnen."""
team_repo = tmp_path / "team-repo"
team_repo.mkdir()
(team_repo / "main.py").write_text(
"# Dieses Skript wurde vom dhive-Team erstellt\n"
"print('hello')\n",
encoding="utf-8",
)
report = federation_manager.verify_isolation(team_repo, "privat")
assert report.is_isolated is False
comment_leaks = [l for l in report.leaks if l.leak_type == "comment"]
assert len(comment_leaks) >= 1
assert comment_leaks[0].leaked_context == "dhive"
def test_non_comment_context_word_not_detected_as_comment(
self, federation_manager: FederationManager, tmp_path: Path
) -> None:
"""Prüft dass Kontextnamen in nicht-Kommentar-Zeilen nicht als comment-Leak gemeldet werden."""
team_repo = tmp_path / "team-repo"
team_repo.mkdir()
(team_repo / "data.txt").write_text(
"Some text that mentions dhive in a regular line\n",
encoding="utf-8",
)
report = federation_manager.verify_isolation(team_repo, "privat")
# Should not have comment-type leaks (may have other types)
comment_leaks = [l for l in report.leaks if l.leak_type == "comment"]
assert len(comment_leaks) == 0
def test_js_comment_detected(
self, federation_manager: FederationManager, tmp_path: Path
) -> None:
"""Prüft Erkennung in JS-Style-Kommentaren (//)."""
team_repo = tmp_path / "team-repo"
team_repo.mkdir()
(team_repo / "index.js").write_text(
"// TODO: Check bahn integration\nconst x = 1;\n",
encoding="utf-8",
)
report = federation_manager.verify_isolation(team_repo, "privat")
comment_leaks = [l for l in report.leaks if l.leak_type == "comment"]
assert len(comment_leaks) >= 1
assert comment_leaks[0].leaked_context == "bahn"
# ---------------------------------------------------------------------------
# Tests: verify_isolation - Mehrfache Leaks
# ---------------------------------------------------------------------------
class TestVerifyIsolationMultipleLeaks:
"""Tests für die Erkennung mehrerer Leaks in einem Repo."""
def test_detects_multiple_leak_types(
self, federation_manager: FederationManager, tmp_path: Path
) -> None:
"""Prüft Erkennung verschiedener Leak-Typen in derselben Datei."""
team_repo = tmp_path / "team-repo"
team_repo.mkdir()
(team_repo / "config.py").write_text(
"# Für das bahn-Team\n"
'PATH = "dhive/src/module"\n'
"TOKEN = BAHN_API_TOKEN\n",
encoding="utf-8",
)
report = federation_manager.verify_isolation(team_repo, "privat")
assert report.is_isolated is False
assert len(report.leaks) >= 3
leak_types = {l.leak_type for l in report.leaks}
assert "comment" in leak_types
assert "path" in leak_types
assert "env_var" in leak_types
def test_detects_leaks_across_multiple_files(
self, federation_manager: FederationManager, tmp_path: Path
) -> None:
"""Prüft Erkennung von Leaks über mehrere Dateien."""
team_repo = tmp_path / "team-repo"
team_repo.mkdir()
(team_repo / "a.py").write_text(
'x = "dhive/config"\n', encoding="utf-8"
)
(team_repo / "b.yaml").write_text(
"context: bahn\n", encoding="utf-8"
)
report = federation_manager.verify_isolation(team_repo, "privat")
assert report.is_isolated is False
files_with_leaks = {l.file_path for l in report.leaks}
assert len(files_with_leaks) == 2
# ---------------------------------------------------------------------------
# Tests: prepare_team_repo
# ---------------------------------------------------------------------------
class TestPrepareTeamRepo:
"""Tests für die Team-Repo-Erstellung mit nur eigenem Kontext."""
def test_prepare_creates_directory(
self, federation_manager: FederationManager, monorepo_root: Path
) -> None:
"""Prüft dass ein Team-Repo-Verzeichnis erstellt wird."""
# Erstelle Quelldateien im Kontextordner
privat_dir = monorepo_root / "privat"
(privat_dir / "project" / "src").mkdir(parents=True)
(privat_dir / "project" / "src" / "main.py").write_text(
"print('hello')\n", encoding="utf-8"
)
result = federation_manager.prepare_team_repo("privat")
assert result.exists()
assert result.is_dir()
def test_prepare_copies_context_files(
self, federation_manager: FederationManager, monorepo_root: Path
) -> None:
"""Prüft dass Dateien des eigenen Kontexts kopiert werden."""
privat_dir = monorepo_root / "privat"
(privat_dir / "src").mkdir()
(privat_dir / "src" / "app.py").write_text(
"print('app')\n", encoding="utf-8"
)
(privat_dir / "README.md").write_text(
"# My Project\n", encoding="utf-8"
)
result = federation_manager.prepare_team_repo("privat")
assert (result / "src" / "app.py").exists()
assert (result / "README.md").exists()
def test_prepare_invalid_context_raises(
self, federation_manager: FederationManager
) -> None:
"""Prüft ValueError bei ungültigem Kontext."""
with pytest.raises(ValueError, match="Ungültiger Kontext"):
federation_manager.prepare_team_repo("invalid")
def test_prepare_unconfigured_context_raises(
self, team_repos_config: Path, monorepo_root: Path
) -> None:
"""Prüft ValueError wenn kein Team-Repo konfiguriert ist."""
# "shared" is not in VALID_CONTEXTS so it raises "Ungültiger Kontext"
fm = FederationManager(
config_path=team_repos_config,
monorepo_root=monorepo_root,
)
with pytest.raises(ValueError, match="Ungültiger Kontext"):
fm.prepare_team_repo("shared")
def test_prepare_sanitizes_cross_context_leaks(
self, federation_manager: FederationManager, monorepo_root: Path
) -> None:
"""Prüft dass Cross-Context-Leaks im Team-Repo bereinigt werden."""
privat_dir = monorepo_root / "privat"
(privat_dir / "config.yaml").write_text(
"path: dhive/some-project\nname: my-project\n",
encoding="utf-8",
)
result = federation_manager.prepare_team_repo("privat")
# Die Datei sollte bereinigt worden sein
config_content = (result / "config.yaml").read_text(encoding="utf-8")
assert "dhive/" not in config_content
def test_prepare_with_encryption_manager(
self,
federation_manager_with_encryption: FederationManager,
monorepo_root: Path,
) -> None:
"""Prüft Integration mit SecretEncryptionManager."""
privat_dir = monorepo_root / "privat"
(privat_dir / "src").mkdir()
(privat_dir / "src" / "main.py").write_text(
"print('hello')\n", encoding="utf-8"
)
result = federation_manager_with_encryption.prepare_team_repo("privat")
assert result.exists()
# EncryptionManager.is_authorized sollte aufgerufen worden sein
federation_manager_with_encryption.encryption_manager.is_authorized.assert_called_with("privat")
def test_prepare_filters_gitattributes(
self,
federation_manager_with_encryption: FederationManager,
monorepo_root: Path,
) -> None:
"""Prüft dass .gitattributes nur eigene Filter enthält."""
privat_dir = monorepo_root / "privat"
(privat_dir / ".gitattributes").write_text(
".env filter=git-crypt-privat diff=git-crypt-privat\n"
"*.pem filter=git-crypt-dhive diff=git-crypt-dhive\n"
"*.key filter=git-crypt-bahn diff=git-crypt-bahn\n",
encoding="utf-8",
)
result = federation_manager_with_encryption.prepare_team_repo("privat")
gitattributes = (result / ".gitattributes").read_text(encoding="utf-8")
assert "git-crypt-privat" in gitattributes
assert "git-crypt-dhive" not in gitattributes
assert "git-crypt-bahn" not in gitattributes
def test_prepare_empty_context_folder(
self, federation_manager: FederationManager, monorepo_root: Path
) -> None:
"""Prüft dass ein leerer Kontextordner ein leeres Team-Repo erzeugt."""
# privat/ exists but is empty
result = federation_manager.prepare_team_repo("privat")
assert result.exists()
# ---------------------------------------------------------------------------
# Tests: Integration verify_isolation + prepare_team_repo
# ---------------------------------------------------------------------------
class TestIsolationIntegration:
"""Integrationstests für Isolation-Verifikation mit Team-Repo-Vorbereitung."""
def test_prepared_repo_passes_isolation_check(
self, federation_manager: FederationManager, monorepo_root: Path
) -> None:
"""Prüft dass ein vorbereitetes Team-Repo die Isolationsprüfung besteht."""
privat_dir = monorepo_root / "privat"
(privat_dir / "src").mkdir()
(privat_dir / "src" / "clean.py").write_text(
"# Clean code without cross-context references\n"
"x = 42\n",
encoding="utf-8",
)
team_repo_path = federation_manager.prepare_team_repo("privat")
report = federation_manager.verify_isolation(team_repo_path, "privat")
assert report.is_isolated is True
def test_prepared_repo_with_initial_leaks_is_sanitized(
self, federation_manager: FederationManager, monorepo_root: Path
) -> None:
"""Prüft dass Leaks beim Vorbereiten automatisch bereinigt werden."""
privat_dir = monorepo_root / "privat"
(privat_dir / "leak.py").write_text(
"# Reference to dhive/project\n"
'path = "bahn/data"\n'
"clean_line = True\n",
encoding="utf-8",
)
team_repo_path = federation_manager.prepare_team_repo("privat")
# Nach Sanitisierung sollte der Report sauber sein
report = federation_manager.verify_isolation(team_repo_path, "privat")
assert report.is_isolated is True