813 lines
31 KiB
Python
813 lines
31 KiB
Python
"""Ergänzende Unit-Tests für FederationManager (Task 15.7).
|
||
|
||
Deckt Edge-Cases und Szenarien ab, die in test_federation.py und
|
||
test_federation_isolation.py noch nicht vorhanden sind:
|
||
|
||
- Subtree-Pull/Push mit verschiedenen Fehlertypen (rename, delete-modify)
|
||
- full_sync: Push-Fehler nach erfolgreichem Pull
|
||
- resolve_conflict mit konfigurierbarer Strategie
|
||
- mirror_shared: Gemischte Erfolge/Fehler
|
||
- Onboarding: Kein Leak von Monorepo-Struktur-Details
|
||
- Sync-Abbruch bei Merge-Konflikten mit Protokollierung
|
||
|
||
Validates: Requirements 10.1, 10.3, 10.5, 10.6, 10.7, 10.9, 10.12
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import subprocess
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import pytest
|
||
import yaml
|
||
|
||
from monorepo.federation import (
|
||
ConflictResult,
|
||
FederationManager,
|
||
MemberInfo,
|
||
MirrorResult,
|
||
SubtreeSyncEngine,
|
||
)
|
||
from monorepo.models import ConflictInfo, SyncResult
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fixtures
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.fixture
|
||
def team_repos_config(tmp_path: Path) -> Path:
|
||
"""Erstellt eine temporäre team-repos.yaml."""
|
||
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",
|
||
"shared_mirror": {
|
||
"enabled": True,
|
||
"paths": [
|
||
"shared/tools/common/",
|
||
"shared/config/base-config.yaml",
|
||
],
|
||
"mode": "read-only",
|
||
},
|
||
},
|
||
{
|
||
"context": "dhive",
|
||
"url": "https://gitlab.dhive.io/team/dhive-mono.git",
|
||
"branch": "main",
|
||
"sync_direction": "bidirectional",
|
||
"sync_frequency": "on-push",
|
||
"shared_mirror": {
|
||
"enabled": True,
|
||
"paths": ["shared/tools/", "shared/powers/"],
|
||
"mode": "read-only",
|
||
},
|
||
},
|
||
{
|
||
"context": "bahn",
|
||
"url": "https://gitlab.example.com/bahn-workspace.git",
|
||
"branch": "develop",
|
||
"sync_direction": "bidirectional",
|
||
"sync_frequency": "daily",
|
||
"shared_mirror": {
|
||
"enabled": True,
|
||
"paths": ["shared/tools/", "shared/knowledge-store/"],
|
||
"mode": "read-only",
|
||
},
|
||
},
|
||
],
|
||
}
|
||
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 Struktur."""
|
||
root = tmp_path / "monorepo"
|
||
root.mkdir()
|
||
(root / "privat").mkdir()
|
||
(root / "dhive").mkdir()
|
||
(root / "bahn").mkdir()
|
||
(root / "shared" / "tools" / "common").mkdir(parents=True)
|
||
(root / "shared" / "config").mkdir(parents=True)
|
||
(root / "shared" / "powers").mkdir(parents=True)
|
||
(root / "shared" / "knowledge-store").mkdir(parents=True)
|
||
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 engine(monorepo_root: Path) -> SubtreeSyncEngine:
|
||
"""Erstellt eine SubtreeSyncEngine mit Test-Root."""
|
||
return SubtreeSyncEngine(monorepo_root=monorepo_root)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: SubtreeSyncEngine - Erweiterte Konflikt-Typen
|
||
# Validates: Requirements 10.1, 10.3
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestSubtreeSyncEngineConflictTypes:
|
||
"""Tests für verschiedene Konflikt-Typen bei Subtree-Operationen."""
|
||
|
||
def test_subtree_pull_rename_conflict(self, engine: SubtreeSyncEngine) -> None:
|
||
"""Prüft Erkennung von Rename-Konflikten beim Pull."""
|
||
with patch.object(engine, "_run_git") as mock_git, \
|
||
patch.object(engine, "_abort_merge"):
|
||
error = subprocess.CalledProcessError(1, "git")
|
||
error.stdout = (
|
||
"CONFLICT (rename/rename): Rename dhive/old.py->dhive/new.py "
|
||
"in HEAD. Rename dhive/old.py->dhive/other.py in remote."
|
||
)
|
||
error.stderr = ""
|
||
mock_git.side_effect = error
|
||
|
||
result = engine.subtree_pull(
|
||
remote="https://example.com/repo.git",
|
||
prefix="dhive",
|
||
branch="main",
|
||
)
|
||
|
||
assert result.success is False
|
||
assert len(result.conflicts) >= 1
|
||
assert result.conflicts[0].conflict_type == "rename"
|
||
|
||
def test_subtree_pull_delete_modify_conflict(
|
||
self, engine: SubtreeSyncEngine
|
||
) -> None:
|
||
"""Prüft Erkennung von Delete-Modify-Konflikten beim Pull."""
|
||
with patch.object(engine, "_run_git") as mock_git, \
|
||
patch.object(engine, "_abort_merge"):
|
||
error = subprocess.CalledProcessError(1, "git")
|
||
error.stdout = (
|
||
"CONFLICT (modify/delete): bahn/removed.py deleted in remote "
|
||
"and modified in HEAD."
|
||
)
|
||
error.stderr = ""
|
||
mock_git.side_effect = error
|
||
|
||
result = engine.subtree_pull(
|
||
remote="https://example.com/repo.git",
|
||
prefix="bahn",
|
||
branch="develop",
|
||
)
|
||
|
||
assert result.success is False
|
||
assert len(result.conflicts) >= 1
|
||
assert result.conflicts[0].conflict_type == "delete-modify"
|
||
|
||
def test_subtree_pull_multiple_conflicts(
|
||
self, engine: SubtreeSyncEngine
|
||
) -> None:
|
||
"""Prüft Erkennung mehrerer Konflikte in einem Pull."""
|
||
with patch.object(engine, "_run_git") as mock_git, \
|
||
patch.object(engine, "_abort_merge"):
|
||
error = subprocess.CalledProcessError(1, "git")
|
||
error.stdout = (
|
||
"CONFLICT (content): Merge conflict in privat/a.py\n"
|
||
"CONFLICT (content): Merge conflict in privat/b.py\n"
|
||
"CONFLICT (rename/rename): Rename privat/c.py\n"
|
||
"Automatic merge failed; fix conflicts and then commit."
|
||
)
|
||
error.stderr = ""
|
||
mock_git.side_effect = error
|
||
|
||
result = engine.subtree_pull(
|
||
remote="https://example.com/repo.git",
|
||
prefix="privat",
|
||
branch="main",
|
||
)
|
||
|
||
assert result.success is False
|
||
assert len(result.conflicts) == 3
|
||
content_conflicts = [
|
||
c for c in result.conflicts if c.conflict_type == "content"
|
||
]
|
||
rename_conflicts = [
|
||
c for c in result.conflicts if c.conflict_type == "rename"
|
||
]
|
||
assert len(content_conflicts) == 2
|
||
assert len(rename_conflicts) == 1
|
||
|
||
def test_subtree_push_auth_error(self, engine: SubtreeSyncEngine) -> None:
|
||
"""Prüft Fehlerbehandlung bei Authentifizierungsfehler."""
|
||
with patch.object(engine, "_run_git") as mock_git:
|
||
error = subprocess.CalledProcessError(128, "git")
|
||
error.stdout = ""
|
||
error.stderr = "fatal: Authentication failed for 'https://...'"
|
||
mock_git.side_effect = error
|
||
|
||
result = engine.subtree_push(
|
||
remote="https://example.com/repo.git",
|
||
prefix="dhive",
|
||
branch="main",
|
||
)
|
||
|
||
assert result.success is False
|
||
assert result.conflicts[0].source == "monorepo"
|
||
assert "fehlgeschlagen" in result.conflicts[0].details
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: full_sync - Push-Fehler nach erfolgreichem Pull
|
||
# Validates: Requirements 10.3, 10.7
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestFullSyncPushFailure:
|
||
"""Tests für full_sync wenn Push nach Pull fehlschlägt."""
|
||
|
||
def test_full_sync_push_fails_after_pull_success(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft dass Push-Fehler nach erfolgreichem Pull korrekt gemeldet wird."""
|
||
with patch.object(
|
||
federation_manager.sync_engine, "subtree_pull"
|
||
) as mock_pull, patch.object(
|
||
federation_manager.sync_engine, "subtree_push"
|
||
) as mock_push, patch.object(
|
||
federation_manager.sync_engine, "detect_conflicts",
|
||
return_value=[],
|
||
):
|
||
mock_pull.return_value = SyncResult(
|
||
success=True, context="dhive", direction="pull",
|
||
commits_synced=5, conflicts=[], timestamp=datetime.now(),
|
||
)
|
||
mock_push.return_value = SyncResult(
|
||
success=False, context="dhive", direction="push",
|
||
commits_synced=0,
|
||
conflicts=[ConflictInfo(
|
||
file_path="",
|
||
conflict_type="content",
|
||
source="monorepo",
|
||
details="Push rejected: non-fast-forward",
|
||
)],
|
||
timestamp=datetime.now(),
|
||
)
|
||
result = federation_manager.full_sync("dhive")
|
||
|
||
assert result.success is False
|
||
assert result.direction == "full"
|
||
# Pull commits are counted even though push failed
|
||
assert result.commits_synced == 5
|
||
assert len(result.conflicts) == 1
|
||
assert "Push rejected" in result.conflicts[0].details
|
||
|
||
def test_full_sync_uses_correct_branch_per_context(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft dass full_sync den konfigurierten Branch verwendet."""
|
||
with patch.object(
|
||
federation_manager.sync_engine, "subtree_pull"
|
||
) as mock_pull, patch.object(
|
||
federation_manager.sync_engine, "subtree_push"
|
||
) as mock_push, patch.object(
|
||
federation_manager.sync_engine, "detect_conflicts",
|
||
return_value=[],
|
||
):
|
||
mock_pull.return_value = SyncResult(
|
||
success=True, context="bahn", direction="pull",
|
||
commits_synced=1, conflicts=[], timestamp=datetime.now(),
|
||
)
|
||
mock_push.return_value = SyncResult(
|
||
success=True, context="bahn", direction="push",
|
||
commits_synced=1, conflicts=[], timestamp=datetime.now(),
|
||
)
|
||
federation_manager.full_sync("bahn")
|
||
|
||
# bahn uses "develop" branch in our config
|
||
mock_pull.assert_called_once_with(
|
||
remote="https://gitlab.example.com/bahn-workspace.git",
|
||
prefix="bahn",
|
||
branch="develop",
|
||
)
|
||
mock_push.assert_called_once_with(
|
||
remote="https://gitlab.example.com/bahn-workspace.git",
|
||
prefix="bahn",
|
||
branch="develop",
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: Sync-Abbruch bei Merge-Konflikten mit Protokollierung
|
||
# Validates: Requirements 10.7
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestSyncAbortOnConflict:
|
||
"""Tests für Sync-Abbruch und Konflikt-Protokollierung."""
|
||
|
||
def test_pull_conflict_aborts_merge_cleanly(
|
||
self, engine: SubtreeSyncEngine
|
||
) -> None:
|
||
"""Prüft dass bei Merge-Konflikt ein abort durchgeführt wird."""
|
||
with patch.object(engine, "_run_git") as mock_git, \
|
||
patch.object(engine, "_abort_merge") as mock_abort:
|
||
error = subprocess.CalledProcessError(1, "git")
|
||
error.stdout = "CONFLICT (content): Merge conflict in file.py"
|
||
error.stderr = ""
|
||
mock_git.side_effect = error
|
||
|
||
result = engine.subtree_pull(
|
||
remote="https://example.com/repo.git",
|
||
prefix="privat",
|
||
branch="main",
|
||
)
|
||
|
||
# Merge abort muss aufgerufen werden
|
||
mock_abort.assert_called_once()
|
||
assert result.success is False
|
||
assert result.commits_synced == 0
|
||
|
||
def test_conflict_details_contain_file_info(
|
||
self, engine: SubtreeSyncEngine
|
||
) -> None:
|
||
"""Prüft dass Konflikt-Details Dateiinformationen enthalten."""
|
||
with patch.object(engine, "_run_git") as mock_git, \
|
||
patch.object(engine, "_abort_merge"):
|
||
error = subprocess.CalledProcessError(1, "git")
|
||
error.stdout = (
|
||
"CONFLICT (content): Merge conflict in privat/src/main.py\n"
|
||
"Automatic merge failed; fix conflicts and then commit."
|
||
)
|
||
error.stderr = ""
|
||
mock_git.side_effect = error
|
||
|
||
result = engine.subtree_pull(
|
||
remote="https://example.com/repo.git",
|
||
prefix="privat",
|
||
branch="main",
|
||
)
|
||
|
||
assert result.conflicts[0].file_path == "privat/src/main.py"
|
||
assert result.conflicts[0].source == "team-repo"
|
||
assert "CONFLICT" in result.conflicts[0].details
|
||
|
||
def test_full_sync_abort_preserves_pull_commit_count(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft dass bei Abbruch die bisher gezählten Commits erhalten bleiben."""
|
||
with patch.object(
|
||
federation_manager.sync_engine, "detect_conflicts",
|
||
return_value=[ConflictInfo(
|
||
file_path="privat/dirty.py",
|
||
conflict_type="content",
|
||
source="monorepo",
|
||
details="Uncommitted lokale Änderungen erkannt",
|
||
)],
|
||
):
|
||
result = federation_manager.full_sync("privat")
|
||
|
||
assert result.success is False
|
||
assert result.commits_synced == 0
|
||
assert result.conflicts[0].file_path == "privat/dirty.py"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: Konflikt-Auflösung mit team-wins Strategie (erweitert)
|
||
# Validates: Requirements 10.6
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestResolveConflictExtended:
|
||
"""Erweiterte Tests für resolve_conflict mit team-wins."""
|
||
|
||
def test_team_wins_is_configured_default_strategy(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft dass team-wins die konfigurierte Standard-Strategie ist."""
|
||
assert federation_manager.get_conflict_strategy() == "team-wins"
|
||
|
||
def test_resolve_uses_theirs_for_team_wins(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft dass team-wins git checkout --theirs für jede Datei verwendet."""
|
||
calls_made: list[list[str]] = []
|
||
|
||
def capture_calls(args: list[str]) -> MagicMock:
|
||
calls_made.append(args)
|
||
if args[0] == "diff":
|
||
return MagicMock(stdout="privat/file1.py\n", stderr="")
|
||
return MagicMock(stdout="", stderr="")
|
||
|
||
with patch.object(
|
||
federation_manager.sync_engine, "_run_git",
|
||
side_effect=capture_calls,
|
||
):
|
||
result = federation_manager.resolve_conflict(
|
||
"privat", strategy="team-wins"
|
||
)
|
||
|
||
assert result.success is True
|
||
# Verify --theirs was used
|
||
checkout_calls = [c for c in calls_made if "checkout" in c]
|
||
assert len(checkout_calls) == 1
|
||
assert "--theirs" in checkout_calls[0]
|
||
|
||
def test_resolve_uses_ours_for_hub_wins(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft dass hub-wins git checkout --ours verwendet."""
|
||
calls_made: list[list[str]] = []
|
||
|
||
def capture_calls(args: list[str]) -> MagicMock:
|
||
calls_made.append(args)
|
||
if args[0] == "diff":
|
||
return MagicMock(stdout="dhive/app.py\n", stderr="")
|
||
return MagicMock(stdout="", stderr="")
|
||
|
||
with patch.object(
|
||
federation_manager.sync_engine, "_run_git",
|
||
side_effect=capture_calls,
|
||
):
|
||
result = federation_manager.resolve_conflict(
|
||
"dhive", strategy="hub-wins"
|
||
)
|
||
|
||
assert result.success is True
|
||
checkout_calls = [c for c in calls_made if "checkout" in c]
|
||
assert "--ours" in checkout_calls[0]
|
||
|
||
def test_resolve_stages_files_after_resolution(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft dass aufgelöste Dateien mit git add gestaged werden."""
|
||
calls_made: list[list[str]] = []
|
||
|
||
def capture_calls(args: list[str]) -> MagicMock:
|
||
calls_made.append(args)
|
||
if args[0] == "diff":
|
||
return MagicMock(stdout="privat/resolved.py\n", stderr="")
|
||
return MagicMock(stdout="", stderr="")
|
||
|
||
with patch.object(
|
||
federation_manager.sync_engine, "_run_git",
|
||
side_effect=capture_calls,
|
||
):
|
||
federation_manager.resolve_conflict("privat")
|
||
|
||
add_calls = [c for c in calls_made if c[0] == "add"]
|
||
assert len(add_calls) == 1
|
||
assert "privat/resolved.py" in add_calls[0]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: Isolation-Check erweitert
|
||
# Validates: Requirements 10.5, 10.9
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestIsolationCheckExtended:
|
||
"""Erweiterte Isolation-Check-Tests für Leakage-Erkennung."""
|
||
|
||
def test_isolation_detects_mixed_path_and_env_leaks(
|
||
self, federation_manager: FederationManager, tmp_path: Path
|
||
) -> None:
|
||
"""Prüft Erkennung gemischter Leaks (Pfad + Env-Var) in einer Datei."""
|
||
team_repo = tmp_path / "team-repo"
|
||
team_repo.mkdir()
|
||
(team_repo / "settings.py").write_text(
|
||
'import os\n'
|
||
'DB_PATH = "bahn/database"\n'
|
||
'KEY = os.getenv("DHIVE_SECRET_KEY")\n',
|
||
encoding="utf-8",
|
||
)
|
||
|
||
report = federation_manager.verify_isolation(team_repo, "privat")
|
||
assert report.is_isolated is False
|
||
leak_types = {l.leak_type for l in report.leaks}
|
||
assert "path" in leak_types
|
||
assert "env_var" in leak_types
|
||
|
||
def test_isolation_binary_files_skipped(
|
||
self, federation_manager: FederationManager, tmp_path: Path
|
||
) -> None:
|
||
"""Prüft dass Binärdateien übersprungen werden (kein Crash)."""
|
||
team_repo = tmp_path / "team-repo"
|
||
team_repo.mkdir()
|
||
# Schreibe Binär-Inhalt (nicht-UTF8)
|
||
(team_repo / "image.png").write_bytes(b'\x89PNG\r\n\x1a\n\x00\x00')
|
||
# Schreibe saubere Textdatei
|
||
(team_repo / "clean.py").write_text("x = 1\n", encoding="utf-8")
|
||
|
||
report = federation_manager.verify_isolation(team_repo, "privat")
|
||
assert report.is_isolated is True
|
||
|
||
def test_isolation_git_directory_skipped(
|
||
self, federation_manager: FederationManager, tmp_path: Path
|
||
) -> None:
|
||
"""Prüft dass .git-Verzeichnisse nicht gescannt werden."""
|
||
team_repo = tmp_path / "team-repo"
|
||
team_repo.mkdir()
|
||
git_dir = team_repo / ".git" / "refs"
|
||
git_dir.mkdir(parents=True)
|
||
# .git enthält Referenzen die wie Leaks aussehen
|
||
(git_dir / "heads").write_text("dhive/feature\n", encoding="utf-8")
|
||
# Saubere Hauptdatei
|
||
(team_repo / "main.py").write_text("# clean\n", encoding="utf-8")
|
||
|
||
report = federation_manager.verify_isolation(team_repo, "privat")
|
||
assert report.is_isolated is True
|
||
|
||
def test_isolation_deep_nested_files_checked(
|
||
self, federation_manager: FederationManager, tmp_path: Path
|
||
) -> None:
|
||
"""Prüft dass tief verschachtelte Dateien auch gescannt werden."""
|
||
team_repo = tmp_path / "team-repo"
|
||
deep_dir = team_repo / "src" / "core" / "utils" / "helpers"
|
||
deep_dir.mkdir(parents=True)
|
||
(deep_dir / "config.yaml").write_text(
|
||
"remote: bahn/api-gateway\n", encoding="utf-8"
|
||
)
|
||
|
||
report = federation_manager.verify_isolation(team_repo, "privat")
|
||
assert report.is_isolated is False
|
||
# Path separator is OS-dependent, check with Path for platform compatibility
|
||
expected_parts = ("src", "core", "utils", "helpers", "config.yaml")
|
||
assert any(
|
||
all(part in l.file_path for part in expected_parts)
|
||
for l in report.leaks
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: Shared-Mirror erweitert
|
||
# Validates: Requirements 10.12
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestMirrorSharedExtended:
|
||
"""Erweiterte Tests für mirror_shared – Read-Only-Spiegelung."""
|
||
|
||
def test_mirror_mixed_success_and_failure(
|
||
self, federation_manager: FederationManager, monorepo_root: Path
|
||
) -> None:
|
||
"""Prüft dass bei gemischten Pfaden Teilfehler korrekt gemeldet werden."""
|
||
# Existierender Pfad
|
||
tools_dir = monorepo_root / "shared" / "tools" / "common"
|
||
tools_dir.mkdir(parents=True, exist_ok=True)
|
||
(tools_dir / "helper.sh").write_text("#!/bin/bash\n", encoding="utf-8")
|
||
|
||
result = federation_manager.mirror_shared("privat", [
|
||
"shared/tools/common/helper.sh", # existiert
|
||
"shared/tools/common/missing.sh", # existiert nicht
|
||
])
|
||
|
||
# Teilerfolg: eine Datei gespiegelt, eine fehlgeschlagen
|
||
assert result.success is False # Fehler vorhanden
|
||
assert "shared/tools/common/helper.sh" in result.mirrored_paths
|
||
assert any("existiert nicht" in e for e in result.errors)
|
||
|
||
def test_mirror_preserves_file_content(
|
||
self, federation_manager: FederationManager, monorepo_root: Path
|
||
) -> None:
|
||
"""Prüft dass der Dateiinhalt exakt gespiegelt wird."""
|
||
tools_dir = monorepo_root / "shared" / "tools" / "common"
|
||
tools_dir.mkdir(parents=True, exist_ok=True)
|
||
original_content = "#!/bin/bash\necho 'Version 2.0'\nexit 0\n"
|
||
(tools_dir / "deploy.sh").write_text(original_content, encoding="utf-8")
|
||
|
||
federation_manager.mirror_shared("privat", ["shared/tools/common/deploy.sh"])
|
||
|
||
mirror_file = (
|
||
monorepo_root / "privat" / ".shared-mirror"
|
||
/ "shared" / "tools" / "common" / "deploy.sh"
|
||
)
|
||
assert mirror_file.read_text(encoding="utf-8") == original_content
|
||
|
||
def test_mirror_multiple_allowed_paths(
|
||
self, federation_manager: FederationManager, monorepo_root: Path
|
||
) -> None:
|
||
"""Prüft Spiegelung mehrerer erlaubter Pfade."""
|
||
# Erstelle Dateien in verschiedenen erlaubten Pfaden
|
||
(monorepo_root / "shared" / "tools" / "common" / "a.sh").write_text(
|
||
"a", encoding="utf-8"
|
||
)
|
||
config_file = monorepo_root / "shared" / "config" / "base-config.yaml"
|
||
config_file.write_text("key: val", encoding="utf-8")
|
||
|
||
result = federation_manager.mirror_shared("privat", [
|
||
"shared/tools/common/a.sh",
|
||
"shared/config/base-config.yaml",
|
||
])
|
||
|
||
assert result.success is True
|
||
assert len(result.mirrored_paths) == 2
|
||
|
||
def test_mirror_read_only_prevents_write(
|
||
self, federation_manager: FederationManager, monorepo_root: Path
|
||
) -> None:
|
||
"""Prüft dass gespiegelte Dateien nicht beschrieben werden können."""
|
||
import stat
|
||
|
||
tools_dir = monorepo_root / "shared" / "tools" / "common"
|
||
tools_dir.mkdir(parents=True, exist_ok=True)
|
||
(tools_dir / "readonly.sh").write_text("content", encoding="utf-8")
|
||
|
||
federation_manager.mirror_shared("privat", ["shared/tools/common/readonly.sh"])
|
||
|
||
mirror_file = (
|
||
monorepo_root / "privat" / ".shared-mirror"
|
||
/ "shared" / "tools" / "common" / "readonly.sh"
|
||
)
|
||
mode = mirror_file.stat().st_mode
|
||
# Prüfe 0o444: nur Leserechte
|
||
assert mode & stat.S_IRUSR # Lesen für Owner
|
||
assert not (mode & stat.S_IWUSR) # Kein Schreiben
|
||
assert not (mode & stat.S_IXUSR) # Kein Execute
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: Onboarding ohne Monorepo-Kenntnis
|
||
# Validates: Requirements 10.9
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestOnboardingNoMonorepoKnowledge:
|
||
"""Tests dass Onboarding keine Monorepo-Struktur offenbart."""
|
||
|
||
def test_onboard_returns_only_team_repo_url(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft dass nur die Team-Repo-URL zurückgegeben wird."""
|
||
member = MemberInfo(
|
||
name="Neue Kollegin", email="kollegin@dhive.io", role="developer"
|
||
)
|
||
result = federation_manager.onboard_member("dhive", member)
|
||
|
||
assert result.success is True
|
||
assert result.team_repo_url == "https://gitlab.dhive.io/team/dhive-mono.git"
|
||
# Keine Erwähnung des Monorepos oder anderer Kontexte
|
||
assert "monorepo" not in result.team_repo_url.lower()
|
||
assert "privat" not in result.team_repo_url
|
||
assert "bahn" not in result.team_repo_url
|
||
|
||
def test_onboard_each_context_provides_correct_url(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft dass jeder Kontext die korrekte Team-Repo-URL bekommt."""
|
||
expected_urls = {
|
||
"privat": "https://github.com/test/privat-team.git",
|
||
"dhive": "https://gitlab.dhive.io/team/dhive-mono.git",
|
||
"bahn": "https://gitlab.example.com/bahn-workspace.git",
|
||
}
|
||
|
||
for context, expected_url in expected_urls.items():
|
||
member = MemberInfo(
|
||
name=f"User {context}",
|
||
email=f"user@{context}.io",
|
||
)
|
||
result = federation_manager.onboard_member(context, member)
|
||
assert result.success is True
|
||
assert result.team_repo_url == expected_url
|
||
assert result.context == context
|
||
|
||
def test_onboard_does_not_reveal_hub_topology(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft dass Hub-and-Spoke-Topologie nicht offenbart wird."""
|
||
member = MemberInfo(name="Team Member", email="member@bahn.de")
|
||
result = federation_manager.onboard_member("bahn", member)
|
||
|
||
# Serialisiere Ergebnis um nach Leaks zu suchen
|
||
result_repr = repr(result)
|
||
assert "hub" not in result_repr.lower()
|
||
assert "spoke" not in result_repr.lower()
|
||
assert "andre" not in result_repr.lower()
|
||
|
||
def test_onboard_invalid_email_format_still_rejected(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft dass leere E-Mail abgelehnt wird."""
|
||
member = MemberInfo(name="Valid Name", email=" ")
|
||
result = federation_manager.onboard_member("privat", member)
|
||
assert result.success is False
|
||
assert any("E-Mail" in e for e in result.errors)
|
||
|
||
def test_onboard_preserves_member_info(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft dass Member-Info im Ergebnis korrekt zurückgegeben wird."""
|
||
member = MemberInfo(
|
||
name="Dr. Anna Becker",
|
||
email="anna.becker@dhive.io",
|
||
role="architect",
|
||
)
|
||
result = federation_manager.onboard_member("dhive", member)
|
||
|
||
assert result.success is True
|
||
assert result.member_name == "Dr. Anna Becker"
|
||
assert result.context == "dhive"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: SubtreeSyncEngine Hilfsmethoden
|
||
# Validates: Requirements 10.1
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestSyncEngineHelpers:
|
||
"""Tests für interne Hilfsmethoden der SubtreeSyncEngine."""
|
||
|
||
def test_is_merge_conflict_various_indicators(
|
||
self, engine: SubtreeSyncEngine
|
||
) -> None:
|
||
"""Prüft Erkennung verschiedener Merge-Konflikt-Indikatoren."""
|
||
indicators = [
|
||
"CONFLICT (content): Merge conflict in file.py",
|
||
"Automatic merge failed; fix conflicts and then commit.",
|
||
"merge conflict detected",
|
||
"Merge conflict in path/to/file",
|
||
"fix conflicts and then commit the result",
|
||
]
|
||
for msg in indicators:
|
||
error = subprocess.CalledProcessError(1, "git")
|
||
error.stdout = msg
|
||
error.stderr = ""
|
||
assert engine._is_merge_conflict(error) is True, (
|
||
f"Should detect merge conflict in: {msg}"
|
||
)
|
||
|
||
def test_is_merge_conflict_false_for_other_errors(
|
||
self, engine: SubtreeSyncEngine
|
||
) -> None:
|
||
"""Prüft dass andere Fehler nicht als Merge-Konflikt erkannt werden."""
|
||
non_conflict_msgs = [
|
||
"fatal: Could not read from remote repository.",
|
||
"error: failed to push some refs",
|
||
"Permission denied (publickey).",
|
||
"fatal: not a git repository",
|
||
]
|
||
for msg in non_conflict_msgs:
|
||
error = subprocess.CalledProcessError(1, "git")
|
||
error.stdout = ""
|
||
error.stderr = msg
|
||
assert engine._is_merge_conflict(error) is False, (
|
||
f"Should NOT detect conflict in: {msg}"
|
||
)
|
||
|
||
def test_count_commits_empty_output(
|
||
self, engine: SubtreeSyncEngine
|
||
) -> None:
|
||
"""Prüft Commit-Zählung bei leerer Ausgabe."""
|
||
assert engine._count_commits("") == 0
|
||
|
||
def test_count_commits_with_changes(
|
||
self, engine: SubtreeSyncEngine
|
||
) -> None:
|
||
"""Prüft Commit-Zählung bei typischer Git-Ausgabe."""
|
||
output = "abc1234 First commit\ndef5678 Second commit\n1 file changed\n"
|
||
count = engine._count_commits(output)
|
||
assert count >= 1
|
||
|
||
def test_extract_conflicts_content_type(
|
||
self, engine: SubtreeSyncEngine
|
||
) -> None:
|
||
"""Prüft Extraktion von Content-Konflikten."""
|
||
error = subprocess.CalledProcessError(1, "git")
|
||
error.stdout = "CONFLICT (content): Merge conflict in src/main.py"
|
||
error.stderr = ""
|
||
conflicts = engine._extract_conflicts(error)
|
||
assert len(conflicts) == 1
|
||
assert conflicts[0].conflict_type == "content"
|
||
assert "src/main.py" in conflicts[0].file_path
|
||
|
||
def test_extract_conflicts_no_conflict_lines_fallback(
|
||
self, engine: SubtreeSyncEngine
|
||
) -> None:
|
||
"""Prüft Fallback wenn keine CONFLICT-Zeilen gefunden werden."""
|
||
error = subprocess.CalledProcessError(1, "git")
|
||
error.stdout = "Automatic merge failed"
|
||
error.stderr = ""
|
||
conflicts = engine._extract_conflicts(error)
|
||
assert len(conflicts) == 1
|
||
assert "Merge-Konflikt erkannt" in conflicts[0].details
|