1063 lines
42 KiB
Python
1063 lines
42 KiB
Python
"""Unit-Tests für FederationManager und SubtreeSyncEngine.
|
||
|
||
Testet die Föderations-Basisklasse mit Hub-and-Spoke-Topologie:
|
||
- Konfigurationsladung aus team-repos.yaml
|
||
- Sync-Richtungsprüfung (bidirectional, hub-to-spoke, spoke-to-hub)
|
||
- SubtreeSyncEngine Pull/Push-Operationen (gemocked)
|
||
- Konflikterkennung
|
||
- Fehlerbehandlung bei ungültigen Kontexten
|
||
|
||
Validates: Requirements 10.1, 10.2, 10.3, 10.4, 10.8
|
||
"""
|
||
|
||
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,
|
||
FederationTopologyConfig,
|
||
MemberInfo,
|
||
MemberOnboardingResult,
|
||
MirrorResult,
|
||
SubtreeSyncEngine,
|
||
)
|
||
from monorepo.models import ConflictInfo, SyncResult, TeamRepoEntry
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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",
|
||
"shared_mirror": {
|
||
"enabled": True,
|
||
"paths": ["shared/tools/common/"],
|
||
"mode": "read-only",
|
||
},
|
||
},
|
||
{
|
||
"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 directional_config(tmp_path: Path) -> Path:
|
||
"""Erstellt eine team-repos.yaml mit verschiedenen Sync-Richtungen."""
|
||
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.git",
|
||
"branch": "main",
|
||
"sync_direction": "hub-to-spoke",
|
||
"sync_frequency": "manual",
|
||
},
|
||
{
|
||
"context": "dhive",
|
||
"url": "https://gitlab.dhive.io/dhive.git",
|
||
"branch": "main",
|
||
"sync_direction": "spoke-to-hub",
|
||
"sync_frequency": "manual",
|
||
},
|
||
],
|
||
}
|
||
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."""
|
||
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,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: Konfigurationsladung
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestFederationConfig:
|
||
"""Tests für die Konfigurationsladung aus team-repos.yaml."""
|
||
|
||
def test_load_config_topology(self, federation_manager: FederationManager) -> None:
|
||
"""Prüft korrekte Topologie-Konfiguration."""
|
||
config = federation_manager.topology_config
|
||
assert config.topology == "hub-and-spoke"
|
||
assert config.hub_owner == "andre"
|
||
assert config.conflict_strategy == "team-wins"
|
||
|
||
def test_load_config_team_repos(self, federation_manager: FederationManager) -> None:
|
||
"""Prüft dass alle Team-Repos geladen werden."""
|
||
repos = federation_manager.get_team_repos()
|
||
assert len(repos) == 3
|
||
contexts = [r.context for r in repos]
|
||
assert "privat" in contexts
|
||
assert "dhive" in contexts
|
||
assert "bahn" in contexts
|
||
|
||
def test_load_config_shared_mirror(self, federation_manager: FederationManager) -> None:
|
||
"""Prüft dass SharedMirrorConfig korrekt geparst wird."""
|
||
repos = federation_manager.get_team_repos()
|
||
privat_repo = next(r for r in repos if r.context == "privat")
|
||
assert privat_repo.shared_mirror is not None
|
||
assert privat_repo.shared_mirror.enabled is True
|
||
assert "shared/tools/common/" in privat_repo.shared_mirror.paths
|
||
assert privat_repo.shared_mirror.mode == "read-only"
|
||
|
||
def test_load_config_no_shared_mirror(self, federation_manager: FederationManager) -> None:
|
||
"""Prüft dass fehlende SharedMirrorConfig als None geparst wird."""
|
||
repos = federation_manager.get_team_repos()
|
||
dhive_repo = next(r for r in repos if r.context == "dhive")
|
||
assert dhive_repo.shared_mirror is None
|
||
|
||
def test_load_config_file_not_found(self, tmp_path: Path) -> None:
|
||
"""Prüft FileNotFoundError bei fehlender Konfiguration."""
|
||
with pytest.raises(FileNotFoundError):
|
||
FederationManager(
|
||
config_path=tmp_path / "nonexistent.yaml",
|
||
monorepo_root=tmp_path,
|
||
)
|
||
|
||
def test_load_config_invalid_yaml(self, tmp_path: Path) -> None:
|
||
"""Prüft ValueError bei ungültigem YAML-Format."""
|
||
config_path = tmp_path / "bad.yaml"
|
||
config_path.write_text("- just a list", encoding="utf-8")
|
||
with pytest.raises(ValueError, match="Ungültiges team-repos.yaml"):
|
||
FederationManager(config_path=config_path, monorepo_root=tmp_path)
|
||
|
||
def test_get_conflict_strategy(self, federation_manager: FederationManager) -> None:
|
||
"""Prüft dass get_conflict_strategy den konfigurierten Wert zurückgibt."""
|
||
assert federation_manager.get_conflict_strategy() == "team-wins"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: sync_from_team
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestSyncFromTeam:
|
||
"""Tests für sync_from_team (Pull vom Team-Repo)."""
|
||
|
||
def test_sync_from_team_success(self, federation_manager: FederationManager) -> None:
|
||
"""Prüft erfolgreichen Pull vom Team-Repo."""
|
||
with patch.object(
|
||
federation_manager.sync_engine, "subtree_pull"
|
||
) as mock_pull:
|
||
mock_pull.return_value = SyncResult(
|
||
success=True,
|
||
context="privat",
|
||
direction="pull",
|
||
commits_synced=3,
|
||
conflicts=[],
|
||
timestamp=datetime.now(),
|
||
)
|
||
result = federation_manager.sync_from_team("privat")
|
||
|
||
assert result.success is True
|
||
assert result.context == "privat"
|
||
assert result.direction == "pull"
|
||
assert result.commits_synced == 3
|
||
mock_pull.assert_called_once_with(
|
||
remote="https://github.com/test/privat-team.git",
|
||
prefix="privat",
|
||
branch="main",
|
||
)
|
||
|
||
def test_sync_from_team_unknown_context(self, federation_manager: FederationManager) -> None:
|
||
"""Prüft Fehlermeldung bei unbekanntem Kontext."""
|
||
result = federation_manager.sync_from_team("nonexistent")
|
||
assert result.success is False
|
||
assert len(result.conflicts) == 1
|
||
assert "nonexistent" in result.conflicts[0].details
|
||
|
||
def test_sync_from_team_hub_to_spoke_blocked(
|
||
self, directional_config: Path, monorepo_root: Path
|
||
) -> None:
|
||
"""Prüft dass Pull bei hub-to-spoke Richtung blockiert wird."""
|
||
fm = FederationManager(config_path=directional_config, monorepo_root=monorepo_root)
|
||
result = fm.sync_from_team("privat")
|
||
assert result.success is False
|
||
assert "hub-to-spoke" in result.conflicts[0].details
|
||
|
||
def test_sync_from_team_merge_conflict(self, federation_manager: FederationManager) -> None:
|
||
"""Prüft Behandlung von Merge-Konflikten beim Pull."""
|
||
with patch.object(
|
||
federation_manager.sync_engine, "subtree_pull"
|
||
) as mock_pull:
|
||
mock_pull.return_value = SyncResult(
|
||
success=False,
|
||
context="dhive",
|
||
direction="pull",
|
||
commits_synced=0,
|
||
conflicts=[ConflictInfo(
|
||
file_path="dhive/src/main.py",
|
||
conflict_type="content",
|
||
source="team-repo",
|
||
details="CONFLICT (content): Merge conflict in dhive/src/main.py",
|
||
)],
|
||
timestamp=datetime.now(),
|
||
)
|
||
result = federation_manager.sync_from_team("dhive")
|
||
|
||
assert result.success is False
|
||
assert result.context == "dhive"
|
||
assert len(result.conflicts) == 1
|
||
assert result.conflicts[0].file_path == "dhive/src/main.py"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: sync_to_team
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestSyncToTeam:
|
||
"""Tests für sync_to_team (Push zum Team-Repo)."""
|
||
|
||
def test_sync_to_team_success(self, federation_manager: FederationManager) -> None:
|
||
"""Prüft erfolgreichen Push zum Team-Repo."""
|
||
with patch.object(
|
||
federation_manager.sync_engine, "subtree_push"
|
||
) as mock_push:
|
||
mock_push.return_value = SyncResult(
|
||
success=True,
|
||
context="bahn",
|
||
direction="push",
|
||
commits_synced=2,
|
||
conflicts=[],
|
||
timestamp=datetime.now(),
|
||
)
|
||
result = federation_manager.sync_to_team("bahn")
|
||
|
||
assert result.success is True
|
||
assert result.context == "bahn"
|
||
assert result.direction == "push"
|
||
assert result.commits_synced == 2
|
||
mock_push.assert_called_once_with(
|
||
remote="https://gitlab.example.com/bahn-workspace.git",
|
||
prefix="bahn",
|
||
branch="main",
|
||
)
|
||
|
||
def test_sync_to_team_unknown_context(self, federation_manager: FederationManager) -> None:
|
||
"""Prüft Fehlermeldung bei unbekanntem Kontext."""
|
||
result = federation_manager.sync_to_team("nonexistent")
|
||
assert result.success is False
|
||
assert "nonexistent" in result.conflicts[0].details
|
||
|
||
def test_sync_to_team_spoke_to_hub_blocked(
|
||
self, directional_config: Path, monorepo_root: Path
|
||
) -> None:
|
||
"""Prüft dass Push bei spoke-to-hub Richtung blockiert wird."""
|
||
fm = FederationManager(config_path=directional_config, monorepo_root=monorepo_root)
|
||
result = fm.sync_to_team("dhive")
|
||
assert result.success is False
|
||
assert "spoke-to-hub" in result.conflicts[0].details
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: full_sync
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestFullSync:
|
||
"""Tests für full_sync (bidirektionale Synchronisation)."""
|
||
|
||
def test_full_sync_success(self, federation_manager: FederationManager) -> None:
|
||
"""Prüft erfolgreiche bidirektionale Synchronisation."""
|
||
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="privat", direction="pull",
|
||
commits_synced=2, conflicts=[], timestamp=datetime.now(),
|
||
)
|
||
mock_push.return_value = SyncResult(
|
||
success=True, context="privat", direction="push",
|
||
commits_synced=1, conflicts=[], timestamp=datetime.now(),
|
||
)
|
||
result = federation_manager.full_sync("privat")
|
||
|
||
assert result.success is True
|
||
assert result.direction == "full"
|
||
assert result.commits_synced == 3 # 2 + 1
|
||
assert result.conflicts == []
|
||
|
||
def test_full_sync_aborts_on_pull_conflict(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft Abbruch bei Merge-Konflikt während Pull."""
|
||
with patch.object(
|
||
federation_manager.sync_engine, "subtree_pull"
|
||
) as mock_pull, patch.object(
|
||
federation_manager.sync_engine, "detect_conflicts",
|
||
return_value=[],
|
||
):
|
||
mock_pull.return_value = SyncResult(
|
||
success=False, context="privat", direction="pull",
|
||
commits_synced=0,
|
||
conflicts=[ConflictInfo(
|
||
file_path="privat/config.yaml",
|
||
conflict_type="content",
|
||
source="team-repo",
|
||
details="Merge-Konflikt",
|
||
)],
|
||
timestamp=datetime.now(),
|
||
)
|
||
result = federation_manager.full_sync("privat")
|
||
|
||
assert result.success is False
|
||
assert result.direction == "full"
|
||
assert len(result.conflicts) == 1
|
||
|
||
def test_full_sync_aborts_on_pre_conflicts(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft Abbruch bei vorab erkannten Konflikten."""
|
||
with patch.object(
|
||
federation_manager.sync_engine, "detect_conflicts",
|
||
return_value=[ConflictInfo(
|
||
file_path="privat/local-change.py",
|
||
conflict_type="content",
|
||
source="monorepo",
|
||
details="Uncommitted lokale Änderungen",
|
||
)],
|
||
):
|
||
result = federation_manager.full_sync("privat")
|
||
|
||
assert result.success is False
|
||
assert len(result.conflicts) == 1
|
||
assert "Uncommitted" in result.conflicts[0].details
|
||
|
||
def test_full_sync_non_bidirectional_blocked(
|
||
self, directional_config: Path, monorepo_root: Path
|
||
) -> None:
|
||
"""Prüft dass full_sync bei nicht-bidirektionaler Konfiguration blockiert."""
|
||
fm = FederationManager(config_path=directional_config, monorepo_root=monorepo_root)
|
||
result = fm.full_sync("privat")
|
||
assert result.success is False
|
||
assert "nicht erlaubt" in result.conflicts[0].details
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: detect_conflicts
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestDetectConflicts:
|
||
"""Tests für detect_conflicts."""
|
||
|
||
def test_detect_conflicts_no_conflicts(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft leere Liste wenn keine Konflikte vorliegen."""
|
||
with patch.object(
|
||
federation_manager.sync_engine, "detect_conflicts",
|
||
return_value=[],
|
||
):
|
||
conflicts = federation_manager.detect_conflicts("privat")
|
||
assert conflicts == []
|
||
|
||
def test_detect_conflicts_unknown_context(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft Fehler bei unbekanntem Kontext."""
|
||
conflicts = federation_manager.detect_conflicts("nonexistent")
|
||
assert len(conflicts) == 1
|
||
assert "nonexistent" in conflicts[0].details
|
||
|
||
def test_detect_conflicts_with_uncommitted_changes(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft Erkennung von uncommitted Änderungen."""
|
||
with patch.object(
|
||
federation_manager.sync_engine, "detect_conflicts",
|
||
return_value=[ConflictInfo(
|
||
file_path="privat/modified.py",
|
||
conflict_type="content",
|
||
source="monorepo",
|
||
details="Uncommitted lokale Änderungen erkannt",
|
||
)],
|
||
):
|
||
conflicts = federation_manager.detect_conflicts("privat")
|
||
assert len(conflicts) == 1
|
||
assert conflicts[0].file_path == "privat/modified.py"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: SubtreeSyncEngine
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestSubtreeSyncEngine:
|
||
"""Tests für die SubtreeSyncEngine."""
|
||
|
||
@pytest.fixture
|
||
def engine(self, monorepo_root: Path) -> SubtreeSyncEngine:
|
||
"""Erstellt eine SubtreeSyncEngine mit Test-Root."""
|
||
return SubtreeSyncEngine(monorepo_root=monorepo_root)
|
||
|
||
def test_subtree_pull_success(self, engine: SubtreeSyncEngine) -> None:
|
||
"""Prüft erfolgreichen subtree pull."""
|
||
with patch.object(engine, "_run_git") as mock_git:
|
||
mock_git.return_value = MagicMock(
|
||
stdout="Squash commit -- not updating HEAD\n1 file changed\n",
|
||
stderr="",
|
||
)
|
||
result = engine.subtree_pull(
|
||
remote="https://github.com/test/repo.git",
|
||
prefix="privat",
|
||
branch="main",
|
||
)
|
||
|
||
assert result.success is True
|
||
assert result.direction == "pull"
|
||
assert result.commits_synced >= 1
|
||
mock_git.assert_called_once_with([
|
||
"subtree", "pull",
|
||
"--prefix=privat",
|
||
"https://github.com/test/repo.git", "main",
|
||
"--squash",
|
||
])
|
||
|
||
def test_subtree_pull_merge_conflict(self, engine: SubtreeSyncEngine) -> None:
|
||
"""Prüft Behandlung von Merge-Konflikten beim Pull."""
|
||
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 privat/main.py"
|
||
error.stderr = ""
|
||
mock_git.side_effect = error
|
||
|
||
result = engine.subtree_pull(
|
||
remote="https://github.com/test/repo.git",
|
||
prefix="privat",
|
||
branch="main",
|
||
)
|
||
|
||
assert result.success is False
|
||
assert len(result.conflicts) >= 1
|
||
assert "privat/main.py" in result.conflicts[0].file_path
|
||
mock_abort.assert_called_once()
|
||
|
||
def test_subtree_pull_network_error(self, engine: SubtreeSyncEngine) -> None:
|
||
"""Prüft Fehlerbehandlung bei Netzwerkfehler."""
|
||
with patch.object(engine, "_run_git") as mock_git:
|
||
error = subprocess.CalledProcessError(128, "git")
|
||
error.stdout = ""
|
||
error.stderr = "fatal: Could not read from remote repository."
|
||
mock_git.side_effect = error
|
||
|
||
result = engine.subtree_pull(
|
||
remote="https://github.com/test/repo.git",
|
||
prefix="privat",
|
||
branch="main",
|
||
)
|
||
|
||
assert result.success is False
|
||
assert len(result.conflicts) >= 1
|
||
assert "fehlgeschlagen" in result.conflicts[0].details
|
||
|
||
def test_subtree_push_success(self, engine: SubtreeSyncEngine) -> None:
|
||
"""Prüft erfolgreichen subtree push."""
|
||
with patch.object(engine, "_run_git") as mock_git:
|
||
mock_git.return_value = MagicMock(
|
||
stdout="git push using: origin main\n",
|
||
stderr="",
|
||
)
|
||
result = engine.subtree_push(
|
||
remote="https://github.com/test/repo.git",
|
||
prefix="dhive",
|
||
branch="main",
|
||
)
|
||
|
||
assert result.success is True
|
||
assert result.direction == "push"
|
||
mock_git.assert_called_once_with([
|
||
"subtree", "push",
|
||
"--prefix=dhive",
|
||
"https://github.com/test/repo.git", "main",
|
||
])
|
||
|
||
def test_subtree_push_failure(self, engine: SubtreeSyncEngine) -> None:
|
||
"""Prüft Fehlerbehandlung beim Push."""
|
||
with patch.object(engine, "_run_git") as mock_git:
|
||
error = subprocess.CalledProcessError(1, "git")
|
||
error.stdout = ""
|
||
error.stderr = "error: failed to push some refs"
|
||
mock_git.side_effect = error
|
||
|
||
result = engine.subtree_push(
|
||
remote="https://github.com/test/repo.git",
|
||
prefix="dhive",
|
||
branch="main",
|
||
)
|
||
|
||
assert result.success is False
|
||
assert "fehlgeschlagen" in result.conflicts[0].details
|
||
|
||
|
||
def test_detect_conflicts_clean(self, engine: SubtreeSyncEngine) -> None:
|
||
"""Prüft leere Konfliktliste bei sauberem Zustand."""
|
||
with patch.object(engine, "_run_git") as mock_git:
|
||
mock_git.return_value = MagicMock(stdout="", stderr="")
|
||
conflicts = engine.detect_conflicts("privat")
|
||
assert conflicts == []
|
||
|
||
def test_detect_conflicts_uncommitted(self, engine: SubtreeSyncEngine) -> None:
|
||
"""Prüft Erkennung von uncommitted Änderungen."""
|
||
with patch.object(engine, "_run_git") as mock_git:
|
||
mock_git.return_value = MagicMock(
|
||
stdout=" M privat/src/app.py\n M privat/config.yaml\n",
|
||
stderr="",
|
||
)
|
||
conflicts = engine.detect_conflicts("privat")
|
||
assert len(conflicts) == 2
|
||
assert conflicts[0].source == "monorepo"
|
||
|
||
def test_detect_conflicts_git_error_returns_empty(
|
||
self, engine: SubtreeSyncEngine
|
||
) -> None:
|
||
"""Prüft dass Git-Fehler zu leerer Liste führen (kein Crash)."""
|
||
with patch.object(engine, "_run_git") as mock_git:
|
||
mock_git.side_effect = subprocess.CalledProcessError(1, "git")
|
||
conflicts = engine.detect_conflicts("privat")
|
||
assert conflicts == []
|
||
|
||
def test_preserve_history_success(self, engine: SubtreeSyncEngine) -> None:
|
||
"""Prüft Historie-Verifikation bei vorhandenen Commits."""
|
||
with patch.object(engine, "_run_git") as mock_git:
|
||
mock_git.return_value = MagicMock(
|
||
stdout="abc1234 Initial commit\ndef5678 Add feature\n",
|
||
stderr="",
|
||
)
|
||
assert engine.preserve_history(Path("privat")) is True
|
||
|
||
def test_preserve_history_empty(self, engine: SubtreeSyncEngine) -> None:
|
||
"""Prüft Historie-Verifikation bei leerer Historie."""
|
||
with patch.object(engine, "_run_git") as mock_git:
|
||
mock_git.return_value = MagicMock(stdout="", stderr="")
|
||
assert engine.preserve_history(Path("privat")) is False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: Onboarding und Hilfsmethoden
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestOnboarding:
|
||
"""Tests für das Mitglied-Onboarding."""
|
||
|
||
def test_onboard_member_success(self, federation_manager: FederationManager) -> None:
|
||
"""Prüft erfolgreiches Onboarding eines Team-Mitglieds."""
|
||
member = MemberInfo(name="Max Mustermann", email="max@dhive.io")
|
||
result = federation_manager.onboard_member("dhive", member)
|
||
assert result.success is True
|
||
assert result.member_name == "Max Mustermann"
|
||
assert "dhive" in result.team_repo_url
|
||
|
||
def test_onboard_member_invalid_context(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft Fehler bei ungültigem Kontext."""
|
||
member = MemberInfo(name="Test User", email="test@example.com")
|
||
result = federation_manager.onboard_member("nonexistent", member)
|
||
assert result.success is False
|
||
assert len(result.errors) > 0
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: mirror_shared (Task 15.3)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestMirrorShared:
|
||
"""Tests für mirror_shared – Spiegelung des shared-Bereichs ins Team_Repo.
|
||
|
||
Validates: Requirements 10.12
|
||
"""
|
||
|
||
def test_mirror_shared_single_file(
|
||
self, federation_manager: FederationManager, monorepo_root: Path
|
||
) -> None:
|
||
"""Prüft Spiegelung einer einzelnen Datei aus dem shared-Bereich."""
|
||
# Erstelle shared-Datei
|
||
shared_dir = monorepo_root / "shared" / "tools" / "common"
|
||
shared_dir.mkdir(parents=True)
|
||
script = shared_dir / "build.sh"
|
||
script.write_text("#!/bin/bash\necho hello", encoding="utf-8")
|
||
|
||
result = federation_manager.mirror_shared(
|
||
"privat", ["shared/tools/common/build.sh"]
|
||
)
|
||
|
||
assert result.success is True
|
||
assert result.context == "privat"
|
||
assert "shared/tools/common/build.sh" in result.mirrored_paths
|
||
assert result.errors == []
|
||
|
||
# Prüfe dass die Datei gespiegelt wurde
|
||
mirror_target = monorepo_root / "privat" / ".shared-mirror" / "shared" / "tools" / "common" / "build.sh"
|
||
assert mirror_target.exists()
|
||
assert mirror_target.read_text(encoding="utf-8") == "#!/bin/bash\necho hello"
|
||
|
||
def test_mirror_shared_directory(
|
||
self, federation_manager: FederationManager, monorepo_root: Path
|
||
) -> None:
|
||
"""Prüft Spiegelung eines kompletten Verzeichnisses."""
|
||
# Erstelle shared-Verzeichnis mit mehreren Dateien
|
||
shared_dir = monorepo_root / "shared" / "tools" / "common"
|
||
shared_dir.mkdir(parents=True)
|
||
(shared_dir / "script1.sh").write_text("script1", encoding="utf-8")
|
||
(shared_dir / "script2.sh").write_text("script2", encoding="utf-8")
|
||
sub = shared_dir / "sub"
|
||
sub.mkdir()
|
||
(sub / "nested.py").write_text("nested", encoding="utf-8")
|
||
|
||
result = federation_manager.mirror_shared("privat", ["shared/tools/common/"])
|
||
|
||
assert result.success is True
|
||
assert "shared/tools/common/" in result.mirrored_paths
|
||
|
||
# Prüfe dass alle Dateien rekursiv gespiegelt wurden
|
||
mirror_base = monorepo_root / "privat" / ".shared-mirror" / "shared" / "tools" / "common"
|
||
assert (mirror_base / "script1.sh").exists()
|
||
assert (mirror_base / "script2.sh").exists()
|
||
assert (mirror_base / "sub" / "nested.py").exists()
|
||
|
||
def test_mirror_shared_read_only_permissions(
|
||
self, federation_manager: FederationManager, monorepo_root: Path
|
||
) -> None:
|
||
"""Prüft dass gespiegelte Dateien read-only sind."""
|
||
shared_dir = monorepo_root / "shared" / "tools" / "common"
|
||
shared_dir.mkdir(parents=True)
|
||
(shared_dir / "config.yaml").write_text("key: value", encoding="utf-8")
|
||
|
||
federation_manager.mirror_shared("privat", ["shared/tools/common/config.yaml"])
|
||
|
||
mirror_file = monorepo_root / "privat" / ".shared-mirror" / "shared" / "tools" / "common" / "config.yaml"
|
||
assert mirror_file.exists()
|
||
# Prüfe read-only (0o444 = r--r--r--)
|
||
import stat
|
||
mode = mirror_file.stat().st_mode
|
||
assert not (mode & stat.S_IWUSR) # Kein Schreibzugriff für Owner
|
||
assert not (mode & stat.S_IWGRP) # Kein Schreibzugriff für Gruppe
|
||
assert not (mode & stat.S_IWOTH) # Kein Schreibzugriff für Other
|
||
|
||
def test_mirror_shared_invalid_context(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft Fehler bei ungültigem Kontext."""
|
||
result = federation_manager.mirror_shared("nonexistent", ["shared/tools/"])
|
||
assert result.success is False
|
||
assert len(result.errors) == 1
|
||
assert "nonexistent" in result.errors[0]
|
||
|
||
def test_mirror_shared_non_shared_path_rejected(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft dass Pfade außerhalb von shared/ abgelehnt werden."""
|
||
result = federation_manager.mirror_shared("privat", ["privat/secret.env"])
|
||
assert result.success is False
|
||
assert len(result.errors) == 1
|
||
assert "shared/" in result.errors[0]
|
||
|
||
def test_mirror_shared_nonexistent_source(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft Fehler bei nicht-existierendem Quellpfad."""
|
||
# dhive hat keine shared_mirror-Konfiguration, daher alle Pfade erlaubt
|
||
result = federation_manager.mirror_shared(
|
||
"dhive", ["shared/tools/nonexistent.sh"]
|
||
)
|
||
assert result.success is False
|
||
assert any("existiert nicht" in e for e in result.errors)
|
||
|
||
def test_mirror_shared_path_not_in_config(
|
||
self, federation_manager: FederationManager, monorepo_root: Path
|
||
) -> None:
|
||
"""Prüft dass Pfade außerhalb der erlaubten Mirror-Pfade abgelehnt werden."""
|
||
# privat hat nur "shared/tools/common/" als erlaubten Mirror-Pfad
|
||
# Versuche einen nicht-erlaubten Pfad zu spiegeln
|
||
other_dir = monorepo_root / "shared" / "powers" / "something"
|
||
other_dir.mkdir(parents=True)
|
||
(other_dir / "power.md").write_text("power", encoding="utf-8")
|
||
|
||
result = federation_manager.mirror_shared(
|
||
"privat", ["shared/powers/something/power.md"]
|
||
)
|
||
assert result.success is False
|
||
assert any("nicht in der Mirror-Konfiguration" in e for e in result.errors)
|
||
|
||
def test_mirror_shared_disabled_config(
|
||
self, tmp_path: Path, monorepo_root: Path
|
||
) -> None:
|
||
"""Prüft dass Spiegelung bei deaktivierter Konfiguration abgelehnt wird."""
|
||
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.git",
|
||
"branch": "main",
|
||
"sync_direction": "bidirectional",
|
||
"sync_frequency": "manual",
|
||
"shared_mirror": {
|
||
"enabled": False,
|
||
"paths": ["shared/tools/"],
|
||
"mode": "read-only",
|
||
},
|
||
},
|
||
],
|
||
}
|
||
with open(config_path, "w", encoding="utf-8") as f:
|
||
yaml.dump(data, f)
|
||
|
||
fm = FederationManager(config_path=config_path, monorepo_root=monorepo_root)
|
||
result = fm.mirror_shared("privat", ["shared/tools/something"])
|
||
assert result.success is False
|
||
assert any("deaktiviert" in e for e in result.errors)
|
||
|
||
def test_mirror_shared_no_mirror_config_allows_all(
|
||
self, federation_manager: FederationManager, monorepo_root: Path
|
||
) -> None:
|
||
"""Prüft dass ohne SharedMirrorConfig alle shared-Pfade erlaubt sind."""
|
||
# dhive hat keine shared_mirror-Konfiguration (None)
|
||
shared_dir = monorepo_root / "shared" / "tools"
|
||
shared_dir.mkdir(parents=True, exist_ok=True)
|
||
(shared_dir / "util.py").write_text("util", encoding="utf-8")
|
||
|
||
result = federation_manager.mirror_shared("dhive", ["shared/tools/util.py"])
|
||
assert result.success is True
|
||
assert "shared/tools/util.py" in result.mirrored_paths
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: resolve_conflict (Task 15.3)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestResolveConflict:
|
||
"""Tests für resolve_conflict – Konflikt-Auflösung mit team-wins Strategie.
|
||
|
||
Validates: Requirements 10.6, 10.7
|
||
"""
|
||
|
||
def test_resolve_conflict_team_wins_default(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft team-wins als Standard-Strategie (Single Source of Truth)."""
|
||
with patch.object(
|
||
federation_manager.sync_engine, "_run_git"
|
||
) as mock_git:
|
||
# Simuliere konfliktierende Dateien
|
||
mock_git.return_value = MagicMock(
|
||
stdout="privat/src/app.py\nprivat/config.yaml\n",
|
||
stderr="",
|
||
)
|
||
result = federation_manager.resolve_conflict("privat")
|
||
|
||
assert result.success is True
|
||
assert result.strategy == "team-wins"
|
||
assert len(result.resolved_files) == 2
|
||
assert "privat/src/app.py" in result.resolved_files
|
||
assert "privat/config.yaml" in result.resolved_files
|
||
|
||
# Prüfe dass --theirs verwendet wurde (team-wins)
|
||
calls = mock_git.call_args_list
|
||
checkout_calls = [c for c in calls if "checkout" in c[0][0]]
|
||
for call in checkout_calls:
|
||
assert "--theirs" in call[0][0]
|
||
|
||
def test_resolve_conflict_hub_wins(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft hub-wins Strategie (Monorepo hat Vorrang)."""
|
||
with patch.object(
|
||
federation_manager.sync_engine, "_run_git"
|
||
) as mock_git:
|
||
mock_git.return_value = MagicMock(
|
||
stdout="dhive/main.py\n",
|
||
stderr="",
|
||
)
|
||
result = federation_manager.resolve_conflict("dhive", strategy="hub-wins")
|
||
|
||
assert result.success is True
|
||
assert result.strategy == "hub-wins"
|
||
assert "dhive/main.py" in result.resolved_files
|
||
|
||
# Prüfe dass --ours verwendet wurde (hub-wins)
|
||
calls = mock_git.call_args_list
|
||
checkout_calls = [c for c in calls if "checkout" in c[0][0]]
|
||
for call in checkout_calls:
|
||
assert "--ours" in call[0][0]
|
||
|
||
def test_resolve_conflict_manual_strategy(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft dass manual-Strategie keine automatische Auflösung durchführt."""
|
||
result = federation_manager.resolve_conflict("privat", strategy="manual")
|
||
assert result.success is True
|
||
assert result.strategy == "manual"
|
||
assert result.resolved_files == []
|
||
|
||
def test_resolve_conflict_invalid_strategy(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft Fehler bei ungültiger Strategie."""
|
||
result = federation_manager.resolve_conflict("privat", strategy="invalid")
|
||
assert result.success is False
|
||
assert any("Ungültige Strategie" in e for e in result.errors)
|
||
|
||
def test_resolve_conflict_invalid_context(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft Fehler bei ungültigem Kontext."""
|
||
result = federation_manager.resolve_conflict("nonexistent")
|
||
assert result.success is False
|
||
assert any("nonexistent" in e for e in result.errors)
|
||
|
||
def test_resolve_conflict_no_conflicts_found(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft Erfolg wenn keine Konflikte vorhanden sind."""
|
||
with patch.object(
|
||
federation_manager.sync_engine, "_run_git"
|
||
) as mock_git:
|
||
mock_git.return_value = MagicMock(stdout="", stderr="")
|
||
result = federation_manager.resolve_conflict("privat")
|
||
|
||
assert result.success is True
|
||
assert result.resolved_files == []
|
||
|
||
def test_resolve_conflict_partial_failure(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft dass bei teilweisem Fehler beide Listen gefüllt werden."""
|
||
call_count = [0]
|
||
|
||
def side_effect(args: list[str]) -> MagicMock:
|
||
call_count[0] += 1
|
||
if args[0] == "diff":
|
||
return MagicMock(
|
||
stdout="privat/ok.py\nprivat/broken.py\n",
|
||
stderr="",
|
||
)
|
||
elif args[0] == "checkout" and "broken.py" in args[-1]:
|
||
error = subprocess.CalledProcessError(1, "git")
|
||
error.stdout = ""
|
||
error.stderr = "error: path not found"
|
||
raise error
|
||
return MagicMock(stdout="", stderr="")
|
||
|
||
with patch.object(
|
||
federation_manager.sync_engine, "_run_git", side_effect=side_effect
|
||
):
|
||
result = federation_manager.resolve_conflict("privat")
|
||
|
||
assert result.success is False
|
||
assert "privat/ok.py" in result.resolved_files
|
||
assert len(result.errors) == 1
|
||
assert "broken.py" in result.errors[0]
|
||
|
||
def test_resolve_conflict_git_diff_fails_uses_ls_files(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft Fallback auf ls-files wenn diff --diff-filter=U fehlschlägt."""
|
||
call_count = [0]
|
||
|
||
def side_effect(args: list[str]) -> MagicMock:
|
||
call_count[0] += 1
|
||
if args[0] == "diff" and "--diff-filter=U" in args:
|
||
error = subprocess.CalledProcessError(1, "git")
|
||
error.stdout = ""
|
||
error.stderr = "error"
|
||
raise error
|
||
elif args[0] == "ls-files":
|
||
return MagicMock(
|
||
stdout="100644 abc123 1\tprivat/conflict.py\n"
|
||
"100644 def456 2\tprivat/conflict.py\n"
|
||
"100644 ghi789 3\tprivat/conflict.py\n",
|
||
stderr="",
|
||
)
|
||
return MagicMock(stdout="", stderr="")
|
||
|
||
with patch.object(
|
||
federation_manager.sync_engine, "_run_git", side_effect=side_effect
|
||
):
|
||
result = federation_manager.resolve_conflict("privat")
|
||
|
||
assert result.success is True
|
||
assert "privat/conflict.py" in result.resolved_files
|
||
|
||
def test_resolve_conflict_only_resolves_own_context(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft dass nur Dateien des eigenen Kontexts aufgelöst werden."""
|
||
with patch.object(
|
||
federation_manager.sync_engine, "_run_git"
|
||
) as mock_git:
|
||
# Git meldet Konflikte in verschiedenen Kontexten
|
||
mock_git.return_value = MagicMock(
|
||
stdout="privat/my-file.py\ndhive/other-file.py\nbahn/third.py\n",
|
||
stderr="",
|
||
)
|
||
result = federation_manager.resolve_conflict("privat")
|
||
|
||
# Nur privat/ Dateien sollten aufgelöst werden
|
||
assert result.success is True
|
||
assert "privat/my-file.py" in result.resolved_files
|
||
assert "dhive/other-file.py" not in result.resolved_files
|
||
assert "bahn/third.py" not in result.resolved_files
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: onboard_member erweitert (Task 15.3)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestOnboardMemberExtended:
|
||
"""Erweiterte Tests für onboard_member – Zugang nur zum Team_Repo.
|
||
|
||
Validates: Requirements 10.9
|
||
"""
|
||
|
||
def test_onboard_member_no_monorepo_info_leaked(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft dass kein Hinweis auf das Monorepo im Ergebnis enthalten ist."""
|
||
member = MemberInfo(name="Lisa Schmidt", email="lisa@dhive.io", role="developer")
|
||
result = federation_manager.onboard_member("dhive", member)
|
||
|
||
assert result.success is True
|
||
# Die URL sollte nur die Team-Repo-URL sein, nicht die Monorepo-URL
|
||
assert "dhive" in result.team_repo_url
|
||
# Keine Referenz auf "monorepo", "hub", andere Kontexte
|
||
result_str = str(result)
|
||
assert "privat" not in result.team_repo_url
|
||
assert "bahn" not in result.team_repo_url
|
||
|
||
def test_onboard_member_empty_name_rejected(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft dass leerer Name abgelehnt wird."""
|
||
member = MemberInfo(name="", email="test@example.com")
|
||
result = federation_manager.onboard_member("privat", member)
|
||
assert result.success is False
|
||
assert any("Name" in e for e in result.errors)
|
||
|
||
def test_onboard_member_empty_email_rejected(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft dass leere E-Mail abgelehnt wird."""
|
||
member = MemberInfo(name="Test User", 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_member_whitespace_name_rejected(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft dass Name mit nur Whitespace abgelehnt wird."""
|
||
member = MemberInfo(name=" ", email="test@example.com")
|
||
result = federation_manager.onboard_member("privat", member)
|
||
assert result.success is False
|
||
|
||
def test_onboard_member_returns_correct_team_repo_url(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft dass die korrekte Team-Repo-URL zurückgegeben wird."""
|
||
member = MemberInfo(name="Hans Müller", email="hans@bahn.de", role="architect")
|
||
result = federation_manager.onboard_member("bahn", member)
|
||
|
||
assert result.success is True
|
||
assert result.team_repo_url == "https://gitlab.example.com/bahn-workspace.git"
|
||
assert result.member_name == "Hans Müller"
|
||
assert result.context == "bahn"
|
||
|
||
def test_onboard_member_custom_role(
|
||
self, federation_manager: FederationManager
|
||
) -> None:
|
||
"""Prüft Onboarding mit benutzerdefinierter Rolle."""
|
||
member = MemberInfo(name="Admin User", email="admin@dhive.io", role="admin")
|
||
result = federation_manager.onboard_member("dhive", member)
|
||
assert result.success is True
|
||
assert result.member_name == "Admin User"
|