import asyncio
from io import BytesIO

import pytest
from asyncssh.connection import SSHClientConnection
from asyncssh.stream import SSHReader

from scrapli.exceptions import ScrapliConnectionNotOpened, ScrapliTimeout


class DumbContainer:
    def __init__(self):
        self.preferred_auth = ()

    def __getattr__(self, item):
        # options has a billion attributes, just return None, doesnt matter for this test
        return None


def test_close(monkeypatch, asyncssh_transport):
    def _close(cls):
        pass

    monkeypatch.setattr(
        "asyncssh.connection.SSHClientConnection.close",
        _close,
    )

    # lie and pretend the session is already assigned
    options = DumbContainer()
    asyncssh_transport.session = SSHClientConnection(
        loop=asyncio.get_event_loop_policy().get_event_loop(), options=options
    )

    asyncssh_transport.close()

    assert asyncssh_transport.session is None
    assert asyncssh_transport.stdin is None
    assert asyncssh_transport.stdout is None


def test_close_catch_brokenpipe(monkeypatch, asyncssh_transport):
    def _close(cls):
        raise BrokenPipeError

    monkeypatch.setattr(
        "asyncssh.connection.SSHClientConnection.close",
        _close,
    )

    # lie and pretend the session is already assigned
    options = DumbContainer()
    asyncssh_transport.session = SSHClientConnection(
        loop=asyncio.get_event_loop_policy().get_event_loop(), options=options
    )

    asyncssh_transport.close()

    assert asyncssh_transport.session is None
    assert asyncssh_transport.stdin is None
    assert asyncssh_transport.stdout is None


def test_isalive_no_session(asyncssh_transport):
    assert asyncssh_transport.isalive() is False


def test_isalive(asyncssh_transport):
    # lie and pretend the session is already assigned
    options = DumbContainer()
    asyncssh_transport.session = SSHClientConnection(
        loop=asyncio.get_event_loop_policy().get_event_loop(), options=options
    )

    # lie and tell asyncssh auth is done
    asyncssh_transport.session._auth_complete = True

    # also have to lie and create a transport and have it return False when is_closing is called
    asyncssh_transport.session._transport = DumbContainer()
    asyncssh_transport.session._transport.is_closing = lambda: False

    assert asyncssh_transport.isalive() is True


def test_isalive_attribute_error(asyncssh_transport):
    # lie and pretend the session is already assigned
    options = DumbContainer()
    asyncssh_transport.session = SSHClientConnection(
        loop=asyncio.get_event_loop_policy().get_event_loop(), options=options
    )

    # lie and tell asyncssh auth is done
    asyncssh_transport.session._auth_complete = True

    assert asyncssh_transport.isalive() is False


async def test_read(monkeypatch, asyncssh_transport):
    async def _read(cls, _):
        return b"somebytes"

    monkeypatch.setattr(
        "asyncssh.stream.SSHReader.read",
        _read,
    )

    # lie and pretend the session is already assigned/stdout is already a thing
    asyncssh_transport.stdout = SSHReader("", "")

    assert await asyncssh_transport.read() == b"somebytes"


async def test_read_exception_not_open(asyncssh_transport):
    with pytest.raises(ScrapliConnectionNotOpened):
        await asyncssh_transport.read()


async def test_read_exception_timeout(monkeypatch, asyncssh_transport):
    async def _read(cls, _):
        await asyncio.sleep(0.5)

    monkeypatch.setattr(
        "asyncssh.stream.SSHReader.read",
        _read,
    )

    # lie and pretend the session is already assigned/stdout is already a thing
    asyncssh_transport.stdout = SSHReader("", "")
    asyncssh_transport._base_transport_args.timeout_transport = 0.1

    with pytest.raises(ScrapliTimeout):
        await asyncssh_transport.read()


def test_write(asyncssh_transport):
    asyncssh_transport.stdin = BytesIO()
    asyncssh_transport.write(b"blah")
    asyncssh_transport.stdin.seek(0)
    assert asyncssh_transport.stdin.read() == b"blah"


def test_write_exception(asyncssh_transport):
    with pytest.raises(ScrapliConnectionNotOpened):
        asyncssh_transport.write("blah")

[((108, 32, 108, 49), 'asyncssh.stream.SSHReader', 'SSHReader', ({(108, 42, 108, 44): '""""""', (108, 46, 108, 48): '""""""'}, {}), "('', '')", False, 'from asyncssh.stream import SSHReader\n'), ((128, 32, 128, 49), 'asyncssh.stream.SSHReader', 'SSHReader', ({(128, 42, 128, 44): '""""""', (128, 46, 128, 48): '""""""'}, {}), "('', '')", False, 'from asyncssh.stream import SSHReader\n'), ((136, 31, 136, 40), 'io.BytesIO', 'BytesIO', ({}, {}), '()', False, 'from io import BytesIO\n'), ((114, 9, 114, 50), 'pytest.raises', 'pytest.raises', ({(114, 23, 114, 49): 'ScrapliConnectionNotOpened'}, {}), '(ScrapliConnectionNotOpened)', False, 'import pytest\n'), ((131, 9, 131, 38), 'pytest.raises', 'pytest.raises', ({(131, 23, 131, 37): 'ScrapliTimeout'}, {}), '(ScrapliTimeout)', False, 'import pytest\n'), ((143, 9, 143, 50), 'pytest.raises', 'pytest.raises', ({(143, 23, 143, 49): 'ScrapliConnectionNotOpened'}, {}), '(ScrapliConnectionNotOpened)', False, 'import pytest\n'), ((120, 14, 120, 32), 'asyncio.sleep', 'asyncio.sleep', ({(120, 28, 120, 31): '(0.5)'}, {}), '(0.5)', False, 'import asyncio\n'), ((32, 13, 32, 44), 'asyncio.get_event_loop_policy', 'asyncio.get_event_loop_policy', ({}, {}), '()', False, 'import asyncio\n'), ((54, 13, 54, 44), 'asyncio.get_event_loop_policy', 'asyncio.get_event_loop_policy', ({}, {}), '()', False, 'import asyncio\n'), ((72, 13, 72, 44), 'asyncio.get_event_loop_policy', 'asyncio.get_event_loop_policy', ({}, {}), '()', False, 'import asyncio\n'), ((89, 13, 89, 44), 'asyncio.get_event_loop_policy', 'asyncio.get_event_loop_policy', ({}, {}), '()', False, 'import asyncio\n')]