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

"""Unit-Tests Cluster: Nachrichten, Registry, Paarung, Discovery
(PLAN.md §6.3, §6.5, §29.1).

Keine Mocks für Logik; Health-Schwellen werden über injizierte Zeitstempel
bestimmt, Multicast selbst gehört zum Gate-1-LAN-Test (ADR-0009).
"""

from __future__ import annotations

import time
import uuid
from pathlib import Path

import pytest
from hms_cluster import (
    ClusterMessage,
    CommandStatus,
    CommandTracker,
    DuplicateNodeError,
    HealthThresholds,
    ManualNodeList,
    NodeCategory,
    NodeHealth,
    NodeRegistry,
    PairingStore,
    Scope,
    ServiceInfo,
    capability_digest,
    hash_token,
    identity_fingerprint,
)


def _uuid() -> str:
    return str(uuid.uuid4())


# ---------- ClusterMessage (§6.5) ----------


def test_message_requires_uuid_fields() -> None:
    with pytest.raises(ValueError, match="must be a UUID"):
        ClusterMessage(
            cluster_id="nope",
            node_id=_uuid(),
            command_id=_uuid(),
            sequence=0,
            project_revision=0,
        )


def test_message_rejects_negative_sequence_and_revision() -> None:
    args = dict(cluster_id=_uuid(), node_id=_uuid(), command_id=_uuid())
    with pytest.raises(ValueError, match="sequence"):
        ClusterMessage(**args, sequence=-1, project_revision=0)
    with pytest.raises(ValueError, match="project_revision"):
        ClusterMessage(**args, sequence=0, project_revision=-2)


def test_message_roundtrip_preserves_fields() -> None:
    msg = ClusterMessage(
        cluster_id=_uuid(),
        node_id=_uuid(),
        command_id=_uuid(),
        sequence=42,
        project_revision=7,
        execute_at_show_time_ns=123456,
        status=CommandStatus.ARMED,
        payload={"preset": "a"},
    )
    restored = ClusterMessage.from_dict(msg.to_dict())
    assert restored.cluster_id == msg.cluster_id
    assert restored.sequence == 42
    assert restored.project_revision == 7
    assert restored.execute_at_show_time_ns == 123456
    assert restored.status is CommandStatus.ARMED
    assert restored.trace_id == msg.trace_id


def test_command_tracker_idempotent_and_forward_only() -> None:
    tracker = CommandTracker()
    node = _uuid()
    cmd = _uuid()
    assert tracker.register(node, cmd) is True
    assert tracker.register(node, cmd) is False  # Duplikat → kein Re-Apply
    assert tracker.advance(node, cmd, CommandStatus.ARMED) is True
    assert tracker.advance(node, cmd, CommandStatus.ACCEPTED) is False  # Rückschritt
    assert tracker.advance(node, cmd, CommandStatus.EXECUTED) is True
    assert tracker.advance(node, cmd, CommandStatus.ARMED) is False  # abgeschlossen
    assert tracker.status(node, cmd) is CommandStatus.EXECUTED


def test_command_tracker_capacity_bound() -> None:
    tracker = CommandTracker(capacity=2)
    for _ in range(3):
        tracker.register(_uuid(), _uuid())
    assert len(tracker._states) <= 2  # kein unbeschränkter Cache (§33)


# ---------- NodeRegistry (§6.3, §6.5) ----------


def test_register_and_update_keeps_identity() -> None:
    reg = NodeRegistry()
    node_id = _uuid()
    reg.register(node_id, "Node A", roles=("RENDER_NODE",), api_port=8000)
    # IP-Wechsel: gleiche node_id, neuer Endpunkt
    reg.register(node_id, "Node A", api_port=8000, endpoint="10.0.0.9:8000")
    reg.register(node_id, "Node A", api_port=8000, endpoint="10.0.1.9:8000")
    entry = reg.get(node_id)
    assert entry is not None
    assert entry.last_known_endpoints[0] == "10.0.1.9:8000"  # zuletzt bekannt
    assert "10.0.0.9:8000" in entry.last_known_endpoints


def test_duplicate_node_id_with_other_identity_blocked() -> None:
    reg = NodeRegistry()
    node_id = _uuid()
    reg.register(node_id, "Node A")
    with pytest.raises(DuplicateNodeError):  # §6.3: Fehler, kein stilles Mischen
        reg.register(node_id, "Node B")


def test_incompatible_protocol_version_categorized() -> None:
    reg = NodeRegistry(protocol_version=1)
    entry = reg.register(_uuid(), "Alte Node", protocol_version=99)
    assert entry.category is NodeCategory.INCOMPATIBLE
    assert reg.by_category(NodeCategory.INCOMPATIBLE)[0].node_id == entry.node_id


def test_health_transitions_by_thresholds() -> None:
    thresholds = HealthThresholds(
        heartbeat_interval_ns=500_000_000,
        degraded_after_ns=2_000_000_000,
        stale_after_ns=5_000_000_000,
    )
    reg = NodeRegistry(thresholds=thresholds)
    node_id = _uuid()
    reg.register(node_id, "Node A")
    assert reg.evaluate_health(node_id) is NodeHealth.OFFLINE  # nie Heartbeat
    reg.record_heartbeat(node_id)
    assert reg.evaluate_health(node_id) is NodeHealth.ONLINE

    # verspäteter Heartbeat simulieren: letzten Heartbeat zurückdatieren
    entry = reg.get(node_id)
    entry.last_heartbeat_ns -= 3_000_000_000  # 3 s alt → degraded
    assert reg.evaluate_health(node_id) is NodeHealth.DEGRADED
    entry.last_heartbeat_ns -= 3_000_000_000  # 6 s alt → offline
    assert reg.evaluate_health(node_id) is NodeHealth.OFFLINE


def test_paired_node_offline_keeps_category_offline() -> None:
    reg = NodeRegistry()
    node_id = _uuid()
    reg.register(node_id, "Node A")
    reg.mark_paired(node_id)
    reg.record_heartbeat(node_id)
    entry = reg.get(node_id)
    assert entry is not None
    entry.last_heartbeat_ns -= 6_000_000_000  # deutlich zu alt
    reg.evaluate_health(node_id)
    assert reg.get(node_id).category is NodeCategory.OFFLINE  # nicht zurückgesetzt


# ---------- Pairing (§6.3, §27.1) ----------


def test_fingerprint_stable_and_distinct() -> None:
    node_id = _uuid()
    a = identity_fingerprint(node_id, "Node A")
    a2 = identity_fingerprint(node_id, "Node A")  # gleiche Eingabe → gleicher Wert
    assert a == a2
    b = identity_fingerprint(_uuid(), "Node B")  # andere Eingabe → anderer Wert
    assert a != b
    # Format: 8 Gruppen à 4 Hex-Zeichen
    groups = a.split(":")
    assert len(groups) == 8 and all(len(g) == 4 for g in groups)


def test_pairing_flow_pin_fingerprint_token() -> None:
    store = PairingStore()
    node_id = _uuid()
    pin = store.issue_pin(node_id)
    fingerprint = identity_fingerprint(node_id, "Node A")
    token = store.complete_pairing(
        node_id,
        entered_pin=pin.value,
        fingerprint_seen=fingerprint,
        expected_fingerprint=fingerprint,
        scopes=frozenset({Scope.READ, Scope.CONTROL}),
    )
    assert token
    assert store.verify(node_id, token, Scope.CONTROL)
    assert store.verify(node_id, token, Scope.READ)
    assert not store.verify(node_id, token, Scope.ADMIN)  # Scope fehlt


def test_wrong_pin_rejected_and_locks_after_attempts() -> None:
    store = PairingStore()
    node_id = _uuid()
    pin = store.issue_pin(node_id)
    fp = identity_fingerprint(node_id, "Node A")
    for _ in range(store.max_pin_attempts):
        with pytest.raises(PermissionError, match="PIN falsch"):
            store.complete_pairing(
                node_id,
                entered_pin="000000" if pin.value != "000000" else "000001",
                fingerprint_seen=fp,
                expected_fingerprint=fp,
                scopes=frozenset({Scope.READ}),
            )
    with pytest.raises(PermissionError, match="gesperrt"):
        store.complete_pairing(
            node_id,
            entered_pin=pin.value,  # jetzt sogar die richtige PIN
            fingerprint_seen=fp,
            expected_fingerprint=fp,
            scopes=frozenset({Scope.READ}),
        )


def test_expired_pin_rejected() -> None:
    store = PairingStore()
    node_id = _uuid()
    pin = store.issue_pin(node_id)
    fp = identity_fingerprint(node_id, "Node A")
    future = pin.expires_ns + 1
    with pytest.raises(PermissionError, match="abgelaufen"):
        store.complete_pairing(
            node_id,
            entered_pin=pin.value,
            fingerprint_seen=fp,
            expected_fingerprint=fp,
            scopes=frozenset({Scope.READ}),
            now_ns=future,
        )


def test_fingerprint_mismatch_rejected() -> None:
    store = PairingStore()
    node_id = _uuid()
    pin = store.issue_pin(node_id)
    with pytest.raises(PermissionError, match="Fingerprint"):
        store.complete_pairing(
            node_id,
            entered_pin=pin.value,
            fingerprint_seen=identity_fingerprint(node_id, "Andere Node"),
            expected_fingerprint=identity_fingerprint(node_id, "Node A"),
            scopes=frozenset({Scope.READ}),
        )


def test_token_revocation_immediate() -> None:
    store = PairingStore()
    node_id = _uuid()
    pin = store.issue_pin(node_id)
    fp = identity_fingerprint(node_id, "Node A")
    token = store.complete_pairing(
        node_id,
        entered_pin=pin.value,
        fingerprint_seen=fp,
        expected_fingerprint=fp,
        scopes=frozenset({Scope.CONTROL}),
    )
    assert store.verify(node_id, token, Scope.CONTROL)
    store.revoke(node_id)
    assert not store.verify(node_id, token, Scope.CONTROL)  # sofort wirkungslos


def test_token_expiry() -> None:
    store = PairingStore()
    node_id = _uuid()
    pin = store.issue_pin(node_id)
    fp = identity_fingerprint(node_id, "Node A")
    token = store.complete_pairing(
        node_id,
        entered_pin=pin.value,
        fingerprint_seen=fp,
        expected_fingerprint=fp,
        scopes=frozenset({Scope.READ}),
        token_ttl_s=0.01,
    )
    time.sleep(0.02)
    assert not store.verify(node_id, token, Scope.READ)  # abgelaufen


def test_token_hash_not_plaintext() -> None:
    store = PairingStore()
    node_id = _uuid()
    pin = store.issue_pin(node_id)
    fp = identity_fingerprint(node_id, "Node A")
    token = store.complete_pairing(
        node_id,
        entered_pin=pin.value,
        fingerprint_seen=fp,
        expected_fingerprint=fp,
        scopes=frozenset({Scope.READ}),
    )
    stored = store._tokens[node_id]  # interne Sichtprüfung (§27.1: nur Hash)
    assert stored.token_hash != token
    assert stored.token_hash == hash_token(token)


# ---------- Discovery (ADR-0009) ----------


def test_service_info_txt_roundtrip() -> None:
    caps = capability_digest({"tier": "DESKTOP_LITE", "outputs": 1})
    info = ServiceInfo(
        node_id=_uuid(),
        display_name="HMS Node A",
        port=8000,
        roles=("RENDER_NODE", "COORDINATOR"),
        capability_digest=caps,
    )
    restored = ServiceInfo.from_txt(info.instance_name, info.port, info.txt())
    assert restored.node_id == info.node_id
    assert restored.roles == info.roles
    assert restored.port == 8000
    assert restored.capability_digest == caps


def test_service_txt_contains_no_secrets() -> None:
    info = ServiceInfo(node_id=_uuid(), display_name="N", port=80, roles=("X",))
    blob = str(info.txt()).lower()
    for forbidden in ("token", "secret", "password", "key"):
        assert forbidden not in blob  # §27.1: Discovery ohne Vertrauliches


def test_service_txt_rejects_incomplete() -> None:
    with pytest.raises(ValueError, match="unvollstaendig"):
        ServiceInfo.from_txt("inst", 8000, {"proto": "1"})  # node/port fehlen


def test_instance_name_sanitized() -> None:
    info = ServiceInfo(node_id=_uuid(), display_name="Böse! Zeichen / 42", port=1, roles=())
    name = info.instance_name
    assert len(name) <= 63
    assert "!" not in name and "/" not in name


def test_capability_digest_stable() -> None:
    a = capability_digest({"b": 1, "a": 2})
    b = capability_digest({"a": 2, "b": 1})  # Reihenfolge egal
    assert a == b
    assert len(a) == 16
    assert capability_digest({"a": 3}) != a


def test_manual_node_list_roundtrip(tmp_path: Path) -> None:
    lst = ManualNodeList(path=tmp_path / "nodes.json")
    lst.add("10.0.0.9", 8000)
    lst.add("10.0.0.9", 8000)  # Duplikat wird ignoriert
    lst.add("node-b.local", 8001, node_id=_uuid())
    entries = lst.load()
    assert len(entries) == 2
    assert entries[0]["host"] == "10.0.0.9"
    assert entries[1]["node_id"]
    lst.remove("10.0.0.9", 8000)
    assert len(lst.load()) == 1


def test_manual_node_list_tolerates_garbage(tmp_path: Path) -> None:
    path = tmp_path / "nodes.json"
    path.write_text('{"broken": true}', encoding="utf-8")  # kein Liste-Objekt
    assert ManualNodeList(path=path).load() == []
    path.write_text('not json at all', encoding="utf-8")
    assert ManualNodeList(path=path).load() == []  # defekt → leer, nie Absturz