Trixy7/tests/test_network.py

# -*- coding: utf-8 -*-
"""
Unit-Tests für trixy_core.network.
"""

import pytest
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path

from trixy_core.network.protocol import (
    TrixyProtocol, ProtocolFlags, ProtocolMessage,
    MAGIC, COMMAND_PING, COMMAND_PONG, COMMAND_NOOP, COMMAND_HELLO
)
from trixy_core.network.encryption import TrixyEncryption
from trixy_core.network.cmd.base import CommandMessage
from trixy_core.network.cmd.satellite import (
    SatelliteConnect, SatelliteDisconnect,
    SatelliteAccepted, SatelliteDenied
)
from trixy_core.network.cmd.wakeword import (
    WakewordDetected, WakewordSelected, WakewordAbort
)
from trixy_core.network.cmd.recording import (
    RecordingStart, RecordingStop, RecordingDone
)
from trixy_core.network.cmd.system import (
    SystemReboot, SystemUpdate, SystemStatus
)


class TestProtocolFlags:
    """Tests für ProtocolFlags."""

    def test_flag_values(self):
        """Testet die Flag-Werte."""
        assert ProtocolFlags.NONE == 0
        assert ProtocolFlags.COMPRESSED == 1
        assert ProtocolFlags.ENCRYPTED == 2
        assert ProtocolFlags.JSON == 4
        assert ProtocolFlags.ACK_REQUIRED == 8

    def test_flag_combination(self):
        """Testet Flag-Kombinationen."""
        flags = ProtocolFlags.COMPRESSED | ProtocolFlags.ENCRYPTED
        assert flags & ProtocolFlags.COMPRESSED
        assert flags & ProtocolFlags.ENCRYPTED
        assert not (flags & ProtocolFlags.JSON)


class TestProtocolMessage:
    """Tests für ProtocolMessage."""

    def test_message_creation(self):
        """Testet die Nachrichtenerstellung."""
        msg = ProtocolMessage(
            class_name="TestMessage",
            data={"key": "value"},
            flags=ProtocolFlags.JSON
        )
        assert msg.class_name == "TestMessage"
        assert msg.data == {"key": "value"}
        assert msg.flags == ProtocolFlags.JSON

    def test_header_size(self):
        """Testet die Header-Größe."""
        # Magic(4) + Version(2) + Timestamp(8) + Flags(4) + Checksum(16) + ClassNameLen(2)
        assert ProtocolMessage.HEADER_SIZE == 36


class TestTrixyProtocol:
    """Tests für TrixyProtocol."""

    @pytest.fixture
    def protocol(self):
        """Erstellt ein TrixyProtocol."""
        return TrixyProtocol()

    @pytest.fixture
    def protocol_with_encryption(self):
        """Erstellt ein TrixyProtocol mit Verschlüsselung."""
        encryption = TrixyEncryption()
        return TrixyProtocol(encryption=encryption)

    def test_serialize_deserialize_basic(self, protocol):
        """Testet einfache Serialisierung/Deserialisierung mit CommandMessage."""
        # Verwende CommandMessage (global definiert, pickle-fähig)
        original = SatelliteConnect(
            mac_address="AA:BB:CC:DD:EE:FF",
            room="test",
            alias="TestDevice"
        )
        serialized = protocol.serialize(original)
        deserialized = protocol.deserialize(serialized)

        assert deserialized.class_name == "SatelliteConnect"
        assert deserialized.data.mac_address == "AA:BB:CC:DD:EE:FF"
        assert deserialized.data.room == "test"

    def test_serialize_with_json_flag(self, protocol):
        """Testet Serialisierung mit JSON-Flag."""
        # Verwende einfaches Dictionary für JSON-Test
        original = {"mac_address": "BB:CC:DD:EE:FF:00", "room": "wohnzimmer"}
        serialized = protocol.serialize(original, ProtocolFlags.JSON)
        deserialized = protocol.deserialize(serialized)

        assert deserialized.flags & ProtocolFlags.JSON
        assert deserialized.data["mac_address"] == "BB:CC:DD:EE:FF:00"

    def test_serialize_binary_data(self, protocol):
        """Testet Serialisierung von Binärdaten."""
        binary = b"\x00\x01\x02\x03\x04"
        serialized = protocol.serialize(binary, ProtocolFlags.BINARY)
        deserialized = protocol.deserialize(serialized)

        assert deserialized.flags & ProtocolFlags.BINARY
        assert deserialized.data == binary

    def test_checksum_validation(self, protocol):
        """Testet die Prüfsummenvalidierung."""
        original = SystemStatus(status="running", uptime_seconds=100.0)
        serialized = protocol.serialize(original)

        # Manipuliere Daten
        corrupted = bytearray(serialized)
        corrupted[-1] ^= 0xFF

        with pytest.raises(ValueError, match="Prüfsummenfehler"):
            protocol.deserialize(bytes(corrupted))

    def test_hard_commands(self, protocol):
        """Testet Hard-coded Befehle."""
        assert protocol.create_ping() == COMMAND_PING
        assert protocol.create_pong() == COMMAND_PONG
        assert protocol.create_noop() == COMMAND_NOOP
        assert protocol.create_hello() == COMMAND_HELLO

    def test_is_hard_command(self, protocol):
        """Testet is_hard_command."""
        assert protocol.is_hard_command(COMMAND_PING)
        assert protocol.is_hard_command(COMMAND_PONG)
        assert not protocol.is_hard_command(b"NOTACMD!")

    def test_get_hard_command(self, protocol):
        """Testet get_hard_command."""
        assert protocol.get_hard_command(COMMAND_PING) == "TRXIPING"
        assert protocol.get_hard_command(b"NOTACMD!") is None

    def test_register_class(self, protocol):
        """Testet die Klassenregistrierung."""
        @dataclass
        class RegisteredClass:
            value: str = "registered"

        protocol.register_class(RegisteredClass)
        assert "RegisteredClass" in protocol._registered_classes

    def test_invalid_magic_raises_error(self, protocol):
        """Testet, dass ungültige Magic-Number Fehler wirft."""
        invalid_data = b"XXXX" + b"\x00" * 50

        with pytest.raises(ValueError, match="Ungültige Magic Number"):
            protocol.deserialize(invalid_data)

    def test_serialization_with_encryption(self, protocol_with_encryption):
        """Testet Serialisierung mit Verschlüsselung."""
        original = SatelliteConnect(
            mac_address="AA:BB:CC:DD:EE:FF",
            room="secret_room",
            alias="SecretDevice"
        )
        flags = ProtocolFlags.ENCRYPTED | ProtocolFlags.PICKLE
        serialized = protocol_with_encryption.serialize(original, flags)
        deserialized = protocol_with_encryption.deserialize(serialized)

        assert deserialized.data.room == "secret_room"


class TestTrixyEncryption:
    """Tests für TrixyEncryption."""

    @pytest.fixture
    def encryption(self):
        """Erstellt eine TrixyEncryption-Instanz."""
        return TrixyEncryption()

    def test_key_generation(self):
        """Testet die Schlüsselgenerierung."""
        key = TrixyEncryption.generate_key()
        assert len(key) == 32

    def test_initialization_with_key(self):
        """Testet Initialisierung mit Schlüssel."""
        key = TrixyEncryption.generate_key()
        enc = TrixyEncryption(key)
        assert enc.key == key

    def test_initialization_generates_key(self):
        """Testet, dass Initialisierung Schlüssel generiert."""
        enc = TrixyEncryption()
        assert len(enc.key) == 32

    def test_invalid_key_length_raises_error(self):
        """Testet, dass ungültige Schlüssellänge Fehler wirft."""
        with pytest.raises(ValueError, match="32 Bytes"):
            TrixyEncryption(b"short")

    def test_encrypt_decrypt(self, encryption):
        """Testet Verschlüsselung und Entschlüsselung."""
        plaintext = b"Hello, World!"
        ciphertext = encryption.encrypt(plaintext)
        decrypted = encryption.decrypt(ciphertext)

        assert decrypted == plaintext
        assert ciphertext != plaintext

    def test_encrypt_with_associated_data(self, encryption):
        """Testet Verschlüsselung mit zusätzlichen Daten."""
        plaintext = b"Secret message"
        aad = b"Additional authenticated data"

        ciphertext = encryption.encrypt(plaintext, aad)
        decrypted = encryption.decrypt(ciphertext, aad)

        assert decrypted == plaintext

    def test_decrypt_with_wrong_aad_fails(self, encryption):
        """Testet, dass Entschlüsselung mit falschen AAD fehlschlägt."""
        plaintext = b"Secret"
        ciphertext = encryption.encrypt(plaintext, b"correct")

        with pytest.raises(ValueError, match="Entschlüsselung fehlgeschlagen"):
            encryption.decrypt(ciphertext, b"wrong")

    def test_save_load_key(self, encryption, temp_dir):
        """Testet Speichern und Laden von Schlüsseln."""
        key_path = temp_dir / "encryption.key"
        encryption.save_key(key_path)

        loaded = TrixyEncryption.load_key(key_path)
        assert loaded.key == encryption.key

    def test_load_or_create_creates_new(self, temp_dir):
        """Testet load_or_create für neue Schlüssel."""
        key_path = temp_dir / "new.key"
        enc = TrixyEncryption.load_or_create(key_path)

        assert key_path.exists()
        assert len(enc.key) == 32

    def test_load_or_create_loads_existing(self, temp_dir):
        """Testet load_or_create für existierende Schlüssel."""
        key_path = temp_dir / "existing.key"
        original = TrixyEncryption()
        original.save_key(key_path)

        loaded = TrixyEncryption.load_or_create(key_path)
        assert loaded.key == original.key

    def test_load_nonexistent_raises_error(self, temp_dir):
        """Testet, dass Laden nicht existierender Datei Fehler wirft."""
        with pytest.raises(FileNotFoundError):
            TrixyEncryption.load_key(temp_dir / "nonexistent.key")


class TestCommandMessages:
    """Tests für Befehlsnachrichten."""

    def test_command_message_base(self):
        """Testet CommandMessage-Basis."""
        msg = CommandMessage()
        assert msg.message_id is not None
        assert isinstance(msg.timestamp, datetime)

    def test_satellite_connect(self):
        """Testet SatelliteConnect."""
        msg = SatelliteConnect(
            mac_address="AA:BB:CC:DD:EE:FF",
            room="wohnzimmer",
            alias="Echo 1",
            ip_address="192.168.1.100",
            version="1.0.0"
        )
        assert msg.mac_address == "AA:BB:CC:DD:EE:FF"
        assert msg.room == "wohnzimmer"

    def test_satellite_accepted(self):
        """Testet SatelliteAccepted."""
        msg = SatelliteAccepted(
            satellite_id="sat-1",
            audio_input_port=2102,
            audio_output_port=2103
        )
        assert msg.satellite_id == "sat-1"

    def test_satellite_denied(self):
        """Testet SatelliteDenied."""
        msg = SatelliteDenied(
            reason="Not registered",
            retry_allowed=True,
            retry_after_seconds=60
        )
        assert msg.reason == "Not registered"
        assert msg.retry_allowed is True

    def test_wakeword_detected(self):
        """Testet WakewordDetected."""
        msg = WakewordDetected(
            satellite_id="sat-1",
            wakeword_model="custom",
            confidence=0.95,
            audio_level=0.7
        )
        assert msg.confidence == 0.95

    def test_wakeword_selected(self):
        """Testet WakewordSelected."""
        msg = WakewordSelected(
            conversation_id="conv-123",
            start_streaming=True
        )
        assert msg.start_streaming is True

    def test_recording_start(self):
        """Testet RecordingStart."""
        msg = RecordingStart(
            conversation_id="conv-123",
            max_duration_seconds=60.0
        )
        assert msg.max_duration_seconds == 60.0

    def test_recording_done(self):
        """Testet RecordingDone."""
        msg = RecordingDone(
            conversation_id="conv-123",
            duration_seconds=5.5,
            total_samples=88000,
            ended_by="silence"
        )
        assert msg.ended_by == "silence"

    def test_system_status(self):
        """Testet SystemStatus."""
        msg = SystemStatus(
            status="running",
            uptime_seconds=3600.0,
            memory_usage_mb=256.0,
            active_satellites=3,
            loaded_plugins=["plugin1", "plugin2"]
        )
        assert msg.active_satellites == 3
        assert len(msg.loaded_plugins) == 2

    def test_command_message_to_dict(self):
        """Testet to_dict."""
        msg = SatelliteConnect(
            mac_address="AA:BB:CC:DD:EE:FF",
            room="test"
        )
        data = msg.to_dict()

        assert "mac_address" in data
        assert "room" in data
        assert data["mac_address"] == "AA:BB:CC:DD:EE:FF"