import asyncio
import socket
import time
import logging
from unittest.mock import Mock
from torba.testcase import IntegrationTestCase, Conductor
import lbry.wallet
from lbry.schema.claim import Claim
from lbry.wallet.transaction import Transaction, Output
from lbry.wallet.dewies import dewies_to_lbc as d2l, lbc_to_dewies as l2d


log = logging.getLogger(__name__)
def wrap_callback_event(fn, callback):
    def inner(*a, **kw):
        callback()
        return fn(*a, **kw)
    return inner


class TestSessionBloat(IntegrationTestCase):
    """
    ERROR:asyncio:Fatal read error on socket transport
    protocol: <lbrynet.wallet.server.session.LBRYElectrumX object at 0x7f7e3bfcaf60>
    transport: <_SelectorSocketTransport fd=3236 read=polling write=<idle, bufsize=0>>
    Traceback (most recent call last):
      File "/usr/lib/python3.7/asyncio/selector_events.py", line 801, in _read_ready__data_received
        data = self._sock.recv(self.max_size)
    TimeoutError: [Errno 110] Connection timed out
    """

    LEDGER = lbry.wallet

    async def asyncSetUp(self):
        self.conductor = Conductor(
            ledger_module=self.LEDGER, manager_module=self.MANAGER, verbosity=self.VERBOSITY
        )
        await self.conductor.start_blockchain()
        self.addCleanup(self.conductor.stop_blockchain)

        await self.conductor.start_spv()

        self.session_manager = self.conductor.spv_node.server.session_mgr
        self.session_manager.servers['TCP'].sockets[0].setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 64)
        self.session_manager.servers['TCP'].sockets[0].setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 64)

        self.addCleanup(self.conductor.stop_spv)
        await self.conductor.start_wallet()
        self.addCleanup(self.conductor.stop_wallet)

        self.client_session = list(self.session_manager.sessions)[0]
        self.client_session.transport.set_write_buffer_limits(0, 0)

        self.paused_session = asyncio.Event(loop=self.loop)
        self.resumed_session = asyncio.Event(loop=self.loop)

        def paused():
            self.resumed_session.clear()
            self.paused_session.set()

        def delayed_resume():
            self.paused_session.clear()

            time.sleep(1)
            self.resumed_session.set()

        self.client_session.pause_writing = wrap_callback_event(self.client_session.pause_writing, paused)
        self.client_session.resume_writing = wrap_callback_event(self.client_session.resume_writing, delayed_resume)

        self.blockchain = self.conductor.blockchain_node
        self.wallet_node = self.conductor.wallet_node
        self.manager = self.wallet_node.manager
        self.ledger = self.wallet_node.ledger
        self.wallet = self.wallet_node.wallet
        self.account = self.wallet_node.wallet.default_account

    async def test_session_bloat_from_socket_timeout(self):
        await self.account.ensure_address_gap()

        address1, address2 = await self.account.receiving.get_addresses(limit=2, only_usable=True)
        sendtxid1 = await self.blockchain.send_to_address(address1, 5)
        sendtxid2 = await self.blockchain.send_to_address(address2, 5)

        await self.blockchain.generate(1)
        await asyncio.wait([
            self.on_transaction_id(sendtxid1),
            self.on_transaction_id(sendtxid2)
        ])

        self.assertEqual(d2l(await self.account.get_balance()), '10.0')

        channel = Claim()
        channel_txo = Output.pay_claim_name_pubkey_hash(
            l2d('1.0'), '@bar', channel, self.account.ledger.address_to_hash160(address1)
        )
        channel_txo.generate_channel_private_key()
        channel_txo.script.generate()
        channel_tx = await Transaction.create([], [channel_txo], [self.account], self.account)

        stream = Claim()
        stream.stream.description = "0" * 8000
        stream_txo = Output.pay_claim_name_pubkey_hash(
            l2d('1.0'), 'foo', stream, self.account.ledger.address_to_hash160(address1)
        )
        stream_tx = await Transaction.create([], [stream_txo], [self.account], self.account)
        stream_txo.sign(channel_txo)
        await stream_tx.sign([self.account])
        self.paused_session.clear()
        self.resumed_session.clear()

        await self.broadcast(channel_tx)
        await self.broadcast(stream_tx)
        await asyncio.wait_for(self.paused_session.wait(), 2)
        self.assertEqual(1, len(self.session_manager.sessions))

        real_sock = self.client_session.transport._extra.pop('socket')
        mock_sock = Mock(spec=socket.socket)

        for attr in dir(real_sock):
            if not attr.startswith('__'):
                setattr(mock_sock, attr, getattr(real_sock, attr))

        def recv(*a, **kw):
            raise TimeoutError("[Errno 110] Connection timed out")

        mock_sock.recv = recv
        self.client_session.transport._sock = mock_sock
        self.client_session.transport._extra['socket'] = mock_sock
        self.assertFalse(self.resumed_session.is_set())
        self.assertFalse(self.session_manager.session_event.is_set())
        await self.session_manager.session_event.wait()
        self.assertEqual(0, len(self.session_manager.sessions))