986 lines
39 KiB
Python
986 lines
39 KiB
Python
"""Unit-Tests für die MigrationEngine.
|
|
|
|
Testet Konflikterkennung, Registry-Verwaltung und Migration via
|
|
gemockte Git-Operationen.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from monorepo.migration import (
|
|
MigrationConflict,
|
|
MigrationEngine,
|
|
MigrationError,
|
|
MigrationRegistryEntry,
|
|
MigrationResult,
|
|
ValidationResult,
|
|
)
|
|
from monorepo.models import MigrationPlan
|
|
|
|
|
|
@pytest.fixture
|
|
def monorepo_root(tmp_path: Path) -> Path:
|
|
"""Erstellt ein temporäres Monorepo-Root ohne git init."""
|
|
root = tmp_path / "monorepo"
|
|
root.mkdir()
|
|
(root / "privat").mkdir()
|
|
(root / "dhive").mkdir()
|
|
(root / "bahn").mkdir()
|
|
(root / "shared" / "config").mkdir(parents=True)
|
|
return root
|
|
|
|
|
|
@pytest.fixture
|
|
def engine(monorepo_root: Path) -> MigrationEngine:
|
|
"""Erstellt eine MigrationEngine-Instanz."""
|
|
return MigrationEngine(monorepo_root)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_plan() -> MigrationPlan:
|
|
"""Erstellt einen Beispiel-Migrationsplan."""
|
|
return MigrationPlan(
|
|
source_repo="https://github.com/example/my-project.git",
|
|
target_context="dhive",
|
|
target_name="my-project",
|
|
mode="subtree",
|
|
dependencies=[],
|
|
order=1,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Konflikterkennung: Namenskonvention
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestNamingConventionConflict:
|
|
"""Tests für Namenskonventions-Konflikterkennung."""
|
|
|
|
def test_valid_kebab_case_no_conflict(self, engine: MigrationEngine) -> None:
|
|
"""Gültiger kebab-case Name erzeugt keinen Namenskonflikt."""
|
|
plan = MigrationPlan(
|
|
source_repo="/tmp/source",
|
|
target_context="dhive",
|
|
target_name="valid-name",
|
|
mode="direct",
|
|
)
|
|
with patch.object(engine, "_get_remote_branches", return_value=[]):
|
|
conflicts = engine._detect_conflicts(plan)
|
|
naming = [c for c in conflicts if c.conflict_type == "naming_convention"]
|
|
assert len(naming) == 0
|
|
|
|
def test_uppercase_name_detected(self, engine: MigrationEngine) -> None:
|
|
"""Großbuchstaben im Namen werden als Verletzung erkannt."""
|
|
plan = MigrationPlan(
|
|
source_repo="/tmp/source",
|
|
target_context="dhive",
|
|
target_name="MyProject",
|
|
mode="direct",
|
|
)
|
|
with patch.object(engine, "_get_remote_branches", return_value=[]):
|
|
conflicts = engine._detect_conflicts(plan)
|
|
naming = [c for c in conflicts if c.conflict_type == "naming_convention"]
|
|
assert len(naming) == 1
|
|
assert "kebab-case" in naming[0].description
|
|
|
|
def test_underscore_name_detected(self, engine: MigrationEngine) -> None:
|
|
"""Unterstriche im Namen werden als Verletzung erkannt."""
|
|
plan = MigrationPlan(
|
|
source_repo="/tmp/source",
|
|
target_context="dhive",
|
|
target_name="my_project",
|
|
mode="direct",
|
|
)
|
|
with patch.object(engine, "_get_remote_branches", return_value=[]):
|
|
conflicts = engine._detect_conflicts(plan)
|
|
naming = [c for c in conflicts if c.conflict_type == "naming_convention"]
|
|
assert len(naming) == 1
|
|
|
|
def test_leading_hyphen_detected(self, engine: MigrationEngine) -> None:
|
|
"""Führender Bindestrich wird als Verletzung erkannt."""
|
|
plan = MigrationPlan(
|
|
source_repo="/tmp/source",
|
|
target_context="dhive",
|
|
target_name="-invalid",
|
|
mode="direct",
|
|
)
|
|
with patch.object(engine, "_get_remote_branches", return_value=[]):
|
|
conflicts = engine._detect_conflicts(plan)
|
|
naming = [c for c in conflicts if c.conflict_type == "naming_convention"]
|
|
assert len(naming) == 1
|
|
|
|
def test_single_char_name_detected(self, engine: MigrationEngine) -> None:
|
|
"""Einbuchstabiger Name ist zu kurz (min. 2 Zeichen)."""
|
|
plan = MigrationPlan(
|
|
source_repo="/tmp/source",
|
|
target_context="dhive",
|
|
target_name="x",
|
|
mode="direct",
|
|
)
|
|
with patch.object(engine, "_get_remote_branches", return_value=[]):
|
|
conflicts = engine._detect_conflicts(plan)
|
|
naming = [c for c in conflicts if c.conflict_type == "naming_convention"]
|
|
assert len(naming) == 1
|
|
|
|
def test_empty_name_detected(self, engine: MigrationEngine) -> None:
|
|
"""Leerer Name wird als Verletzung erkannt."""
|
|
plan = MigrationPlan(
|
|
source_repo="/tmp/source",
|
|
target_context="dhive",
|
|
target_name="",
|
|
mode="direct",
|
|
)
|
|
with patch.object(engine, "_get_remote_branches", return_value=[]):
|
|
conflicts = engine._detect_conflicts(plan)
|
|
naming = [c for c in conflicts if c.conflict_type == "naming_convention"]
|
|
assert len(naming) == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Konflikterkennung: Pfadkollision
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestPathCollisionConflict:
|
|
"""Tests für Pfadkollisions-Konflikterkennung."""
|
|
|
|
def test_no_collision_when_path_free(self, engine: MigrationEngine) -> None:
|
|
"""Kein Konflikt wenn Zielpfad nicht existiert."""
|
|
plan = MigrationPlan(
|
|
source_repo="/tmp/source",
|
|
target_context="dhive",
|
|
target_name="new-project",
|
|
mode="direct",
|
|
)
|
|
with patch.object(engine, "_get_remote_branches", return_value=[]):
|
|
conflicts = engine._detect_conflicts(plan)
|
|
path_conflicts = [c for c in conflicts if c.conflict_type == "path_collision"]
|
|
assert len(path_conflicts) == 0
|
|
|
|
def test_collision_detected_when_path_exists(
|
|
self, engine: MigrationEngine, monorepo_root: Path
|
|
) -> None:
|
|
"""Pfadkollision wird erkannt wenn Zielpfad bereits existiert."""
|
|
(monorepo_root / "dhive" / "existing-project").mkdir(parents=True)
|
|
plan = MigrationPlan(
|
|
source_repo="/tmp/source",
|
|
target_context="dhive",
|
|
target_name="existing-project",
|
|
mode="direct",
|
|
)
|
|
with patch.object(engine, "_get_remote_branches", return_value=[]):
|
|
conflicts = engine._detect_conflicts(plan)
|
|
path_conflicts = [c for c in conflicts if c.conflict_type == "path_collision"]
|
|
assert len(path_conflicts) == 1
|
|
assert "existiert bereits" in path_conflicts[0].description
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Konflikterkennung: Branch-Namenskonflikt
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestBranchConflict:
|
|
"""Tests für Branch-Namens-Konflikterkennung."""
|
|
|
|
def test_no_branch_conflict_for_main(self, engine: MigrationEngine) -> None:
|
|
"""main/master Branches werden nicht als Konflikte gewertet."""
|
|
plan = MigrationPlan(
|
|
source_repo="/tmp/source",
|
|
target_context="dhive",
|
|
target_name="new-project",
|
|
mode="direct",
|
|
)
|
|
with patch.object(engine, "_get_remote_branches", return_value=["main", "master"]):
|
|
with patch.object(engine, "_get_local_branches", return_value=["main", "master"]):
|
|
conflicts = engine._detect_conflicts(plan)
|
|
branch = [c for c in conflicts if c.conflict_type == "branch_name_conflict"]
|
|
assert len(branch) == 0
|
|
|
|
def test_branch_conflict_detected(self, engine: MigrationEngine) -> None:
|
|
"""Branch-Namenskonflikt wird erkannt bei gleichnamigem Branch."""
|
|
plan = MigrationPlan(
|
|
source_repo="/tmp/source",
|
|
target_context="dhive",
|
|
target_name="new-project",
|
|
mode="direct",
|
|
)
|
|
with patch.object(
|
|
engine, "_get_remote_branches", return_value=["main", "feature-x"]
|
|
):
|
|
with patch.object(
|
|
engine, "_get_local_branches", return_value=["main", "feature-x"]
|
|
):
|
|
conflicts = engine._detect_conflicts(plan)
|
|
branch = [c for c in conflicts if c.conflict_type == "branch_name_conflict"]
|
|
assert len(branch) == 1
|
|
assert "feature-x" in branch[0].description
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Migration pausiert bei Konflikten
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMigrationPauses:
|
|
"""Tests dass Migration bei Konflikten pausiert."""
|
|
|
|
def test_migration_pauses_on_naming_conflict(self, engine: MigrationEngine) -> None:
|
|
"""Migration pausiert und gibt Konflikte zurück bei Namensverstoß."""
|
|
plan = MigrationPlan(
|
|
source_repo="/tmp/source",
|
|
target_context="dhive",
|
|
target_name="Invalid_Name",
|
|
mode="direct",
|
|
)
|
|
with patch.object(engine, "_get_remote_branches", return_value=[]):
|
|
result = engine.migrate(plan)
|
|
assert result.success is False
|
|
assert len(result.conflicts) >= 1
|
|
assert "pausiert" in result.error_message
|
|
|
|
def test_migration_pauses_on_path_collision(
|
|
self, engine: MigrationEngine, monorepo_root: Path
|
|
) -> None:
|
|
"""Migration pausiert bei Pfadkollision."""
|
|
(monorepo_root / "dhive" / "existing-repo").mkdir()
|
|
plan = MigrationPlan(
|
|
source_repo="/tmp/source",
|
|
target_context="dhive",
|
|
target_name="existing-repo",
|
|
mode="direct",
|
|
)
|
|
with patch.object(engine, "_get_remote_branches", return_value=[]):
|
|
result = engine.migrate(plan)
|
|
assert result.success is False
|
|
path_conflicts = [c for c in result.conflicts if c.conflict_type == "path_collision"]
|
|
assert len(path_conflicts) == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Registry-Verwaltung
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMigrationRegistry:
|
|
"""Tests für die Migrations-Registry."""
|
|
|
|
def test_empty_registry_on_init(self, engine: MigrationEngine) -> None:
|
|
"""Neue Engine hat leere Registry."""
|
|
assert engine.get_registry() == []
|
|
|
|
def test_registry_updated_on_conflict(self, engine: MigrationEngine) -> None:
|
|
"""Registry wird mit 'failed' Status bei Konflikten aktualisiert."""
|
|
plan = MigrationPlan(
|
|
source_repo="/tmp/source",
|
|
target_context="dhive",
|
|
target_name="Invalid_Name",
|
|
mode="direct",
|
|
)
|
|
with patch.object(engine, "_get_remote_branches", return_value=[]):
|
|
engine.migrate(plan)
|
|
registry = engine.get_registry()
|
|
assert len(registry) == 1
|
|
assert registry[0].status == "failed"
|
|
|
|
def test_registry_persisted_to_yaml(
|
|
self, engine: MigrationEngine, monorepo_root: Path
|
|
) -> None:
|
|
"""Registry wird als YAML-Datei persistiert."""
|
|
plan = MigrationPlan(
|
|
source_repo="/tmp/source",
|
|
target_context="dhive",
|
|
target_name="Invalid_Name",
|
|
mode="direct",
|
|
)
|
|
with patch.object(engine, "_get_remote_branches", return_value=[]):
|
|
engine.migrate(plan)
|
|
assert engine.registry_path.exists()
|
|
content = engine.registry_path.read_text(encoding="utf-8")
|
|
assert "Invalid_Name" in content
|
|
assert "failed" in content
|
|
|
|
def test_registry_loaded_on_init(self, monorepo_root: Path) -> None:
|
|
"""Registry wird beim Initialisieren aus vorhandener Datei geladen."""
|
|
registry_path = monorepo_root / "shared" / "config" / "migrations.yaml"
|
|
registry_path.write_text(
|
|
"version: '1.0'\n"
|
|
"migrations:\n"
|
|
"- repo_name: old-project\n"
|
|
" source_repo: /tmp/old\n"
|
|
" target_context: privat\n"
|
|
" target_name: old-project\n"
|
|
" target_path: privat/old-project\n"
|
|
" mode: direct\n"
|
|
" status: completed\n"
|
|
" migrated_at: '2025-01-01T00:00:00'\n"
|
|
" backup_ref: pre-migration/old-project/20250101\n"
|
|
" commits_count: 42\n"
|
|
" branches: [main, develop]\n"
|
|
" tags: [v1.0, v2.0]\n",
|
|
encoding="utf-8",
|
|
)
|
|
engine = MigrationEngine(monorepo_root)
|
|
registry = engine.get_registry()
|
|
assert len(registry) == 1
|
|
assert registry[0].repo_name == "old-project"
|
|
assert registry[0].status == "completed"
|
|
assert registry[0].commits_count == 42
|
|
|
|
def test_is_migrated_returns_true_for_completed(
|
|
self, monorepo_root: Path
|
|
) -> None:
|
|
"""is_migrated gibt True für abgeschlossene Migrationen."""
|
|
registry_path = monorepo_root / "shared" / "config" / "migrations.yaml"
|
|
registry_path.write_text(
|
|
"version: '1.0'\n"
|
|
"migrations:\n"
|
|
"- repo_name: done-project\n"
|
|
" source_repo: /tmp/done\n"
|
|
" target_context: privat\n"
|
|
" target_name: done-project\n"
|
|
" target_path: privat/done-project\n"
|
|
" mode: direct\n"
|
|
" status: completed\n"
|
|
" migrated_at: '2025-01-01T00:00:00'\n",
|
|
encoding="utf-8",
|
|
)
|
|
engine = MigrationEngine(monorepo_root)
|
|
assert engine.is_migrated("done-project") is True
|
|
assert engine.is_migrated("unknown-project") is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Erfolgreiche Migration (mit gemocktem Git)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSuccessfulMigration:
|
|
"""Tests für eine erfolgreiche Migration mit gemockten Git-Operationen."""
|
|
|
|
def test_successful_migration_returns_result(
|
|
self, engine: MigrationEngine
|
|
) -> None:
|
|
"""Erfolgreiche Migration gibt MigrationResult mit success=True."""
|
|
plan = MigrationPlan(
|
|
source_repo="/tmp/source-repo",
|
|
target_context="dhive",
|
|
target_name="new-project",
|
|
mode="subtree",
|
|
)
|
|
mock_result = MagicMock()
|
|
mock_result.stdout = ""
|
|
mock_result.stderr = ""
|
|
|
|
with patch.object(engine, "_run_git", return_value=mock_result):
|
|
with patch.object(engine, "_get_remote_branches", return_value=[]):
|
|
with patch.object(engine, "_detect_main_branch", return_value="main"):
|
|
with patch.object(
|
|
engine, "_get_fetched_branches",
|
|
return_value=["main", "develop"],
|
|
):
|
|
with patch.object(
|
|
engine, "_get_fetched_tags",
|
|
return_value=["v1.0", "v2.0"],
|
|
):
|
|
with patch.object(
|
|
engine, "_count_commits_in_subtree",
|
|
return_value=25,
|
|
):
|
|
result = engine.migrate(plan)
|
|
|
|
assert result.success is True
|
|
assert result.repo_name == "new-project"
|
|
assert result.target_path == "dhive/new-project"
|
|
assert result.commits_migrated == 25
|
|
assert result.branches_migrated == ["main", "develop"]
|
|
assert result.tags_migrated == ["v1.0", "v2.0"]
|
|
|
|
def test_successful_migration_updates_registry(
|
|
self, engine: MigrationEngine
|
|
) -> None:
|
|
"""Erfolgreiche Migration aktualisiert Registry mit 'completed'."""
|
|
plan = MigrationPlan(
|
|
source_repo="/tmp/source-repo",
|
|
target_context="dhive",
|
|
target_name="new-project",
|
|
mode="subtree",
|
|
)
|
|
mock_result = MagicMock()
|
|
mock_result.stdout = ""
|
|
mock_result.stderr = ""
|
|
|
|
with patch.object(engine, "_run_git", return_value=mock_result):
|
|
with patch.object(engine, "_get_remote_branches", return_value=[]):
|
|
with patch.object(engine, "_detect_main_branch", return_value="main"):
|
|
with patch.object(
|
|
engine, "_get_fetched_branches", return_value=["main"]
|
|
):
|
|
with patch.object(engine, "_get_fetched_tags", return_value=[]):
|
|
with patch.object(
|
|
engine, "_count_commits_in_subtree", return_value=10
|
|
):
|
|
engine.migrate(plan)
|
|
|
|
assert engine.is_migrated("new-project") is True
|
|
registry = engine.get_registry()
|
|
assert any(
|
|
e.repo_name == "new-project" and e.status == "completed"
|
|
for e in registry
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fehlerbehandlung
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMigrationErrors:
|
|
"""Tests für Fehlerbehandlung bei der Migration."""
|
|
|
|
def test_git_error_returns_failed_result(self, engine: MigrationEngine) -> None:
|
|
"""Git-Fehler führt zu MigrationResult mit success=False."""
|
|
plan = MigrationPlan(
|
|
source_repo="/tmp/broken-repo",
|
|
target_context="dhive",
|
|
target_name="broken-project",
|
|
mode="subtree",
|
|
)
|
|
|
|
call_count = {"n": 0}
|
|
|
|
def mock_run_git(args: list[str], cwd=None): # type: ignore[no-untyped-def]
|
|
call_count["n"] += 1
|
|
# Let the backup tag call succeed, then fail on subtree add
|
|
if "subtree" in args:
|
|
err = subprocess.CalledProcessError(128, "git")
|
|
err.stdout = ""
|
|
err.stderr = "fatal: refusing to merge"
|
|
raise err
|
|
result = MagicMock()
|
|
result.stdout = ""
|
|
result.stderr = ""
|
|
return result
|
|
|
|
with patch.object(engine, "_run_git", side_effect=mock_run_git):
|
|
with patch.object(engine, "_get_remote_branches", return_value=[]):
|
|
with patch.object(engine, "_detect_main_branch", return_value="main"):
|
|
result = engine.migrate(plan)
|
|
|
|
assert result.success is False
|
|
assert result.error_message != ""
|
|
|
|
def test_failed_migration_in_registry(self, engine: MigrationEngine) -> None:
|
|
"""Fehlgeschlagene Migration wird mit 'failed' in Registry gespeichert."""
|
|
plan = MigrationPlan(
|
|
source_repo="/tmp/broken-repo",
|
|
target_context="dhive",
|
|
target_name="broken-project",
|
|
mode="subtree",
|
|
)
|
|
|
|
def mock_run_git(args: list[str], cwd=None): # type: ignore[no-untyped-def]
|
|
if "subtree" in args:
|
|
err = subprocess.CalledProcessError(1, "git")
|
|
err.stdout = ""
|
|
err.stderr = "fatal: refusing to merge"
|
|
raise err
|
|
result = MagicMock()
|
|
result.stdout = ""
|
|
result.stderr = ""
|
|
return result
|
|
|
|
with patch.object(engine, "_run_git", side_effect=mock_run_git):
|
|
with patch.object(engine, "_get_remote_branches", return_value=[]):
|
|
with patch.object(engine, "_detect_main_branch", return_value="main"):
|
|
engine.migrate(plan)
|
|
|
|
registry = engine.get_registry()
|
|
failed = [e for e in registry if e.status == "failed"]
|
|
assert len(failed) >= 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Inkrementelle Migration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestIncrementalMigration:
|
|
"""Tests für inkrementelle Migration."""
|
|
|
|
def _run_successful_migration(
|
|
self, engine: MigrationEngine, plan: MigrationPlan
|
|
) -> MigrationResult:
|
|
"""Hilfsmethode: führt eine erfolgreiche Migration mit Mocks durch."""
|
|
mock_result = MagicMock()
|
|
mock_result.stdout = ""
|
|
mock_result.stderr = ""
|
|
|
|
with patch.object(engine, "_run_git", return_value=mock_result):
|
|
with patch.object(engine, "_get_remote_branches", return_value=[]):
|
|
with patch.object(engine, "_detect_main_branch", return_value="main"):
|
|
with patch.object(
|
|
engine, "_get_fetched_branches", return_value=["main"]
|
|
):
|
|
with patch.object(engine, "_get_fetched_tags", return_value=[]):
|
|
with patch.object(
|
|
engine, "_count_commits_in_subtree", return_value=5
|
|
):
|
|
return engine.migrate(plan)
|
|
|
|
def test_independent_migrations_dont_interfere(
|
|
self, engine: MigrationEngine
|
|
) -> None:
|
|
"""Zwei unabhängige Migrationen beeinflussen sich nicht."""
|
|
plan_a = MigrationPlan(
|
|
source_repo="/tmp/repo-a",
|
|
target_context="dhive",
|
|
target_name="repo-a",
|
|
mode="subtree",
|
|
)
|
|
plan_b = MigrationPlan(
|
|
source_repo="/tmp/repo-b",
|
|
target_context="bahn",
|
|
target_name="repo-b",
|
|
mode="subtree",
|
|
)
|
|
|
|
result_a = self._run_successful_migration(engine, plan_a)
|
|
result_b = self._run_successful_migration(engine, plan_b)
|
|
|
|
assert result_a.success is True
|
|
assert result_b.success is True
|
|
assert engine.is_migrated("repo-a") is True
|
|
assert engine.is_migrated("repo-b") is True
|
|
|
|
def test_failed_migration_doesnt_affect_previous(
|
|
self, engine: MigrationEngine
|
|
) -> None:
|
|
"""Eine fehlgeschlagene Migration beeinträchtigt vorherige nicht."""
|
|
plan_a = MigrationPlan(
|
|
source_repo="/tmp/repo-a",
|
|
target_context="dhive",
|
|
target_name="repo-a",
|
|
mode="subtree",
|
|
)
|
|
self._run_successful_migration(engine, plan_a)
|
|
|
|
# Zweite Migration schlägt fehl (Namenskonvention)
|
|
plan_b = MigrationPlan(
|
|
source_repo="/tmp/repo-b",
|
|
target_context="bahn",
|
|
target_name="Invalid_B",
|
|
mode="subtree",
|
|
)
|
|
with patch.object(engine, "_get_remote_branches", return_value=[]):
|
|
engine.migrate(plan_b)
|
|
|
|
# Erste Migration ist weiterhin erfolgreich
|
|
assert engine.is_migrated("repo-a") is True
|
|
registry = engine.get_registry()
|
|
repo_a_entry = next(e for e in registry if e.repo_name == "repo-a")
|
|
assert repo_a_entry.status == "completed"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Backup-Referenz
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestBackupRef:
|
|
"""Tests für Backup-Referenz-Erstellung."""
|
|
|
|
def test_backup_tag_created_before_migration(
|
|
self, engine: MigrationEngine
|
|
) -> None:
|
|
"""Ein Backup-Tag wird vor der Migration erstellt."""
|
|
plan = MigrationPlan(
|
|
source_repo="/tmp/source",
|
|
target_context="dhive",
|
|
target_name="backup-test",
|
|
mode="subtree",
|
|
)
|
|
|
|
git_calls: list[list[str]] = []
|
|
mock_result = MagicMock()
|
|
mock_result.stdout = ""
|
|
mock_result.stderr = ""
|
|
|
|
def tracking_run_git(args: list[str], cwd=None): # type: ignore[no-untyped-def]
|
|
git_calls.append(args)
|
|
return mock_result
|
|
|
|
with patch.object(engine, "_run_git", side_effect=tracking_run_git):
|
|
with patch.object(engine, "_get_remote_branches", return_value=[]):
|
|
with patch.object(engine, "_detect_main_branch", return_value="main"):
|
|
with patch.object(
|
|
engine, "_get_fetched_branches", return_value=["main"]
|
|
):
|
|
with patch.object(engine, "_get_fetched_tags", return_value=[]):
|
|
with patch.object(
|
|
engine, "_count_commits_in_subtree", return_value=1
|
|
):
|
|
engine.migrate(plan)
|
|
|
|
# Find the tag call
|
|
tag_calls = [c for c in git_calls if c[0] == "tag" and "pre-migration" in c[1]]
|
|
assert len(tag_calls) == 1
|
|
assert "backup-test" in tag_calls[0][1]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Validierung (validate)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestValidation:
|
|
"""Tests für die Migrations-Validierung."""
|
|
|
|
def _setup_registry_with_completed_migration(
|
|
self, monorepo_root: Path, *, create_files: bool = True
|
|
) -> MigrationEngine:
|
|
"""Erstellt Engine mit einem abgeschlossenen Migrations-Eintrag."""
|
|
registry_path = monorepo_root / "shared" / "config" / "migrations.yaml"
|
|
registry_path.write_text(
|
|
"version: '1.0'\n"
|
|
"migrations:\n"
|
|
"- repo_name: my-project\n"
|
|
" source_repo: /tmp/source\n"
|
|
" target_context: dhive\n"
|
|
" target_name: my-project\n"
|
|
" target_path: dhive/my-project\n"
|
|
" mode: subtree\n"
|
|
" status: completed\n"
|
|
" migrated_at: '2025-01-01T00:00:00'\n"
|
|
" backup_ref: pre-migration/my-project/20250101-120000\n"
|
|
" commits_count: 10\n"
|
|
" branches: [main, develop]\n"
|
|
" tags: [v1.0, v2.0]\n",
|
|
encoding="utf-8",
|
|
)
|
|
if create_files:
|
|
project_dir = monorepo_root / "dhive" / "my-project"
|
|
project_dir.mkdir(parents=True, exist_ok=True)
|
|
(project_dir / "README.md").write_text("# My Project\n")
|
|
return MigrationEngine(monorepo_root)
|
|
|
|
def test_validate_unknown_repo(self, monorepo_root: Path) -> None:
|
|
"""Validierung eines unbekannten Repos gibt Fehler zurück."""
|
|
engine = MigrationEngine(monorepo_root)
|
|
result = engine.validate("unknown-repo")
|
|
assert result.success is False
|
|
assert "nicht in der Migrations-Registry" in result.error_message
|
|
|
|
def test_validate_successful_all_checks_pass(self, monorepo_root: Path) -> None:
|
|
"""Validierung besteht wenn alle Checks erfolgreich sind."""
|
|
engine = self._setup_registry_with_completed_migration(monorepo_root)
|
|
|
|
mock_git_result = MagicMock()
|
|
mock_git_result.stdout = "\n".join([f"commit{i}" for i in range(10)])
|
|
|
|
def mock_run_git(args: list[str], cwd=None): # type: ignore[no-untyped-def]
|
|
result = MagicMock()
|
|
if args[0] == "log":
|
|
result.stdout = "\n".join([f"abc{i} Commit msg {i}" for i in range(10)])
|
|
elif args[0] == "branch":
|
|
result.stdout = "main\ndevelop\n"
|
|
elif args[0] == "tag":
|
|
result.stdout = "v1.0\nv2.0\npre-migration/my-project/20250101-120000\n"
|
|
else:
|
|
result.stdout = ""
|
|
result.stderr = ""
|
|
return result
|
|
|
|
with patch.object(engine, "_run_git", side_effect=mock_run_git):
|
|
result = engine.validate("my-project")
|
|
|
|
assert result.success is True
|
|
assert result.checks["commits"] is True
|
|
assert result.checks["branches"] is True
|
|
assert result.checks["tags"] is True
|
|
assert result.checks["files"] is True
|
|
assert result.checks["tests"] is True
|
|
assert len(result.details) == 5
|
|
|
|
def test_validate_fails_when_commits_missing(self, monorepo_root: Path) -> None:
|
|
"""Validierung schlägt fehl wenn zu wenige Commits vorhanden."""
|
|
engine = self._setup_registry_with_completed_migration(monorepo_root)
|
|
|
|
def mock_run_git(args: list[str], cwd=None): # type: ignore[no-untyped-def]
|
|
result = MagicMock()
|
|
if args[0] == "log":
|
|
# Nur 3 Commits statt erwartet 10
|
|
result.stdout = "abc1 msg\nabc2 msg\nabc3 msg\n"
|
|
elif args[0] == "branch":
|
|
result.stdout = "main\ndevelop\n"
|
|
elif args[0] == "tag":
|
|
result.stdout = "v1.0\nv2.0\n"
|
|
else:
|
|
result.stdout = ""
|
|
result.stderr = ""
|
|
return result
|
|
|
|
with patch.object(engine, "_run_git", side_effect=mock_run_git):
|
|
result = engine.validate("my-project")
|
|
|
|
assert result.success is False
|
|
assert result.checks["commits"] is False
|
|
|
|
def test_validate_fails_when_tags_missing(self, monorepo_root: Path) -> None:
|
|
"""Validierung schlägt fehl wenn Tags fehlen."""
|
|
engine = self._setup_registry_with_completed_migration(monorepo_root)
|
|
|
|
def mock_run_git(args: list[str], cwd=None): # type: ignore[no-untyped-def]
|
|
result = MagicMock()
|
|
if args[0] == "log":
|
|
result.stdout = "\n".join([f"abc{i} msg" for i in range(10)])
|
|
elif args[0] == "branch":
|
|
result.stdout = "main\ndevelop\n"
|
|
elif args[0] == "tag":
|
|
# v2.0 fehlt
|
|
result.stdout = "v1.0\n"
|
|
else:
|
|
result.stdout = ""
|
|
result.stderr = ""
|
|
return result
|
|
|
|
with patch.object(engine, "_run_git", side_effect=mock_run_git):
|
|
result = engine.validate("my-project")
|
|
|
|
assert result.success is False
|
|
assert result.checks["tags"] is False
|
|
|
|
def test_validate_fails_when_directory_missing(self, monorepo_root: Path) -> None:
|
|
"""Validierung schlägt fehl wenn das Zielverzeichnis nicht existiert."""
|
|
engine = self._setup_registry_with_completed_migration(
|
|
monorepo_root, create_files=False
|
|
)
|
|
|
|
def mock_run_git(args: list[str], cwd=None): # type: ignore[no-untyped-def]
|
|
result = MagicMock()
|
|
if args[0] == "log":
|
|
result.stdout = "\n".join([f"abc{i} msg" for i in range(10)])
|
|
elif args[0] == "branch":
|
|
result.stdout = "main\ndevelop\n"
|
|
elif args[0] == "tag":
|
|
result.stdout = "v1.0\nv2.0\n"
|
|
else:
|
|
result.stdout = ""
|
|
result.stderr = ""
|
|
return result
|
|
|
|
with patch.object(engine, "_run_git", side_effect=mock_run_git):
|
|
result = engine.validate("my-project")
|
|
|
|
assert result.success is False
|
|
assert result.checks["files"] is False
|
|
|
|
def test_validate_no_branches_registered_passes(self, monorepo_root: Path) -> None:
|
|
"""Validierung besteht wenn keine Branches in der Registry registriert sind."""
|
|
registry_path = monorepo_root / "shared" / "config" / "migrations.yaml"
|
|
registry_path.write_text(
|
|
"version: '1.0'\n"
|
|
"migrations:\n"
|
|
"- repo_name: simple-project\n"
|
|
" source_repo: /tmp/source\n"
|
|
" target_context: privat\n"
|
|
" target_name: simple-project\n"
|
|
" target_path: privat/simple-project\n"
|
|
" mode: direct\n"
|
|
" status: completed\n"
|
|
" migrated_at: '2025-01-01T00:00:00'\n"
|
|
" commits_count: 5\n"
|
|
" branches: []\n"
|
|
" tags: []\n",
|
|
encoding="utf-8",
|
|
)
|
|
project_dir = monorepo_root / "privat" / "simple-project"
|
|
project_dir.mkdir(parents=True)
|
|
(project_dir / "main.py").write_text("print('hello')\n")
|
|
engine = MigrationEngine(monorepo_root)
|
|
|
|
def mock_run_git(args: list[str], cwd=None): # type: ignore[no-untyped-def]
|
|
result = MagicMock()
|
|
if args[0] == "log":
|
|
result.stdout = "\n".join([f"abc{i} msg" for i in range(5)])
|
|
else:
|
|
result.stdout = ""
|
|
result.stderr = ""
|
|
return result
|
|
|
|
with patch.object(engine, "_run_git", side_effect=mock_run_git):
|
|
result = engine.validate("simple-project")
|
|
|
|
assert result.success is True
|
|
assert result.checks["branches"] is True
|
|
assert result.checks["tags"] is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Rollback
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRollback:
|
|
"""Tests für den Migrations-Rollback."""
|
|
|
|
def _setup_engine_with_migration(
|
|
self, monorepo_root: Path, *, status: str = "completed"
|
|
) -> MigrationEngine:
|
|
"""Erstellt Engine mit einer Migration im gegebenen Status."""
|
|
registry_path = monorepo_root / "shared" / "config" / "migrations.yaml"
|
|
registry_path.write_text(
|
|
"version: '1.0'\n"
|
|
"migrations:\n"
|
|
f"- repo_name: rollback-project\n"
|
|
f" source_repo: /tmp/source\n"
|
|
f" target_context: dhive\n"
|
|
f" target_name: rollback-project\n"
|
|
f" target_path: dhive/rollback-project\n"
|
|
f" mode: subtree\n"
|
|
f" status: {status}\n"
|
|
f" migrated_at: '2025-01-01T00:00:00'\n"
|
|
f" backup_ref: pre-migration/rollback-project/20250101-120000\n"
|
|
f" commits_count: 15\n"
|
|
f" branches: [main]\n"
|
|
f" tags: [v1.0]\n",
|
|
encoding="utf-8",
|
|
)
|
|
# Erstelle das migrierte Verzeichnis
|
|
project_dir = monorepo_root / "dhive" / "rollback-project"
|
|
project_dir.mkdir(parents=True, exist_ok=True)
|
|
(project_dir / "README.md").write_text("# Rollback Project\n")
|
|
return MigrationEngine(monorepo_root)
|
|
|
|
def test_rollback_unknown_repo_raises_error(self, monorepo_root: Path) -> None:
|
|
"""Rollback eines unbekannten Repos wirft MigrationError."""
|
|
engine = MigrationEngine(monorepo_root)
|
|
with pytest.raises(MigrationError, match="nicht in der Migrations-Registry"):
|
|
engine.rollback("unknown-repo")
|
|
|
|
def test_rollback_already_rolled_back_raises_error(
|
|
self, monorepo_root: Path
|
|
) -> None:
|
|
"""Rollback eines bereits zurückgerollten Repos wirft MigrationError."""
|
|
engine = self._setup_engine_with_migration(monorepo_root, status="rolled_back")
|
|
with pytest.raises(MigrationError, match="bereits zurückgerollt"):
|
|
engine.rollback("rollback-project")
|
|
|
|
def test_rollback_removes_directory_and_updates_registry(
|
|
self, monorepo_root: Path
|
|
) -> None:
|
|
"""Rollback entfernt das Verzeichnis und setzt Registry auf 'rolled_back'."""
|
|
engine = self._setup_engine_with_migration(monorepo_root)
|
|
mock_result = MagicMock()
|
|
mock_result.stdout = ""
|
|
mock_result.stderr = ""
|
|
|
|
with patch.object(engine, "_run_git", return_value=mock_result):
|
|
engine.rollback("rollback-project")
|
|
|
|
registry = engine.get_registry()
|
|
entry = next(e for e in registry if e.repo_name == "rollback-project")
|
|
assert entry.status == "rolled_back"
|
|
|
|
def test_rollback_calls_git_rm_and_commit(self, monorepo_root: Path) -> None:
|
|
"""Rollback ruft git rm -rf und git commit auf."""
|
|
engine = self._setup_engine_with_migration(monorepo_root)
|
|
git_calls: list[list[str]] = []
|
|
mock_result = MagicMock()
|
|
mock_result.stdout = ""
|
|
mock_result.stderr = ""
|
|
|
|
def tracking_run_git(args: list[str], cwd=None): # type: ignore[no-untyped-def]
|
|
git_calls.append(args)
|
|
return mock_result
|
|
|
|
with patch.object(engine, "_run_git", side_effect=tracking_run_git):
|
|
engine.rollback("rollback-project")
|
|
|
|
# Prüfe git rm -rf
|
|
rm_calls = [c for c in git_calls if c[0] == "rm"]
|
|
assert len(rm_calls) == 1
|
|
assert "-rf" in rm_calls[0]
|
|
assert "dhive/rollback-project" in rm_calls[0]
|
|
|
|
# Prüfe git commit
|
|
commit_calls = [c for c in git_calls if c[0] == "commit"]
|
|
assert len(commit_calls) == 1
|
|
assert "Rollback" in commit_calls[0][2]
|
|
|
|
def test_rollback_does_not_affect_other_repos(self, monorepo_root: Path) -> None:
|
|
"""Rollback eines Repos beeinflusst andere migrierte Repos nicht."""
|
|
registry_path = monorepo_root / "shared" / "config" / "migrations.yaml"
|
|
registry_path.write_text(
|
|
"version: '1.0'\n"
|
|
"migrations:\n"
|
|
"- repo_name: rollback-project\n"
|
|
" source_repo: /tmp/source-a\n"
|
|
" target_context: dhive\n"
|
|
" target_name: rollback-project\n"
|
|
" target_path: dhive/rollback-project\n"
|
|
" mode: subtree\n"
|
|
" status: completed\n"
|
|
" migrated_at: '2025-01-01T00:00:00'\n"
|
|
" backup_ref: pre-migration/rollback-project/20250101-120000\n"
|
|
" commits_count: 10\n"
|
|
" branches: [main]\n"
|
|
" tags: []\n"
|
|
"- repo_name: other-project\n"
|
|
" source_repo: /tmp/source-b\n"
|
|
" target_context: bahn\n"
|
|
" target_name: other-project\n"
|
|
" target_path: bahn/other-project\n"
|
|
" mode: subtree\n"
|
|
" status: completed\n"
|
|
" migrated_at: '2025-01-01T00:00:00'\n"
|
|
" backup_ref: pre-migration/other-project/20250101-120000\n"
|
|
" commits_count: 20\n"
|
|
" branches: [main]\n"
|
|
" tags: [v2.0]\n",
|
|
encoding="utf-8",
|
|
)
|
|
# Erstelle Verzeichnisse
|
|
(monorepo_root / "dhive" / "rollback-project").mkdir(parents=True)
|
|
(monorepo_root / "dhive" / "rollback-project" / "file.txt").write_text("x")
|
|
(monorepo_root / "bahn" / "other-project").mkdir(parents=True)
|
|
(monorepo_root / "bahn" / "other-project" / "file.txt").write_text("y")
|
|
|
|
engine = MigrationEngine(monorepo_root)
|
|
mock_result = MagicMock()
|
|
mock_result.stdout = ""
|
|
mock_result.stderr = ""
|
|
|
|
with patch.object(engine, "_run_git", return_value=mock_result):
|
|
engine.rollback("rollback-project")
|
|
|
|
# other-project bleibt unverändert
|
|
registry = engine.get_registry()
|
|
other = next(e for e in registry if e.repo_name == "other-project")
|
|
assert other.status == "completed"
|
|
|
|
def test_rollback_git_rm_failure_raises_error(self, monorepo_root: Path) -> None:
|
|
"""Wenn git rm fehlschlägt, wird MigrationError geworfen."""
|
|
engine = self._setup_engine_with_migration(monorepo_root)
|
|
|
|
def failing_git(args: list[str], cwd=None): # type: ignore[no-untyped-def]
|
|
if args[0] == "rm":
|
|
err = subprocess.CalledProcessError(1, "git")
|
|
err.stdout = ""
|
|
err.stderr = "fatal: pathspec not found"
|
|
raise err
|
|
result = MagicMock()
|
|
result.stdout = ""
|
|
result.stderr = ""
|
|
return result
|
|
|
|
with patch.object(engine, "_run_git", side_effect=failing_git):
|
|
with pytest.raises(MigrationError, match="Rollback fehlgeschlagen"):
|
|
engine.rollback("rollback-project")
|