Trixy7/tests/test_network_keepalive.py

# -*- coding: utf-8 -*-
"""
Tests für Network KeepAlive.
"""

import asyncio
import pytest
from datetime import datetime, timedelta

from trixy_core.network.keepalive import (
    KeepAliveManager,
    KeepAliveSession,
    KeepAliveConfig,
    ConnectionState,
    ConnectionStats,
)


class TestKeepAliveConfig:
    """Tests für KeepAliveConfig."""

    def test_default_config(self):
        """Testet Standard-Konfiguration."""
        config = KeepAliveConfig()
        assert config.ping_interval == 30.0
        assert config.pong_timeout == 10.0
        assert config.max_missed_pongs == 3

    def test_custom_config(self):
        """Testet benutzerdefinierte Konfiguration."""
        config = KeepAliveConfig(
            ping_interval=60.0,
            pong_timeout=5.0,
            max_missed_pongs=5,
        )
        assert config.ping_interval == 60.0
        assert config.pong_timeout == 5.0
        assert config.max_missed_pongs == 5


class TestConnectionStats:
    """Tests für ConnectionStats."""

    def test_initial_stats(self):
        """Testet initiale Statistiken."""
        stats = ConnectionStats()
        assert stats.pings_sent == 0
        assert stats.pongs_received == 0
        assert stats.missed_pongs == 0
        assert stats.state == ConnectionState.CONNECTED

    def test_state_values(self):
        """Testet Connection-States."""
        assert ConnectionState.CONNECTED.name == "CONNECTED"
        assert ConnectionState.DISCONNECTED.name == "DISCONNECTED"
        assert ConnectionState.STALE.name == "STALE"


class TestKeepAliveSession:
    """Tests für KeepAliveSession."""

    @pytest.fixture
    def config(self):
        """Erstellt Test-Konfiguration."""
        return KeepAliveConfig(
            ping_interval=0.1,  # Schnelle Tests
            pong_timeout=0.05,
            max_missed_pongs=2,
            initial_delay=0.01,
        )

    @pytest.fixture
    def ping_func(self):
        """Erstellt Mock-Ping-Funktion."""
        calls = []

        async def ping():
            calls.append(datetime.now())

        ping.calls = calls
        return ping

    @pytest.mark.asyncio
    async def test_session_creation(self, config, ping_func):
        """Testet Session-Erstellung."""
        session = KeepAliveSession(
            connection_id="test-001",
            ping_func=ping_func,
            config=config,
        )
        assert session.connection_id == "test-001"
        assert session.state == ConnectionState.CONNECTED
        assert not session.is_running

    @pytest.mark.asyncio
    async def test_session_start_stop(self, config, ping_func):
        """Testet Start/Stop."""
        session = KeepAliveSession(
            connection_id="test-001",
            ping_func=ping_func,
            config=config,
        )

        await session.start()
        assert session.is_running

        await asyncio.sleep(0.05)

        await session.stop()
        assert not session.is_running

    @pytest.mark.asyncio
    async def test_handle_pong(self, config, ping_func):
        """Testet Pong-Verarbeitung."""
        session = KeepAliveSession(
            connection_id="test-001",
            ping_func=ping_func,
            config=config,
        )

        await session.start()
        await asyncio.sleep(0.02)

        # Simuliere Ping gesendet
        session._waiting_for_pong = True
        session._ping_sent_time = datetime.now()
        session.stats.pings_sent = 1

        # Pong empfangen
        session.handle_pong()

        assert session.stats.pongs_received == 1
        assert not session._waiting_for_pong

        await session.stop()

    @pytest.mark.asyncio
    async def test_rtt_calculation(self, config, ping_func):
        """Testet RTT-Berechnung."""
        session = KeepAliveSession(
            connection_id="test-001",
            ping_func=ping_func,
            config=config,
        )

        # Simuliere Ping/Pong
        session._waiting_for_pong = True
        session._ping_sent_time = datetime.now() - timedelta(milliseconds=50)

        session.handle_pong()

        assert session.stats.last_rtt_ms >= 50
        assert session.stats.min_rtt_ms >= 50
        assert session.stats.max_rtt_ms >= 50


class TestKeepAliveManager:
    """Tests für KeepAliveManager."""

    @pytest.fixture
    def config(self):
        """Erstellt Test-Konfiguration."""
        return KeepAliveConfig(
            ping_interval=0.1,
            pong_timeout=0.05,
            max_missed_pongs=2,
            initial_delay=0.01,
        )

    @pytest.fixture
    def manager(self, config):
        """Erstellt KeepAliveManager."""
        return KeepAliveManager(default_config=config)

    @pytest.mark.asyncio
    async def test_register_connection(self, manager):
        """Testet Verbindungs-Registrierung."""
        async def ping():
            pass

        session = await manager.register("conn-001", ping, auto_start=False)

        assert session is not None
        assert manager.session_count == 1
        assert "conn-001" in manager._sessions

    @pytest.mark.asyncio
    async def test_unregister_connection(self, manager):
        """Testet Verbindungs-Entfernung."""
        async def ping():
            pass

        await manager.register("conn-001", ping, auto_start=False)
        assert await manager.unregister("conn-001")
        assert manager.session_count == 0

    @pytest.mark.asyncio
    async def test_unregister_nonexistent(self, manager):
        """Testet Entfernung nicht existierender Verbindung."""
        assert not await manager.unregister("nonexistent")

    @pytest.mark.asyncio
    async def test_get_session(self, manager):
        """Testet Session-Abruf."""
        async def ping():
            pass

        await manager.register("conn-001", ping, auto_start=False)

        session = manager.get_session("conn-001")
        assert session is not None
        assert session.connection_id == "conn-001"

    @pytest.mark.asyncio
    async def test_handle_pong(self, manager):
        """Testet Pong-Verarbeitung über Manager."""
        async def ping():
            pass

        await manager.register("conn-001", ping, auto_start=False)

        # Simuliere Ping
        session = manager.get_session("conn-001")
        session._waiting_for_pong = True
        session._ping_sent_time = datetime.now()
        session.stats.pings_sent = 1

        manager.handle_pong("conn-001")

        assert session.stats.pongs_received == 1

    @pytest.mark.asyncio
    async def test_get_stats(self, manager):
        """Testet Statistik-Abruf."""
        async def ping():
            pass

        await manager.register("conn-001", ping, auto_start=False)

        stats = manager.get_stats("conn-001")
        assert stats is not None
        assert isinstance(stats, ConnectionStats)

    @pytest.mark.asyncio
    async def test_get_all_stats(self, manager):
        """Testet Abruf aller Statistiken."""
        async def ping():
            pass

        await manager.register("conn-001", ping, auto_start=False)
        await manager.register("conn-002", ping, auto_start=False)

        all_stats = manager.get_all_stats()
        assert len(all_stats) == 2
        assert "conn-001" in all_stats
        assert "conn-002" in all_stats

    @pytest.mark.asyncio
    async def test_get_summary(self, manager):
        """Testet Zusammenfassung."""
        async def ping():
            pass

        await manager.register("conn-001", ping, auto_start=False)

        summary = manager.get_summary()
        assert "total_sessions" in summary
        assert "active_sessions" in summary
        assert "states" in summary

    @pytest.mark.asyncio
    async def test_start_stop_all(self, manager):
        """Testet Start/Stop aller Sessions."""
        async def ping():
            pass

        await manager.register("conn-001", ping, auto_start=False)
        await manager.register("conn-002", ping, auto_start=False)

        assert manager.active_count == 0

        await manager.start_all()
        assert manager.active_count == 2

        await manager.stop_all()
        assert manager.active_count == 0

    @pytest.mark.asyncio
    async def test_timeout_callback(self, manager):
        """Testet Timeout-Callback."""
        timeout_received = []

        async def on_timeout(conn_id):
            timeout_received.append(conn_id)

        manager.on_timeout(on_timeout)

        async def ping():
            pass

        await manager.register("conn-001", ping, auto_start=False)

        # Simuliere Timeout
        await manager._global_timeout_handler("conn-001")

        assert "conn-001" in timeout_received

    @pytest.mark.asyncio
    async def test_disconnect_callback(self, manager):
        """Testet Disconnect-Callback."""
        disconnect_received = []

        async def on_disconnect(conn_id):
            disconnect_received.append(conn_id)

        manager.on_disconnect(on_disconnect)

        async def ping():
            pass

        await manager.register("conn-001", ping, auto_start=False)

        # Simuliere Disconnect
        await manager._global_disconnect_handler("conn-001")

        assert "conn-001" in disconnect_received

    @pytest.mark.asyncio
    async def test_get_stale_connections(self, manager):
        """Testet Erkennung von Stale-Verbindungen."""
        async def ping():
            pass

        await manager.register("conn-001", ping, auto_start=False)
        session = manager.get_session("conn-001")
        session.stats.state = ConnectionState.STALE

        stale = manager.get_stale_connections()
        assert "conn-001" in stale

    @pytest.mark.asyncio
    async def test_get_disconnected(self, manager):
        """Testet Erkennung getrennter Verbindungen."""
        async def ping():
            pass

        await manager.register("conn-001", ping, auto_start=False)
        session = manager.get_session("conn-001")
        session.stats.state = ConnectionState.DISCONNECTED

        disconnected = manager.get_disconnected()
        assert "conn-001" in disconnected

    @pytest.mark.asyncio
    async def test_cleanup(self, manager):
        """Testet Cleanup."""
        async def ping():
            pass

        await manager.register("conn-001", ping, auto_start=False)
        await manager.register("conn-002", ping, auto_start=False)

        # Markiere eine als disconnected
        session = manager.get_session("conn-001")
        session.stats.state = ConnectionState.DISCONNECTED

        removed = await manager.cleanup()
        assert removed == 1
        assert manager.session_count == 1