"""Переходы протокола и жизненный цикл таймера отдельной сессии."""

import asyncio

import pytest


def run(scenario):
    asyncio.run(asyncio.wait_for(scenario(), 3))


async def tick(request, session):
    request.release()
    await asyncio.sleep(0)
    if session.watcher_task.done() and not session.watcher_task.cancelled():
        error = session.watcher_task.exception()
        if error is not None:
            raise error


def test_initial_deadline_and_watcher(make_session, timing):
    """Проверяет начальное окно в 60 секунд и запуск собственного воркера сессии."""
    async def scenario():
        async with timing:
            case = make_session()
            assert case.session.state == "new"
            assert case.session.timeout == timing.clock() + 60
            request = await timing.sleeping("session", case.session.watcher_task)
            assert request.delay > 0
            assert case.events == []
    run(scenario)


@pytest.mark.parametrize("state", ["new", "token", "ready", "air"])
def test_expired_deadline_emits_one_disconnect(make_session, timing, state):
    """Проверяет однократное timeout-событие во всех незавершённых и активном состояниях."""
    async def scenario():
        async with timing:
            case = make_session()
            case.session.state = state
            request = await timing.sleeping("session", case.session.watcher_task)
            timing.clock.advance(60)
            request.release()
            await asyncio.wait_for(case.session.watcher_task, 1)
            assert case.events == [(case.session.uid, {"type": "disconnect"})]
    run(scenario)


def test_watcher_does_not_disconnect_before_deadline(make_session, timing):
    """Проверяет, что воркер не отключает сессию раньше её дедлайна."""
    async def scenario():
        async with timing:
            case = make_session()
            request = await timing.sleeping("session", case.session.watcher_task)
            timing.clock.advance(59)
            await tick(request, case.session)
            assert case.events == []
            assert not case.session.watcher_task.done()
            await timing.sleeping("session", case.session.watcher_task)
    run(scenario)


def test_heartbeat_extends_current_deadline(make_session, timing):
    """Проверяет перенос дедлайна heartbeat и отсутствие отключения по старому сроку."""
    async def scenario():
        async with timing:
            case = make_session()
            case.session.state = "air"
            request = await timing.sleeping("session", case.session.watcher_task)
            timing.clock.advance(10)
            await case.session.feed({"command": "heartbeat_ask", "payload": {}})
            await asyncio.sleep(0)
            assert case.session.timeout == timing.clock() + 5
            assert case.connection.sent == [{"command": "heartbeat_answer"}]
            assert case.session.last_data == {}

            timing.clock.advance(4)
            await case.session.feed({"command": "heartbeat_ask", "payload": {}})
            new_deadline = case.session.timeout
            timing.clock.advance(1)
            await tick(request, case.session)
            assert case.events == []
            request = await timing.sleeping("session", case.session.watcher_task)
            timing.clock.advance(new_deadline - timing.clock())
            request.release()
            await asyncio.wait_for(case.session.watcher_task, 1)
            assert case.events == [(case.session.uid, {"type": "disconnect"})]
    run(scenario)


@pytest.mark.parametrize("state", ["stop", "migrated"])
def test_terminal_session_does_not_emit_timeout(make_session, timing, state):
    """Проверяет завершение воркера без timeout для отключённой или перенесённой сессии."""
    async def scenario():
        async with timing:
            case = make_session()
            request = await timing.sleeping("session", case.session.watcher_task)
            case.session.state = state
            timing.clock.advance(100)
            request.release()
            await asyncio.wait_for(case.session.watcher_task, 1)
            assert case.events == []
    run(scenario)


def test_stop_closes_transport_and_finishes_watcher(make_session, timing):
    """Проверяет закрытие соединения, состояние stop и завершение таймерного воркера."""
    async def scenario():
        async with timing:
            case = make_session()
            request = await timing.sleeping("session", case.session.watcher_task)
            await case.session.stop()
            assert case.session.state == "stop"
            assert case.connection.close_count == 1
            request.release()
            await asyncio.wait_for(case.session.watcher_task, 1)
            assert case.events == []
    run(scenario)


def test_connection_and_successful_handshake(make_session, timing):
    """Проверяет new→token→ready→air, ответы сервера и метаданные портов."""
    async def scenario():
        async with timing:
            case = make_session()
            session = case.session
            assert await session.feed({"command": "connection", "payload": {
                "name": "rover", "type": "robot", "in": ["control"], "out": ["telemetry"],
            }}) is None
            assert session.state == "token"
            await asyncio.sleep(0)
            assert case.connection.sent == [{"command": "token_ask", "payload": session.uid, "upf": session.upf}]
            assert session.static_meta["name"] == "rover"
            assert session.static_meta["type"] == "robot"
            assert session.static_meta["in"] == {"control"}
            assert session.static_meta["out"] == {"telemetry"}
            assert session.dynamic_meta == {"active_in": set(), "active_out": set()}

            assert await session.feed({"command": "token_answer", "payload": {"token": None, "upf": None}}) is None
            assert session.state == "ready"
            await asyncio.sleep(0)
            assert case.connection.sent[-1] == {"command": "handshake_ask"}
            assert await session.feed({"command": "handshake_answer", "payload": {"status": True, "state": None}}) == {
                "type": "connect", "uid": session.uid,
            }
            assert session.state == "air"
    run(scenario)


def test_recovery_request_still_requires_own_handshake(make_session, timing):
    """Проверяет нормализацию старого UID/upf и ожидание отдельного нового handshake."""
    async def scenario():
        async with timing:
            case = make_session()
            case.session.state = "token"
            event = await case.session.feed({"command": "token_answer", "payload": {"token": 123, "upf": 456}})
            assert event == {"type": "reconnect", "uid": 123, "upf": 456}
            assert case.session.state == "ready"
            event = await case.session.feed({"command": "handshake_answer", "payload": {"status": True}})
            assert event == {"type": "connect", "uid": case.session.uid}
            assert case.session.state == "air"
    run(scenario)


def test_data_updates_latest_value_for_each_command(make_session, timing):
    """Проверяет нормализованные data-события и замену последнего значения по команде."""
    async def scenario():
        async with timing:
            case = make_session()
            case.session.state = "air"
            for command, payload in [("telemetry", {"battery": 87}), ("status", {"ok": True}), ("telemetry", {"battery": 86})]:
                assert await case.session.feed({"command": command, "payload": payload}) == {"type": "data", "data": command}
            assert case.session.last_data == {"telemetry": {"battery": 86}, "status": {"ok": True}}
    run(scenario)


@pytest.mark.parametrize("state", ["new", "token", "ready"])
def test_wrong_sequence_disconnects_at_error_limit(make_session, timing, state):
    """Проверяет четыре допустимых логических ошибки и disconnect на пятой."""
    async def scenario():
        async with timing:
            case = make_session()
            case.session.state = state
            for _ in range(4):
                assert await case.session.feed({"command": "telemetry", "payload": {}}) is None
            assert await case.session.feed({"command": "telemetry", "payload": {}}) == {"type": "disconnect"}
    run(scenario)


def test_missing_command_requests_disconnect(make_session, timing):
    """Проверяет немедленное disconnect-событие у пакета без команды."""
    async def scenario():
        async with timing:
            case = make_session()
            assert await case.session.feed({"payload": {}}) == {"type": "disconnect"}
    run(scenario)


def test_failed_handshake_waits_before_retry(make_session, timing):
    """Проверяет реальное ожидание задержки перед повторным запросом handshake."""
    async def scenario():
        async with timing:
            case = make_session()
            case.session.state = "ready"
            task = timing.spawn(case.session.feed({"command": "handshake_answer", "payload": {"status": False, "state": "booting"}}))
            await asyncio.sleep(0)
            assert not task.done(), "Повторный handshake выполняется без ожидания задержки"
            request = await timing.sleeping("session", task)
            assert request.delay == 1
            assert case.connection.sent == []
            request.release()
            assert await task == {"type": "await handshake", "state": "booting"}
            await asyncio.sleep(0)
            assert case.connection.sent == [{"command": "handshake_ask"}]
            assert case.session.state == "ready"
    run(scenario)


def test_handshake_retry_limit_requests_disconnect(make_session, timing):
    """Проверяет отказ от дальнейших повторов после пяти неудачных retry handshake."""
    async def scenario():
        async with timing:
            case = make_session()
            case.session.state = "ready"
            case.session.timeout_attempt = 5
            assert await case.session.feed({"command": "handshake_answer", "payload": {"status": False}}) == {"type": "disconnect"}
    run(scenario)


@pytest.mark.parametrize("state", ["stop", "migrated"])
def test_stopped_session_does_not_send_packets(make_session, timing, state):
    """Проверяет запрет новых отправок в stop и migrated."""
    async def scenario():
        async with timing:
            case = make_session()
            await case.session.stop()
            case.session.state = state
            case.session.send({"command": "telemetry", "payload": {}})
            await asyncio.sleep(0)
            assert case.connection.sent == []
    run(scenario)


def test_background_send_does_not_block_processing_data(make_session, timing, monkeypatch):
    """Проверяет обработку новых данных, пока фоновая сетевая отправка ожидает завершения."""
    async def scenario():
        async with timing:
            case = make_session()
            case.session.state = "air"
            started, release, finished = asyncio.Event(), asyncio.Event(), asyncio.Event()
            original_send = case.connection.send

            async def pending_send(packet):
                started.set()
                try:
                    await release.wait()
                    await original_send(packet)
                finally:
                    finished.set()

            monkeypatch.setattr(case.connection, "send", pending_send)
            packet = {"command": "telemetry", "payload": {"battery": 87}}
            try:
                case.session.send(packet)
                await asyncio.wait_for(started.wait(), 1)
                assert case.connection.sent == []
                event = await asyncio.wait_for(case.session.feed({
                    "command": "status", "payload": {"ready": True},
                }), 1)
                assert event == {"type": "data", "data": "status"}
                assert case.session.last_data == {"status": {"ready": True}}
            finally:
                release.set()
            await asyncio.wait_for(finished.wait(), 1)
            assert case.connection.sent == [packet]
    run(scenario)