hms-mediaengine/_entwicklung/tests/unit/test_plugin_lifecycle.py

"""Unit-Tests Plugin-Lifecycle (PLAN.md §14.5, §14.6, §26.3, §29.2)."""

from __future__ import annotations

import json
from pathlib import Path

import pytest
from hms_plugin_sdk import (
    InvalidTransitionError,
    LifecycleState,
    PluginLifecycleManager,
)

REPO = Path(__file__).resolve().parents[2]
EXAMPLES = REPO / "plugins" / "examples"


def _valid_plugin_dir(tmp_path: Path) -> Path:
    """Kopiert das gültige Passthrough-Beispiel in ein Test-Verzeichnis."""
    target = tmp_path / "plugins"
    target.mkdir()
    for item in EXAMPLES.rglob("*"):
        if item.is_file():
            rel = item.relative_to(EXAMPLES)
            dest = target / rel
            dest.parent.mkdir(parents=True, exist_ok=True)
            dest.write_bytes(item.read_bytes())
    return target


# ---------- Discovery & Validierung (§14.5, §29.2) ----------


def test_discover_validates_example_plugins(tmp_path: Path) -> None:
    """Externe Beispielplugins werden ohne Kernänderung entdeckt und
    validiert (§29.2 / Gate 3 Vorbereitung)."""
    plugin_dir = _valid_plugin_dir(tmp_path)
    mgr = PluginLifecycleManager()
    errors = mgr.discover(plugin_dir)
    assert errors == []
    by_id = {r.plugin_id: r for r in mgr.all()}
    assert "com.hms.fx.example_passthrough" in by_id
    assert "com.hms.fx.gaussian_blur" in by_id
    assert all(r.state is LifecycleState.VALIDATED for r in by_id.values())


def test_discover_quarantines_invalid_manifest(tmp_path: Path) -> None:
    """Ungültiges Manifest → QUARANTINED mit dokumentierter Ursache (§3.5)."""
    plugin_dir = _valid_plugin_dir(tmp_path)
    broken = plugin_dir / "com.broken.plugin"
    broken.mkdir()
    (broken / "plugin.json").write_text(
        json.dumps({"schema_version": 99, "id": "com.broken.plugin"}),
        encoding="utf-8",
    )
    mgr = PluginLifecycleManager()
    errors = mgr.discover(plugin_dir)
    record = mgr.get("com.broken.plugin")
    assert record is not None
    assert record.state is LifecycleState.QUARANTINED
    assert record.last_error  # Ursache sichtbar, nicht geschluckt (§33)
    assert any("com.broken.plugin" in e for e in errors)
    # die gültigen Plugins bleiben unbeeinflusst
    assert mgr.get("com.hms.fx.example_passthrough").state is LifecycleState.VALIDATED


def test_discover_missing_dir_reports_error(tmp_path: Path) -> None:
    mgr = PluginLifecycleManager()
    errors = mgr.discover(tmp_path / "gibts-nicht")
    assert errors and "fehlt" in errors[0]


# ---------- Übergänge (§14.5) ----------


@pytest.fixture()
def manager_with_plugin(tmp_path: Path) -> PluginLifecycleManager:
    mgr = PluginLifecycleManager()
    mgr.discover(_valid_plugin_dir(tmp_path))
    return mgr


def _advance_to(
    mgr: PluginLifecycleManager, pid: str, *states: LifecycleState
) -> None:
    """Führt ein Plugin schrittweise durch die angegebenen Zustände."""
    for state in states:
        mgr.advance(pid, state)


_ACTIVE_CHAIN = (
    LifecycleState.INSTALLED,
    LifecycleState.ENABLED,
    LifecycleState.COMPILED,
    LifecycleState.ACTIVE,
)


def test_full_lifecycle_happy_path(manager_with_plugin: PluginLifecycleManager) -> None:
    """discovered→validated→installed→enabled→compiled→active (§14.5)."""
    mgr = manager_with_plugin
    pid = "com.hms.fx.example_passthrough"
    _advance_to(mgr, pid, *_ACTIVE_CHAIN)
    record = mgr.get(pid)
    assert record.state is LifecycleState.ACTIVE
    assert record.last_error is None
    assert mgr.active_plugins() == [record]


def test_skip_transition_rejected(manager_with_plugin: PluginLifecycleManager) -> None:
    """Sprünge im Graph sind Fehler: validated → active illegal."""
    mgr = manager_with_plugin
    with pytest.raises(InvalidTransitionError, match="nicht erlaubt"):
        mgr.advance("com.hms.fx.example_passthrough", LifecycleState.ACTIVE)


def test_backward_transition_rejected(manager_with_plugin: PluginLifecycleManager) -> None:
    """Rücksprünge sind ebenfalls illegal (active → enabled)."""
    mgr = manager_with_plugin
    pid = "com.hms.fx.example_passthrough"
    _advance_to(mgr, pid, *_ACTIVE_CHAIN)
    with pytest.raises(InvalidTransitionError):
        mgr.advance(pid, LifecycleState.ENABLED)


def test_disable_and_reenable_cycle(manager_with_plugin: PluginLifecycleManager) -> None:
    """active → disabled → enabled: Live-Betrieb bleibt steuerbar (§17.7)."""
    mgr = manager_with_plugin
    pid = "com.hms.fx.example_passthrough"
    _advance_to(mgr, pid, *_ACTIVE_CHAIN)
    mgr.advance(pid, LifecycleState.DISABLED)
    mgr.advance(pid, LifecycleState.ENABLED)  # Re-Enable: Rücksprung in Betrieb
    assert mgr.get(pid).state is LifecycleState.ENABLED


def test_unknown_plugin_rejected() -> None:
    mgr = PluginLifecycleManager()
    with pytest.raises(KeyError):
        mgr.advance("gibts-nicht", LifecycleState.INSTALLED)


def test_quarantine_from_active_documents_reason(
    manager_with_plugin: PluginLifecycleManager,
) -> None:
    """Fehlerfall im Betrieb: Quarantäne mit Ursache, Projekt bleibt
    nutzbar (§3.5, §12.2 Bypass)."""
    mgr = manager_with_plugin
    pid = "com.hms.fx.gaussian_blur"
    _advance_to(mgr, pid, *_ACTIVE_CHAIN)
    mgr.quarantine(pid, "shader compile failed")
    record = mgr.get(pid)
    assert record.state is LifecycleState.QUARANTINED
    assert record.last_error == "shader compile failed"
    assert mgr.by_state(LifecycleState.QUARANTINED) == [record]
    # quarantined hat keine Folgezustände: Neustart nur über Neuinstallation
    with pytest.raises(InvalidTransitionError):
        mgr.advance(pid, LifecycleState.ENABLED)


# ---------- Show-Lock (§26.3) ----------


def test_show_lock_blocks_installation(manager_with_plugin: PluginLifecycleManager) -> None:
    """§26.3: Plugininstallation im Show-Lock gesperrt."""
    mgr = manager_with_plugin
    mgr.set_show_lock(True)
    with pytest.raises(InvalidTransitionError, match="Show-Lock"):
        mgr.advance("com.hms.fx.example_passthrough", LifecycleState.INSTALLED)


def test_show_lock_allows_enable_of_installed(manager_with_plugin: PluginLifecycleManager) -> None:
    """§17.7/§26.3: Bereits installiertes Plugin darf im Live-Betrieb
    aktiviert werden (Layerparameter und Livefunktionen)."""
    mgr = manager_with_plugin
    pid = "com.hms.fx.example_passthrough"
    mgr.advance(pid, LifecycleState.INSTALLED)
    mgr.set_show_lock(True)
    _advance_to(
        mgr,
        pid,
        LifecycleState.ENABLED,
        LifecycleState.COMPILED,
        LifecycleState.ACTIVE,
    )
    assert mgr.get(pid).state is LifecycleState.ACTIVE


def test_show_lock_blocks_version_update(tmp_path: Path) -> None:
    """§26.3: Versionswechsel (= Update) im Show-Lock gesperrt."""
    plugin_dir = _valid_plugin_dir(tmp_path)
    mgr = PluginLifecycleManager()
    mgr.discover(plugin_dir)
    mgr.set_show_lock(True)
    # gleiche Plugin-ID mit neuer Version einschleusen
    manifest_path = plugin_dir / "com.hms.fx.example_passthrough" / "plugin.json"
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    manifest["version"] = "2.0.0"
    manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
    with pytest.raises(InvalidTransitionError, match="Versionswechsel"):
        mgr.discover(plugin_dir)


def test_version_update_without_lock_starts_new_cycle(tmp_path: Path) -> None:
    """Update ohne Show-Lock: Plugin startet einen neuen Zyklus in DISCOVERED."""
    plugin_dir = _valid_plugin_dir(tmp_path)
    mgr = PluginLifecycleManager()
    mgr.discover(plugin_dir)
    mgr.advance("com.hms.fx.example_passthrough", LifecycleState.INSTALLED)
    manifest_path = plugin_dir / "com.hms.fx.example_passthrough" / "plugin.json"
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    manifest["version"] = "2.0.0"
    manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
    mgr.discover(plugin_dir)  # Re-Scan erkennt neue Version
    record = mgr.get("com.hms.fx.example_passthrough")
    assert record.version == "2.0.0"
    assert record.state is LifecycleState.VALIDATED  # neuer Zyklus


# ---------- Backend-Kompatibilität (§14.5, §12.6) ----------


def test_compatible_backends_intersection(manager_with_plugin: PluginLifecycleManager) -> None:
    """Nur Backends mit Entrypoint UND supported_backends gelten (§12.6)."""
    mgr = manager_with_plugin
    backends = mgr.compatible_backends("com.hms.fx.example_passthrough")
    assert backends == frozenset({"d3d11", "gl", "gles"})
    assert mgr.compatible_backends("gibts-nicht") == frozenset()