Trixy7/tests/test_connection_stability.py

# -*- coding: utf-8 -*-
"""
Tests für Verbindungsstabilität.

Testet, dass Client-Server-Verbindungen über längere Zeit stabil bleiben.
"""

import asyncio
import pytest
from datetime import datetime
from unittest.mock import AsyncMock, MagicMock, patch

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


class TestKeepAliveStability:
    """Tests für KeepAlive-Stabilität."""

    @pytest.fixture
    def keepalive_config(self):
        """Schnelle KeepAlive-Konfiguration für Tests."""
        return KeepAliveConfig(
            ping_interval=1.0,      # 1 Sekunde zwischen Pings
            pong_timeout=0.5,       # 0.5 Sekunden Timeout
            max_missed_pongs=3,
            initial_delay=0.1,      # Kurze initiale Verzögerung
            jitter=0.0,             # Kein Jitter für deterministische Tests
            adaptive=False,         # Kein adaptives Intervall
        )

    @pytest.mark.asyncio
    async def test_keepalive_session_basic(self, keepalive_config):
        """Test: KeepAlive-Session funktioniert grundlegend."""
        ping_count = 0
        pong_received = asyncio.Event()

        async def mock_ping():
            nonlocal ping_count
            ping_count += 1
            # Simuliere Pong-Empfang nach kurzem Delay
            await asyncio.sleep(0.1)
            pong_received.set()

        session = KeepAliveSession(
            connection_id="test-1",
            ping_func=mock_ping,
            config=keepalive_config,
        )

        await session.start()

        # Warte auf ersten Ping
        await asyncio.sleep(0.3)

        # Simuliere Pong
        session.handle_pong()

        # Warte weitere Zeit
        await asyncio.sleep(1.5)

        # Simuliere weitere Pongs
        session.handle_pong()
        await asyncio.sleep(0.5)

        await session.stop()

        assert ping_count >= 1
        assert session.stats.missed_pongs < 3
        assert session.state != ConnectionState.DISCONNECTED

    @pytest.mark.asyncio
    async def test_keepalive_survives_without_disconnect(self, keepalive_config):
        """Test: Verbindung bleibt bei regelmäßigen Pongs bestehen."""
        disconnect_called = False

        async def mock_on_disconnect(conn_id):
            nonlocal disconnect_called
            disconnect_called = True

        ping_count = 0

        async def mock_ping():
            nonlocal ping_count
            ping_count += 1

        session = KeepAliveSession(
            connection_id="test-stable",
            ping_func=mock_ping,
            config=keepalive_config,
            on_disconnect=mock_on_disconnect,
        )

        await session.start()

        # Lasse Session 3 Sekunden laufen und beantworte alle Pings
        start_time = asyncio.get_event_loop().time()
        while asyncio.get_event_loop().time() - start_time < 3.0:
            await asyncio.sleep(0.2)
            # Simuliere erfolgreichen Pong
            if session._waiting_for_pong:
                session.handle_pong()

        await session.stop()

        # Verbindung sollte NICHT getrennt worden sein
        assert not disconnect_called, "Verbindung wurde unerwartet getrennt"
        assert session.state != ConnectionState.DISCONNECTED
        assert ping_count >= 2, f"Erwartete mindestens 2 Pings, bekam {ping_count}"

    @pytest.mark.asyncio
    async def test_keepalive_disconnects_on_timeout(self, keepalive_config):
        """Test: Verbindung wird bei ausbleibenden Pongs getrennt."""
        disconnect_called = False
        disconnect_reason = None

        async def mock_on_disconnect(conn_id):
            nonlocal disconnect_called, disconnect_reason
            disconnect_called = True
            disconnect_reason = conn_id

        async def mock_ping():
            # Sende Ping, aber kein Pong kommt zurück
            pass

        session = KeepAliveSession(
            connection_id="test-timeout",
            ping_func=mock_ping,
            config=keepalive_config,
            on_disconnect=mock_on_disconnect,
        )

        await session.start()

        # Warte lange genug für 3 Timeouts
        # initial_delay(0.1) + 3 * (ping_interval(1.0) + pong_timeout(0.5)) = 4.6s
        await asyncio.sleep(5.0)

        await session.stop()

        # Verbindung sollte getrennt worden sein
        assert disconnect_called, "Disconnect wurde nicht aufgerufen"
        assert session.stats.missed_pongs >= 3

    @pytest.mark.asyncio
    async def test_keepalive_manager_multiple_sessions(self, keepalive_config):
        """Test: Manager verwaltet mehrere Sessions korrekt."""
        manager = KeepAliveManager(default_config=keepalive_config)

        sessions = []
        for i in range(3):
            async def make_ping(idx=i):
                pass

            session = await manager.register(
                f"session-{i}",
                make_ping,
                auto_start=True,
            )
            sessions.append(session)

        assert manager.session_count == 3
        assert manager.active_count == 3

        # Simuliere Pongs für alle Sessions
        for i in range(3):
            manager.handle_pong(f"session-{i}")

        await asyncio.sleep(0.5)

        # Alle Sessions sollten noch aktiv sein
        for i in range(3):
            stats = manager.get_stats(f"session-{i}")
            assert stats is not None
            assert stats.state == ConnectionState.CONNECTED

        await manager.stop_all()


class TestConnectionStability:
    """Integration Tests für Verbindungsstabilität über längere Zeit."""

    @pytest.mark.asyncio
    @pytest.mark.timeout(80)  # 80 Sekunden Timeout für den Test
    async def test_connection_stable_for_70_seconds(self):
        """
        Test: Verbindung bleibt mindestens 70 Sekunden stabil.

        Dieser Test simuliert eine Client-Server-Verbindung und
        verifiziert, dass bei korrektem Ping/Pong-Austausch die
        Verbindung nicht unterbrochen wird.
        """
        config = KeepAliveConfig(
            ping_interval=5.0,      # Alle 5 Sekunden
            pong_timeout=2.0,       # 2 Sekunden Timeout
            max_missed_pongs=3,
            initial_delay=1.0,
            jitter=0.1,
            adaptive=False,
        )

        connection_lost = False
        disconnect_reason = ""

        async def on_disconnect(conn_id):
            nonlocal connection_lost, disconnect_reason
            connection_lost = True
            disconnect_reason = f"Connection {conn_id} disconnected"

        ping_count = 0
        pong_count = 0

        async def mock_ping():
            nonlocal ping_count
            ping_count += 1

        manager = KeepAliveManager(default_config=config)
        manager.on_disconnect(on_disconnect)

        session = await manager.register(
            "long-test-session",
            mock_ping,
            auto_start=True,
        )

        start_time = asyncio.get_event_loop().time()
        test_duration = 70.0  # 70 Sekunden

        while asyncio.get_event_loop().time() - start_time < test_duration:
            await asyncio.sleep(0.5)

            # Simuliere Pong-Empfang wenn auf Pong gewartet wird
            if session._waiting_for_pong:
                session.handle_pong()
                pong_count += 1

            # Prüfe ob Verbindung noch besteht
            if connection_lost:
                pytest.fail(
                    f"Verbindung wurde nach {asyncio.get_event_loop().time() - start_time:.1f}s "
                    f"unterbrochen. Pings: {ping_count}, Pongs: {pong_count}. "
                    f"Grund: {disconnect_reason}"
                )

        await manager.stop_all()

        # Finale Assertions
        assert not connection_lost, "Verbindung wurde während des 70s-Tests unterbrochen"
        assert session.state != ConnectionState.DISCONNECTED
        assert ping_count >= 12, f"Zu wenige Pings gesendet: {ping_count}"
        assert pong_count >= 12, f"Zu wenige Pongs empfangen: {pong_count}"
        assert session.stats.missed_pongs == 0, f"Pongs wurden verpasst: {session.stats.missed_pongs}"


class TestPingPongTimings:
    """Tests für Ping/Pong-Timing-Verhalten."""

    @pytest.mark.asyncio
    async def test_pong_timeout_triggers_correctly(self):
        """Test: Pong-Timeout wird korrekt erkannt."""
        config = KeepAliveConfig(
            ping_interval=0.5,
            pong_timeout=0.2,
            max_missed_pongs=2,
            initial_delay=0.1,
            jitter=0.0,
            adaptive=False,
        )

        timeout_count = 0

        async def on_timeout(conn_id):
            nonlocal timeout_count
            timeout_count += 1

        async def mock_ping():
            pass

        session = KeepAliveSession(
            connection_id="timeout-test",
            ping_func=mock_ping,
            config=config,
            on_timeout=on_timeout,
        )

        await session.start()

        # Warte auf 2 Timeouts (0.1 initial + 2*(0.5 ping + 0.2 timeout) = 1.5s)
        await asyncio.sleep(2.0)

        await session.stop()

        # Sollte mindestens 2 Timeouts gegeben haben
        assert timeout_count >= 2, f"Erwartete mindestens 2 Timeouts, bekam {timeout_count}"

    @pytest.mark.asyncio
    async def test_late_pong_prevents_timeout(self):
        """Test: Später (aber rechtzeitiger) Pong verhindert Timeout."""
        config = KeepAliveConfig(
            ping_interval=0.3,
            pong_timeout=0.5,  # Längerer Timeout
            max_missed_pongs=3,
            initial_delay=0.1,
            jitter=0.0,
            adaptive=False,
        )

        timeout_count = 0

        async def on_timeout(conn_id):
            nonlocal timeout_count
            timeout_count += 1

        async def mock_ping():
            pass

        session = KeepAliveSession(
            connection_id="late-pong-test",
            ping_func=mock_ping,
            config=config,
            on_timeout=on_timeout,
        )

        await session.start()

        # Reagiere schnell auf Pongs
        for _ in range(10):
            await asyncio.sleep(0.1)
            if session._waiting_for_pong:
                # Sende Pong sofort (weit vor dem 0.5s Timeout)
                session.handle_pong()

        await session.stop()

        # Sollte keine Timeouts geben
        assert timeout_count == 0, f"Unerwartet {timeout_count} Timeouts"