import asyncio
from contextlib import suppress
from unittest import mock

import pytest

from aiohttp.base_protocol import BaseProtocol


async def test_loop() -> None:
    loop = asyncio.get_event_loop()
    asyncio.set_event_loop(None)
    pr = BaseProtocol(loop)
    assert pr._loop is loop


async def test_pause_writing() -> None:
    loop = asyncio.get_event_loop()
    pr = BaseProtocol(loop)
    assert not pr._paused
    pr.pause_writing()
    assert pr._paused


async def test_resume_writing_no_waiters() -> None:
    loop = asyncio.get_event_loop()
    pr = BaseProtocol(loop=loop)
    pr.pause_writing()
    assert pr._paused
    pr.resume_writing()
    assert not pr._paused


async def test_connection_made() -> None:
    loop = asyncio.get_event_loop()
    pr = BaseProtocol(loop=loop)
    tr = mock.Mock()
    assert pr.transport is None
    pr.connection_made(tr)
    assert pr.transport is not None


async def test_connection_lost_not_paused() -> None:
    loop = asyncio.get_event_loop()
    pr = BaseProtocol(loop=loop)
    tr = mock.Mock()
    pr.connection_made(tr)
    assert not pr._connection_lost
    pr.connection_lost(None)
    assert pr.transport is None
    assert pr._connection_lost


async def test_connection_lost_paused_without_waiter() -> None:
    loop = asyncio.get_event_loop()
    pr = BaseProtocol(loop=loop)
    tr = mock.Mock()
    pr.connection_made(tr)
    assert not pr._connection_lost
    pr.pause_writing()
    pr.connection_lost(None)
    assert pr.transport is None
    assert pr._connection_lost


async def test_drain_lost() -> None:
    loop = asyncio.get_event_loop()
    pr = BaseProtocol(loop=loop)
    tr = mock.Mock()
    pr.connection_made(tr)
    pr.connection_lost(None)
    with pytest.raises(ConnectionResetError):
        await pr._drain_helper()


async def test_drain_not_paused() -> None:
    loop = asyncio.get_event_loop()
    pr = BaseProtocol(loop=loop)
    tr = mock.Mock()
    pr.connection_made(tr)
    assert pr._drain_waiter is None
    await pr._drain_helper()
    assert pr._drain_waiter is None


async def test_resume_drain_waited() -> None:
    loop = asyncio.get_event_loop()
    pr = BaseProtocol(loop=loop)
    tr = mock.Mock()
    pr.connection_made(tr)
    pr.pause_writing()

    t = loop.create_task(pr._drain_helper())
    await asyncio.sleep(0)

    assert pr._drain_waiter is not None
    pr.resume_writing()
    assert (await t) is None
    assert pr._drain_waiter is None


async def test_lost_drain_waited_ok() -> None:
    loop = asyncio.get_event_loop()
    pr = BaseProtocol(loop=loop)
    tr = mock.Mock()
    pr.connection_made(tr)
    pr.pause_writing()

    t = loop.create_task(pr._drain_helper())
    await asyncio.sleep(0)

    assert pr._drain_waiter is not None
    pr.connection_lost(None)
    assert (await t) is None
    assert pr._drain_waiter is None


async def test_lost_drain_waited_exception() -> None:
    loop = asyncio.get_event_loop()
    pr = BaseProtocol(loop=loop)
    tr = mock.Mock()
    pr.connection_made(tr)
    pr.pause_writing()

    t = loop.create_task(pr._drain_helper())
    await asyncio.sleep(0)

    assert pr._drain_waiter is not None
    exc = RuntimeError()
    pr.connection_lost(exc)
    with pytest.raises(RuntimeError) as cm:
        await t
    assert cm.value is exc
    assert pr._drain_waiter is None


async def test_lost_drain_cancelled() -> None:
    loop = asyncio.get_event_loop()
    pr = BaseProtocol(loop=loop)
    tr = mock.Mock()
    pr.connection_made(tr)
    pr.pause_writing()

    fut = loop.create_future()

    async def wait():
        fut.set_result(None)
        await pr._drain_helper()

    t = loop.create_task(wait())
    await fut
    t.cancel()

    assert pr._drain_waiter is not None
    pr.connection_lost(None)
    with suppress(asyncio.CancelledError):
        await t
    assert pr._drain_waiter is None


async def test_resume_drain_cancelled() -> None:
    loop = asyncio.get_event_loop()
    pr = BaseProtocol(loop=loop)
    tr = mock.Mock()
    pr.connection_made(tr)
    pr.pause_writing()

    fut = loop.create_future()

    async def wait():
        fut.set_result(None)
        await pr._drain_helper()

    t = loop.create_task(wait())
    await fut
    t.cancel()

    assert pr._drain_waiter is not None
    pr.resume_writing()
    with suppress(asyncio.CancelledError):
        await t
    assert pr._drain_waiter is None

[(11, 'asyncio.get_event_loop', 'asyncio.get_event_loop', False, 'import asyncio\n'), (12, 'asyncio.set_event_loop', 'asyncio.set_event_loop', False, 'import asyncio\n'), (13, 'aiohttp.base_protocol.BaseProtocol', 'BaseProtocol', False, 'from aiohttp.base_protocol import BaseProtocol\n'), (18, 'asyncio.get_event_loop', 'asyncio.get_event_loop', False, 'import asyncio\n'), (19, 'aiohttp.base_protocol.BaseProtocol', 'BaseProtocol', False, 'from aiohttp.base_protocol import BaseProtocol\n'), (26, 'asyncio.get_event_loop', 'asyncio.get_event_loop', False, 'import asyncio\n'), (27, 'aiohttp.base_protocol.BaseProtocol', 'BaseProtocol', False, 'from aiohttp.base_protocol import BaseProtocol\n'), (35, 'asyncio.get_event_loop', 'asyncio.get_event_loop', False, 'import asyncio\n'), (36, 'aiohttp.base_protocol.BaseProtocol', 'BaseProtocol', False, 'from aiohttp.base_protocol import BaseProtocol\n'), (37, 'unittest.mock.Mock', 'mock.Mock', False, 'from unittest import mock\n'), (44, 'asyncio.get_event_loop', 'asyncio.get_event_loop', False, 'import asyncio\n'), (45, 'aiohttp.base_protocol.BaseProtocol', 'BaseProtocol', False, 'from aiohttp.base_protocol import BaseProtocol\n'), (46, 'unittest.mock.Mock', 'mock.Mock', False, 'from unittest import mock\n'), (55, 'asyncio.get_event_loop', 'asyncio.get_event_loop', False, 'import asyncio\n'), (56, 'aiohttp.base_protocol.BaseProtocol', 'BaseProtocol', False, 'from aiohttp.base_protocol import BaseProtocol\n'), (57, 'unittest.mock.Mock', 'mock.Mock', False, 'from unittest import mock\n'), (67, 'asyncio.get_event_loop', 'asyncio.get_event_loop', False, 'import asyncio\n'), (68, 'aiohttp.base_protocol.BaseProtocol', 'BaseProtocol', False, 'from aiohttp.base_protocol import BaseProtocol\n'), (69, 'unittest.mock.Mock', 'mock.Mock', False, 'from unittest import mock\n'), (77, 'asyncio.get_event_loop', 'asyncio.get_event_loop', False, 'import asyncio\n'), (78, 'aiohttp.base_protocol.BaseProtocol', 'BaseProtocol', False, 'from aiohttp.base_protocol import BaseProtocol\n'), (79, 'unittest.mock.Mock', 'mock.Mock', False, 'from unittest import mock\n'), (87, 'asyncio.get_event_loop', 'asyncio.get_event_loop', False, 'import asyncio\n'), (88, 'aiohttp.base_protocol.BaseProtocol', 'BaseProtocol', False, 'from aiohttp.base_protocol import BaseProtocol\n'), (89, 'unittest.mock.Mock', 'mock.Mock', False, 'from unittest import mock\n'), (103, 'asyncio.get_event_loop', 'asyncio.get_event_loop', False, 'import asyncio\n'), (104, 'aiohttp.base_protocol.BaseProtocol', 'BaseProtocol', False, 'from aiohttp.base_protocol import BaseProtocol\n'), (105, 'unittest.mock.Mock', 'mock.Mock', False, 'from unittest import mock\n'), (119, 'asyncio.get_event_loop', 'asyncio.get_event_loop', False, 'import asyncio\n'), (120, 'aiohttp.base_protocol.BaseProtocol', 'BaseProtocol', False, 'from aiohttp.base_protocol import BaseProtocol\n'), (121, 'unittest.mock.Mock', 'mock.Mock', False, 'from unittest import mock\n'), (138, 'asyncio.get_event_loop', 'asyncio.get_event_loop', False, 'import asyncio\n'), (139, 'aiohttp.base_protocol.BaseProtocol', 'BaseProtocol', False, 'from aiohttp.base_protocol import BaseProtocol\n'), (140, 'unittest.mock.Mock', 'mock.Mock', False, 'from unittest import mock\n'), (162, 'asyncio.get_event_loop', 'asyncio.get_event_loop', False, 'import asyncio\n'), (163, 'aiohttp.base_protocol.BaseProtocol', 'BaseProtocol', False, 'from aiohttp.base_protocol import BaseProtocol\n'), (164, 'unittest.mock.Mock', 'mock.Mock', False, 'from unittest import mock\n'), (72, 'pytest.raises', 'pytest.raises', False, 'import pytest\n'), (94, 'asyncio.sleep', 'asyncio.sleep', False, 'import asyncio\n'), (110, 'asyncio.sleep', 'asyncio.sleep', False, 'import asyncio\n'), (126, 'asyncio.sleep', 'asyncio.sleep', False, 'import asyncio\n'), (131, 'pytest.raises', 'pytest.raises', False, 'import pytest\n'), (156, 'contextlib.suppress', 'suppress', False, 'from contextlib import suppress\n'), (180, 'contextlib.suppress', 'suppress', False, 'from contextlib import suppress\n')]