"""События менеджера, окно восстановления и гонки с отложенной очисткой."""

import asyncio
from copy import deepcopy

import pytest


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


async def new_session(case, state="air"):
    connection = case.connection()
    uid = await case.manager.new(connection)
    case.manager[uid].state = state
    if state == "air":
        case.router.add(uid)
    return uid, connection


async def finish_wait(request, allow_cancel=False):
    request.release()
    try:
        await asyncio.wait_for(request.task, 1)
    except asyncio.CancelledError:
        if not allow_cancel:
            raise


def test_new_connections_have_distinct_live_uids(make_manager, timing):
    """Проверяет разные UID одновременно существующих соединений."""
    async def scenario():
        async with timing:
            case = make_manager()
            first, _ = await new_session(case)
            second, _ = await new_session(case)
            assert first != second
            assert len(case.manager.sessions) == 2
    run(scenario)


def test_connect_and_data_events_reach_router(make_manager, timing):
    """Проверяет регистрацию connect и передачу UID/команды data роутеру."""
    async def scenario():
        async with timing:
            case = make_manager()
            uid, _ = await new_session(case, "ready")
            await case.manager.process(uid, {"type": "connect", "uid": uid})
            assert uid in case.router.routes
            await case.manager.process(uid, {"type": "data", "data": "telemetry"})
            assert case.router.data == [(uid, "telemetry")]
    run(scenario)


def test_disconnect_event_stops_transport_and_retains_record(make_manager, timing):
    """Проверяет немедленное закрытие транспорта по событию disconnect и начало окна восстановления."""
    async def scenario():
        async with timing:
            case = make_manager()
            uid, connection = await new_session(case)
            task = timing.spawn(case.manager.process(uid, {"type": "disconnect"}))
            await asyncio.sleep(0)
            assert connection.close_count == 1, "Менеджер не обработал событие disconnect"
            assert case.manager[uid].state == "stop"
            assert uid in case.manager.sessions
            request = await timing.sleeping("manager")
            assert request.delay == 60
            timing.clock.advance(60)
            await finish_wait(request)
            await asyncio.wait_for(task, 1)
            # If cleanup is detached, wait for its observable effect, not an arbitrary pause.
            async with asyncio.timeout(1):
                while uid in case.manager.sessions:
                    await asyncio.sleep(0)
            assert uid not in case.router.routes
    run(scenario)


def test_close_preserves_session_until_recovery_window_expires(make_manager, timing):
    """Проверяет сохранение сессии на 60 секунд и удаление записи/маршрутов после окна."""
    async def scenario():
        async with timing:
            case = make_manager()
            uid, connection = await new_session(case)
            case.manager[uid].last_data = {"telemetry": {"battery": 87}}
            task = timing.spawn(case.manager.close(uid))
            request = await timing.sleeping("manager")
            assert request.delay == 60
            assert connection.close_count == 1
            assert case.manager[uid].state == "stop"
            assert case.manager[uid].last_data == {"telemetry": {"battery": 87}}
            assert uid in case.router.routes
            timing.clock.advance(60)
            await finish_wait(request)
            await task
            assert uid not in case.manager.sessions
            assert uid not in case.router.routes
    run(scenario)


@pytest.mark.parametrize("reason", ["active", "wrong_upf", "missing", "migrated"])
def test_ineligible_recovery_does_not_move_data(make_manager, timing, reason):
    """Проверяет отказ восстановления активной, отсутствующей, перенесённой сессии и неверного upf."""
    async def scenario():
        async with timing:
            case = make_manager()
            old_uid, _ = await new_session(case)
            old = case.manager[old_uid]
            old.state = "air" if reason == "active" else ("migrated" if reason == "migrated" else "stop")
            old.last_data = {"telemetry": {"battery": 87}}
            new_uid, _ = await new_session(case, "ready")
            await case.manager.process(new_uid, {
                "type": "reconnect",
                "uid": -1 if reason == "missing" else old_uid,
                "upf": old.upf + 1 if reason == "wrong_upf" else old.upf,
            })
            assert case.router.moves == []
            assert case.manager[new_uid].last_data == {}
            assert old.last_data == {"telemetry": {"battery": 87}}
            assert case.manager[new_uid].state == "ready"
    run(scenario)


def test_successful_recovery_preserves_data_and_is_consumed_once(make_manager, timing):
    """Проверяет перенос динамических данных ровно один раз и сохранение собственного handshake."""
    async def scenario():
        async with timing:
            case = make_manager()
            old_uid, _ = await new_session(case)
            old = case.manager[old_uid]
            old.state = "stop"
            old.last_data = {"telemetry": {"battery": 87}}
            old.dynamic_meta = {"active_in": {"control"}, "active_out": {"telemetry"}}
            expected_data = deepcopy(old.last_data)
            expected_meta = deepcopy(old.dynamic_meta)
            new_uid, _ = await new_session(case, "ready")
            event = {"type": "reconnect", "uid": old_uid, "upf": old.upf}
            await case.manager.process(new_uid, event)
            assert case.manager[new_uid].last_data == expected_data
            assert case.manager[new_uid].dynamic_meta == expected_meta
            assert case.manager[new_uid].state == "ready"
            assert case.router.moves == [(old_uid, new_uid)]

            repeated_uid, _ = await new_session(case, "ready")
            await case.manager.process(repeated_uid, event)
            assert case.manager[repeated_uid].last_data == {}
            assert case.router.moves == [(old_uid, new_uid)]
            assert case.manager[new_uid].last_data == expected_data
    run(scenario)


def test_expired_session_cannot_be_recovered(make_manager, timing):
    """Проверяет отказ восстановления после удаления старой сессии по 60-секундному таймеру."""
    async def scenario():
        async with timing:
            case = make_manager()
            old_uid, _ = await new_session(case)
            old_upf = case.manager[old_uid].upf
            task = timing.spawn(case.manager.close(old_uid))
            request = await timing.sleeping("manager")
            timing.clock.advance(60)
            await finish_wait(request)
            await task
            new_uid, _ = await new_session(case, "ready")
            await case.manager.process(new_uid, {"type": "reconnect", "uid": old_uid, "upf": old_upf})
            assert case.router.moves == []
            assert case.manager[new_uid].last_data == {}
    run(scenario)


def test_old_cleanup_cannot_delete_recovered_session(make_manager, timing):
    """Проверяет, что старый таймер после восстановления не удаляет новую запись и её данные."""
    async def scenario():
        async with timing:
            case = make_manager()
            old_uid, _ = await new_session(case)
            old = case.manager[old_uid]
            old.last_data = {"telemetry": {"battery": 87}}
            cleanup = timing.spawn(case.manager.close(old_uid))
            request = await timing.sleeping("manager")
            new_uid, _ = await new_session(case, "ready")
            restored = case.manager[new_uid]
            await case.manager.process(new_uid, {"type": "reconnect", "uid": old_uid, "upf": old.upf})
            assert restored.last_data == {"telemetry": {"battery": 87}}
            timing.clock.advance(60)
            await finish_wait(request, allow_cancel=True)
            if not cleanup.cancelled():
                await cleanup
            assert case.manager[new_uid] is restored
            assert restored.last_data == {"telemetry": {"battery": 87}}
            assert new_uid in case.router.routes
    run(scenario)


def test_close_after_record_was_deleted_is_safe(make_manager, timing):
    """Проверяет повторный close после истечения окна и удаления UID."""
    async def scenario():
        async with timing:
            case = make_manager()
            uid, connection = await new_session(case)
            task = timing.spawn(case.manager.close(uid))
            request = await timing.sleeping("manager")
            timing.clock.advance(60)
            await finish_wait(request)
            await task
            await asyncio.wait_for(case.manager.close(uid), 1)
            assert connection.close_count == 1
            assert case.router.removed == [uid]
    run(scenario)


def test_transport_close_error_does_not_prevent_expiry(make_manager, timing):
    """Проверяет очистку сессии/маршрутов при ConnectionResetError из writer.wait_closed."""
    from transport.tcp.connection import Connection

    async def scenario():
        async with timing:
            case = make_manager()
            failure = ConnectionResetError("close regression")

            class BrokenWriter:
                close_count = 0

                def close(self):
                    self.close_count += 1

                async def wait_closed(self):
                    raise failure

            writer = BrokenWriter()
            connection = Connection(None, writer)
            uid = await case.manager.new(connection)
            case.manager[uid].state = "air"
            case.router.add(uid)
            task = asyncio.create_task(case.manager.close(uid))
            try:
                request = await timing.sleeping("manager")
                assert request.delay == 60
                assert case.manager[uid].state == "stop"
                timing.clock.advance(60)
                try:
                    await finish_wait(request)
                except ConnectionResetError as error:
                    assert error is failure
                result = (await asyncio.gather(task, return_exceptions=True))[0]
                assert result is None or result is failure
                assert uid not in case.manager.sessions
                assert uid not in case.router.routes
                assert writer.close_count == 1
                assert case.router.removed == [uid]
            finally:
                if not task.done():
                    task.cancel()
                await asyncio.gather(task, return_exceptions=True)
    run(scenario)