463 lines
16 KiB
Python
463 lines
16 KiB
Python
"""Tests für SecretEncryptionManager – Maschinenkontext-Verwaltung (Task 4.2).
|
||
|
||
Testet:
|
||
- get_context_key: Schlüsselabruf aus verschiedenen Quellen
|
||
- onboard_machine: Einrichtung neuer Maschinen
|
||
- resolve_merge: Merge-Konflikt-Auflösung auf verschlüsselter Ebene
|
||
- MachineContextManager: Laden der Konfiguration aus YAML
|
||
- Passwort-Manager-Integration
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
import yaml
|
||
|
||
from monorepo.encryption import (
|
||
MachineContextManager,
|
||
PasswordManagerError,
|
||
SecretEncryptionManager,
|
||
)
|
||
from monorepo.models import (
|
||
EncryptionKey,
|
||
MachineContext,
|
||
OnboardingResult,
|
||
PasswordManagerConfig,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fixtures
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.fixture
|
||
def tmp_monorepo(tmp_path: Path) -> Path:
|
||
"""Erstellt eine minimale Monorepo-Struktur in tmp_path."""
|
||
# Kontextordner
|
||
for ctx in ("privat", "dhive", "bahn", "shared"):
|
||
(tmp_path / ctx).mkdir()
|
||
|
||
# .git/git-crypt/keys Struktur simulieren
|
||
keys_dir = tmp_path / ".git" / "git-crypt" / "keys"
|
||
keys_dir.mkdir(parents=True)
|
||
(keys_dir / "default").mkdir()
|
||
(keys_dir / "privat").mkdir()
|
||
(keys_dir / "dhive").mkdir()
|
||
|
||
# Konfigurationsdateien
|
||
config_dir = tmp_path / "shared" / "config"
|
||
config_dir.mkdir(parents=True)
|
||
|
||
machine_context_yaml = {
|
||
"machine": {
|
||
"name": "test-rechner",
|
||
"description": "Testmaschine",
|
||
"authorized_contexts": ["privat", "dhive", "bahn"],
|
||
"key_source": "keyring",
|
||
"password_manager": {
|
||
"type": "bitwarden",
|
||
"vault": "monorepo-keys",
|
||
},
|
||
}
|
||
}
|
||
(config_dir / "machine-context.yaml").write_text(
|
||
yaml.dump(machine_context_yaml), encoding="utf-8"
|
||
)
|
||
|
||
return tmp_path
|
||
|
||
|
||
@pytest.fixture
|
||
def full_access_context() -> MachineContext:
|
||
"""MachineContext mit Zugriff auf alle Kontexte."""
|
||
return MachineContext(
|
||
name="andre-hauptrechner",
|
||
description="Voller Zugriff",
|
||
authorized_contexts=["privat", "dhive", "bahn"],
|
||
key_source="keyring",
|
||
password_manager=PasswordManagerConfig(
|
||
type="bitwarden",
|
||
vault="monorepo-keys",
|
||
),
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
def limited_context() -> MachineContext:
|
||
"""MachineContext mit eingeschränktem Zugriff (nur dhive)."""
|
||
return MachineContext(
|
||
name="dhive-laptop",
|
||
description="Nur dhive-Zugriff",
|
||
authorized_contexts=["dhive"],
|
||
key_source="keyring",
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
def manager(tmp_monorepo: Path, full_access_context: MachineContext) -> SecretEncryptionManager:
|
||
"""SecretEncryptionManager mit vollem Zugriff."""
|
||
return SecretEncryptionManager(tmp_monorepo, full_access_context)
|
||
|
||
|
||
@pytest.fixture
|
||
def limited_manager(
|
||
tmp_monorepo: Path, limited_context: MachineContext
|
||
) -> SecretEncryptionManager:
|
||
"""SecretEncryptionManager mit eingeschränktem Zugriff."""
|
||
return SecretEncryptionManager(tmp_monorepo, limited_context)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: get_context_key
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestGetContextKey:
|
||
"""Tests für die Methode get_context_key."""
|
||
|
||
def test_returns_key_for_authorized_context_from_keyring(
|
||
self, manager: SecretEncryptionManager
|
||
) -> None:
|
||
"""Autorisierter Kontext mit vorhandenem Schlüssel liefert EncryptionKey."""
|
||
key = manager.get_context_key("privat")
|
||
assert key is not None
|
||
assert isinstance(key, EncryptionKey)
|
||
assert key.context == "privat"
|
||
assert key.source == "keyring"
|
||
|
||
def test_returns_key_for_dhive_context(
|
||
self, manager: SecretEncryptionManager
|
||
) -> None:
|
||
"""Dhive-Kontext liefert ebenfalls einen Schlüssel."""
|
||
key = manager.get_context_key("dhive")
|
||
assert key is not None
|
||
assert key.context == "dhive"
|
||
|
||
def test_returns_none_for_unauthorized_context(
|
||
self, limited_manager: SecretEncryptionManager
|
||
) -> None:
|
||
"""Nicht-autorisierter Kontext liefert None."""
|
||
key = limited_manager.get_context_key("privat")
|
||
assert key is None
|
||
|
||
def test_returns_none_for_nonexistent_key(
|
||
self, tmp_monorepo: Path
|
||
) -> None:
|
||
"""Kontext ohne vorhandenen Schlüssel liefert None."""
|
||
# Kein .git/git-crypt/keys/bahn Verzeichnis
|
||
ctx = MachineContext(
|
||
name="test",
|
||
description="",
|
||
authorized_contexts=["bahn"],
|
||
key_source="keyring",
|
||
)
|
||
mgr = SecretEncryptionManager(tmp_monorepo, ctx)
|
||
# bahn existiert nicht im keys-Verzeichnis, aber default existiert
|
||
key = mgr.get_context_key("bahn")
|
||
# Fallback auf default key
|
||
assert key is not None
|
||
assert "default" in key.key_id
|
||
|
||
def test_file_source_returns_key_when_file_exists(
|
||
self, tmp_monorepo: Path
|
||
) -> None:
|
||
"""Dateiquelle liefert Schlüssel wenn Datei vorhanden."""
|
||
# Schlüsseldatei anlegen
|
||
key_dir = Path.home() / ".monorepo" / "keys"
|
||
key_dir.mkdir(parents=True, exist_ok=True)
|
||
key_file = key_dir / "privat.key"
|
||
key_file.write_text("test-key-content")
|
||
|
||
try:
|
||
ctx = MachineContext(
|
||
name="file-test",
|
||
description="",
|
||
authorized_contexts=["privat"],
|
||
key_source="file",
|
||
)
|
||
mgr = SecretEncryptionManager(tmp_monorepo, ctx)
|
||
key = mgr.get_context_key("privat")
|
||
assert key is not None
|
||
assert key.source == "file"
|
||
assert key.context == "privat"
|
||
finally:
|
||
key_file.unlink(missing_ok=True)
|
||
|
||
def test_file_source_returns_none_when_no_file(
|
||
self, tmp_monorepo: Path
|
||
) -> None:
|
||
"""Dateiquelle liefert None wenn keine Datei vorhanden."""
|
||
# Stelle sicher, dass keine Datei existiert
|
||
key_file = Path.home() / ".monorepo" / "keys" / "nonexistent-ctx.key"
|
||
key_file.unlink(missing_ok=True)
|
||
|
||
ctx = MachineContext(
|
||
name="file-test",
|
||
description="",
|
||
authorized_contexts=["nonexistent-ctx"],
|
||
key_source="file",
|
||
)
|
||
mgr = SecretEncryptionManager(tmp_monorepo, ctx)
|
||
key = mgr.get_context_key("nonexistent-ctx")
|
||
assert key is None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: onboard_machine
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestOnboardMachine:
|
||
"""Tests für die Methode onboard_machine."""
|
||
|
||
def test_successful_onboarding(self, manager: SecretEncryptionManager) -> None:
|
||
"""Erfolgreiches Onboarding installiert Schlüssel und aktualisiert Config."""
|
||
result = manager.onboard_machine("neuer-rechner", ["privat", "dhive"])
|
||
assert isinstance(result, OnboardingResult)
|
||
assert result.success is True
|
||
assert result.machine_name == "neuer-rechner"
|
||
assert "privat" in result.authorized_contexts
|
||
assert "dhive" in result.authorized_contexts
|
||
assert len(result.installed_keys) >= 1
|
||
assert len(result.errors) == 0
|
||
|
||
def test_onboarding_with_invalid_context(
|
||
self, manager: SecretEncryptionManager
|
||
) -> None:
|
||
"""Ungültiger Kontext erzeugt Fehler im OnboardingResult."""
|
||
result = manager.onboard_machine("test-rechner", ["invalid-context"])
|
||
assert result.success is False
|
||
assert "Ungültiger Kontext: 'invalid-context'" in result.errors
|
||
|
||
def test_onboarding_partial_success(
|
||
self, manager: SecretEncryptionManager
|
||
) -> None:
|
||
"""Mix aus gültigen und ungültigen Kontexten."""
|
||
result = manager.onboard_machine("mix-rechner", ["privat", "invalid"])
|
||
# privat sollte erfolgreich sein, invalid erzeugt Fehler
|
||
assert "Ungültiger Kontext: 'invalid'" in result.errors
|
||
assert "privat" in result.authorized_contexts or len(result.installed_keys) > 0
|
||
|
||
def test_onboarding_updates_config_file(
|
||
self, tmp_monorepo: Path, manager: SecretEncryptionManager
|
||
) -> None:
|
||
"""Onboarding aktualisiert die machine-context.yaml."""
|
||
manager.onboard_machine("config-test-rechner", ["privat"])
|
||
|
||
config_path = tmp_monorepo / "shared" / "config" / "machine-context.yaml"
|
||
assert config_path.exists()
|
||
|
||
with open(config_path, encoding="utf-8") as f:
|
||
data = yaml.safe_load(f)
|
||
|
||
assert data["machine"]["name"] == "config-test-rechner"
|
||
assert "privat" in data["machine"]["authorized_contexts"]
|
||
|
||
def test_onboarding_empty_contexts(
|
||
self, manager: SecretEncryptionManager
|
||
) -> None:
|
||
"""Leere Kontextliste führt zu Misserfolg."""
|
||
result = manager.onboard_machine("empty-rechner", [])
|
||
assert result.success is False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: resolve_merge
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestResolveMerge:
|
||
"""Tests für die Methode resolve_merge."""
|
||
|
||
def test_identical_versions_return_same(
|
||
self, tmp_monorepo: Path, manager: SecretEncryptionManager
|
||
) -> None:
|
||
"""Identische Versionen erzeugen keinen Konflikt."""
|
||
file_path = tmp_monorepo / "privat" / "test.env"
|
||
file_path.write_bytes(b"content")
|
||
|
||
result = manager.resolve_merge(file_path, b"same", b"same")
|
||
assert result == b"same"
|
||
|
||
def test_theirs_wins_for_authorized_context(
|
||
self, tmp_monorepo: Path, manager: SecretEncryptionManager
|
||
) -> None:
|
||
"""Bei autorisiertem Kontext gewinnt theirs (Team-Repo-Vorrang)."""
|
||
file_path = tmp_monorepo / "privat" / "secret.key"
|
||
file_path.write_bytes(b"dummy")
|
||
|
||
result = manager.resolve_merge(file_path, b"ours-version", b"theirs-version")
|
||
assert result == b"theirs-version"
|
||
|
||
def test_ours_wins_for_unauthorized_context(
|
||
self, tmp_monorepo: Path, limited_manager: SecretEncryptionManager
|
||
) -> None:
|
||
"""Bei nicht-autorisiertem Kontext wird ours beibehalten."""
|
||
file_path = tmp_monorepo / "privat" / "secret.key"
|
||
file_path.write_bytes(b"dummy")
|
||
|
||
result = limited_manager.resolve_merge(
|
||
file_path, b"ours-version", b"theirs-version"
|
||
)
|
||
assert result == b"ours-version"
|
||
|
||
def test_ours_wins_for_unknown_context(
|
||
self, tmp_monorepo: Path, manager: SecretEncryptionManager
|
||
) -> None:
|
||
"""Datei außerhalb der Kontextordner: ours beibehalten."""
|
||
file_path = tmp_monorepo / "unknown-dir" / "file.txt"
|
||
|
||
result = manager.resolve_merge(file_path, b"ours", b"theirs")
|
||
assert result == b"ours"
|
||
|
||
def test_binary_content_handled(
|
||
self, tmp_monorepo: Path, manager: SecretEncryptionManager
|
||
) -> None:
|
||
"""Binärdaten werden korrekt verarbeitet."""
|
||
file_path = tmp_monorepo / "dhive" / "cert.pem"
|
||
file_path.write_bytes(b"\x00\x01\x02")
|
||
|
||
ours = b"\x00\x01\x02\x03"
|
||
theirs = b"\x00\x01\x02\x04\x05"
|
||
result = manager.resolve_merge(file_path, ours, theirs)
|
||
assert result == theirs # dhive ist autorisiert
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: MachineContextManager
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestMachineContextManager:
|
||
"""Tests für den MachineContextManager."""
|
||
|
||
def test_load_valid_config(self, tmp_monorepo: Path) -> None:
|
||
"""Lädt gültige machine-context.yaml korrekt."""
|
||
mcm = MachineContextManager(tmp_monorepo)
|
||
ctx = mcm.load()
|
||
|
||
assert ctx.name == "test-rechner"
|
||
assert ctx.description == "Testmaschine"
|
||
assert "privat" in ctx.authorized_contexts
|
||
assert "dhive" in ctx.authorized_contexts
|
||
assert "bahn" in ctx.authorized_contexts
|
||
assert ctx.key_source == "keyring"
|
||
assert ctx.password_manager is not None
|
||
assert ctx.password_manager.type == "bitwarden"
|
||
assert ctx.password_manager.vault == "monorepo-keys"
|
||
|
||
def test_load_missing_config_raises(self, tmp_path: Path) -> None:
|
||
"""Fehlende Konfigurationsdatei wirft FileNotFoundError."""
|
||
mcm = MachineContextManager(tmp_path)
|
||
with pytest.raises(FileNotFoundError):
|
||
mcm.load()
|
||
|
||
def test_load_invalid_config_raises(self, tmp_path: Path) -> None:
|
||
"""Ungültiges YAML-Format wirft ValueError."""
|
||
config_dir = tmp_path / "shared" / "config"
|
||
config_dir.mkdir(parents=True)
|
||
(config_dir / "machine-context.yaml").write_text("invalid: true")
|
||
|
||
mcm = MachineContextManager(tmp_path)
|
||
with pytest.raises(ValueError, match="machine"):
|
||
mcm.load()
|
||
|
||
def test_machine_context_property_before_load_raises(
|
||
self, tmp_monorepo: Path
|
||
) -> None:
|
||
"""Zugriff auf machine_context vor load() wirft RuntimeError."""
|
||
mcm = MachineContextManager(tmp_monorepo)
|
||
with pytest.raises(RuntimeError, match="load"):
|
||
_ = mcm.machine_context
|
||
|
||
def test_machine_context_property_after_load(
|
||
self, tmp_monorepo: Path
|
||
) -> None:
|
||
"""Nach load() ist machine_context verfügbar."""
|
||
mcm = MachineContextManager(tmp_monorepo)
|
||
mcm.load()
|
||
ctx = mcm.machine_context
|
||
assert ctx.name == "test-rechner"
|
||
|
||
def test_create_encryption_manager(self, tmp_monorepo: Path) -> None:
|
||
"""create_encryption_manager erstellt funktionsfähigen Manager."""
|
||
mcm = MachineContextManager(tmp_monorepo)
|
||
mgr = mcm.create_encryption_manager()
|
||
assert isinstance(mgr, SecretEncryptionManager)
|
||
assert mgr.machine_context.name == "test-rechner"
|
||
|
||
def test_custom_config_path(self, tmp_path: Path) -> None:
|
||
"""Benutzerdefinierter config_path wird korrekt verwendet."""
|
||
custom_config = tmp_path / "custom-config.yaml"
|
||
custom_config.write_text(
|
||
yaml.dump({
|
||
"machine": {
|
||
"name": "custom-machine",
|
||
"description": "Custom",
|
||
"authorized_contexts": ["privat"],
|
||
"key_source": "file",
|
||
}
|
||
}),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
mcm = MachineContextManager(tmp_path, config_path=custom_config)
|
||
ctx = mcm.load()
|
||
assert ctx.name == "custom-machine"
|
||
assert ctx.key_source == "file"
|
||
|
||
def test_config_without_password_manager(self, tmp_path: Path) -> None:
|
||
"""Konfiguration ohne password_manager-Feld funktioniert."""
|
||
config_dir = tmp_path / "shared" / "config"
|
||
config_dir.mkdir(parents=True)
|
||
(config_dir / "machine-context.yaml").write_text(
|
||
yaml.dump({
|
||
"machine": {
|
||
"name": "no-pm-machine",
|
||
"description": "Ohne PM",
|
||
"authorized_contexts": ["bahn"],
|
||
"key_source": "keyring",
|
||
}
|
||
}),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
mcm = MachineContextManager(tmp_path)
|
||
ctx = mcm.load()
|
||
assert ctx.password_manager is None
|
||
assert ctx.key_source == "keyring"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: OnboardingResult Dataclass
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestOnboardingResultDataclass:
|
||
"""Tests für die OnboardingResult Dataclass."""
|
||
|
||
def test_default_values(self) -> None:
|
||
"""Standard-Initialwerte sind korrekt."""
|
||
result = OnboardingResult(success=True, machine_name="test")
|
||
assert result.success is True
|
||
assert result.machine_name == "test"
|
||
assert result.authorized_contexts == []
|
||
assert result.installed_keys == []
|
||
assert result.errors == []
|
||
|
||
def test_full_initialization(self) -> None:
|
||
"""Vollständige Initialisierung funktioniert."""
|
||
result = OnboardingResult(
|
||
success=True,
|
||
machine_name="full-test",
|
||
authorized_contexts=["privat", "dhive"],
|
||
installed_keys=["key-1", "key-2"],
|
||
errors=[],
|
||
)
|
||
assert result.machine_name == "full-test"
|
||
assert len(result.authorized_contexts) == 2
|
||
assert len(result.installed_keys) == 2
|