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', (['None'], {}), False, 'import asyncio\n'), (13, 'aiohttp.base_protocol.BaseProtocol', 'BaseProtocol', (['loop'], {}), 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', (['loop'], {}), 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', ([], {'loop': 'loop'}), 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', ([], {'loop': 'loop'}), 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', ([], {'loop': 'loop'}), 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', ([], {'loop': 'loop'}), 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', ([], {'loop': 'loop'}), 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', ([], {'loop': 'loop'}), 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', ([], {'loop': 'loop'}), 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', ([], {'loop': 'loop'}), 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', ([], {'loop': 'loop'}), 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', ([], {'loop': 'loop'}), 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', ([], {'loop': 'loop'}), 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', (['ConnectionResetError'], {}), False, 'import pytest\n'), (94, 'asyncio.sleep', 'asyncio.sleep', (['(0)'], {}), False, 'import asyncio\n'), (110, 'asyncio.sleep', 'asyncio.sleep', (['(0)'], {}), False, 'import asyncio\n'), (126, 'asyncio.sleep', 'asyncio.sleep', (['(0)'], {}), False, 'import asyncio\n'), (131, 'pytest.raises', 'pytest.raises', (['RuntimeError'], {}), False, 'import pytest\n'), (156, 'contextlib.suppress', 'suppress', (['asyncio.CancelledError'], {}), False, 'from contextlib import suppress\n'), (180, 'contextlib.suppress', 'suppress', (['asyncio.CancelledError'], {}), False, 'from contextlib import suppress\n')]