"""Tests for proactor_events.py"""

import socket
import unittest
from unittest import mock

import asyncio
from asyncio.proactor_events import BaseProactorEventLoop
from asyncio.proactor_events import _ProactorSocketTransport
from asyncio.proactor_events import _ProactorWritePipeTransport
from asyncio.proactor_events import _ProactorDuplexPipeTransport
from asyncio import test_utils


class ProactorSocketTransportTests(unittest.TestCase):

    def setUp(self):
        self.loop = test_utils.TestLoop()
        self.proactor = mock.Mock()
        self.loop._proactor = self.proactor
        self.protocol = test_utils.make_test_protocol(asyncio.Protocol)
        self.sock = mock.Mock(socket.socket)

    def test_ctor(self):
        fut = asyncio.Future(loop=self.loop)
        tr = _ProactorSocketTransport(
            self.loop, self.sock, self.protocol, fut)
        test_utils.run_briefly(self.loop)
        self.assertIsNone(fut.result())
        self.protocol.connection_made(tr)
        self.proactor.recv.assert_called_with(self.sock, 4096)

    def test_loop_reading(self):
        tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
        tr._loop_reading()
        self.loop._proactor.recv.assert_called_with(self.sock, 4096)
        self.assertFalse(self.protocol.data_received.called)
        self.assertFalse(self.protocol.eof_received.called)

    def test_loop_reading_data(self):
        res = asyncio.Future(loop=self.loop)
        res.set_result(b'data')

        tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)

        tr._read_fut = res
        tr._loop_reading(res)
        self.loop._proactor.recv.assert_called_with(self.sock, 4096)
        self.protocol.data_received.assert_called_with(b'data')

    def test_loop_reading_no_data(self):
        res = asyncio.Future(loop=self.loop)
        res.set_result(b'')

        tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)

        self.ass
ertRaises(AssertionError, tr._loop_reading, res)

        tr.close = mock.Mock()
        tr._read_
fut = res
        tr._loop_reading(res)
        self.assertFalse(self.loop._proactor.recv.called)
        self.assertTrue(self.protocol.eof_received.called)
        self.assertTrue(tr.close.called)

    def test_loop_reading_aborted(self):
        err = self.loop._proactor.recv.side_effect = ConnectionAbortedError()

        tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
        tr._fatal_error = mock.Mock()
        tr._loop_reading()
        tr._fatal_error.assert_called_with(
                            err,
                            'Fatal read error on pipe transport')

    def test_loop_reading_aborted_closing(self):
        self.loop._proactor.recv.side_effect = ConnectionAbortedError()

        tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
        tr._closing = True
        tr._fatal_error = mock.Mock()
        tr._loop_reading()
        self.assertFalse(tr._fatal_error.called)

    def test_loop_reading_aborted_is_fatal(self):
        self.loop._proactor.recv.side_effect = ConnectionAbortedError()
        tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
        tr._closing = False
        tr._fatal_error = mock.Mock()
        tr._loop_reading()
        self.assertTrue(tr._fatal_error.called)

    def test_loop_reading_conn_reset_lost(self):
        err = self.loop._proactor.recv.side_effect = ConnectionResetError()

        tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
        tr._closing = False
        tr._fatal_error = mock.Mock()
        tr._force_close = mock.Mock()
        tr._loop_reading()
        self.assertFalse(tr._fatal_error.called)
        tr._force_close.assert_called_with(err)

    def test_loop_reading_exception(self):
        err = self.loop._proactor.recv.side_effect = (OSError())

        tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
        tr._fatal_error = mock.Mock()
        tr._loop_reading()
        tr._fatal_error.assert_called_with(
                            err,
                            'Fatal read error on pipe transport')

    def test_write(self):
        tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
        tr._loop_writing = mock.Mock()
        tr.write(b'data')
        self.assertEqual(tr._buffer, None)
        tr._loop_writing.assert_called_with(data=b'data')

    def test_write_no_data(self):
        tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
        tr.write(b'')
        self.assertFalse(tr._buffer)

    def test_write_more(self):
        tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
        tr._write_fut = mock.Mock()
        tr._loop_writing = mock.Mock()
        tr.write(b'data')
        self.assertEqual(tr._buffer, b'data')
        self.assertFalse(tr._loop_writing.called)

    def test_loop_writing(self):
        tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
        tr._buffer = bytearray(b'data')
        tr._loop_writing()
        self.loop._proactor.send.assert_called_with(self.sock, b'data')
        self.loop._proactor.send.return_value.add_done_callback.\
            assert_called_with(tr._loop_writing)

    @mock.patch('asyncio.proactor_events.logger')
    def test_loop_writing_err(self, m_log):
        err = self.loop._proactor.send.side_effect = OSError()
        tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
        tr._fatal_error = mock.Mock()
        tr._buffer = [b'da', b'ta']
        tr._loop_writing()
        tr._fatal_error.assert_called_with(
                            err,
                            'Fatal write error on pipe transport')
        tr._conn_lost = 1

        tr.write(b'data')
        tr.write(b'data')
        tr.write(b'data')
        tr.write(b'data')
        tr.write(b'data')
        self.assertEqual(tr._buffer, None)
        m_log.warning.assert_called_with('socket.send() raised exception.')

    def test_loop_writing_stop(self):
        fut = asyncio.Future(loop=self.loop)
        fut.set_result(b'data')

        tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
        tr._write_fut = fut
        tr._loop_writing(fut)
        self.assertIsNone(tr._write_fut)

    def test_loop_writing_closing(self):
        fut = asyncio.Future(loop=self.loop)
        fut.set_result(1)

        tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
        tr._write_fut = fut
        tr.close()
        tr._loop_writing(fut)
        self.assertIsNone(tr._write_fut)
        test_utils.run_briefly(self.loop)
        self.protocol.connection_lost.assert_called_with(None)

    def test_abort(self):
        tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
        tr._force_close = mock.Mock()
        tr.abort()
        tr._force_close.assert_called_with(None)

    def test_close(self):
        tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
        tr.close()
        test_utils.run_briefly(self.loop)
        self.protocol.connection_lost.assert_called_with(None)
        self.assertTrue(tr._closing)
        self.assertEqual(tr._conn_lost, 1)

        self.protocol.connection_lost.reset_mock()
        tr.close()
        test_utils.run_briefly(self.loop)
        self.assertFalse(self.protocol.connection_lost.called)

    def test_close_write_fut(self):
        tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
        tr._write_fut = mock.Mock()
        tr.close()
        test_utils.run_briefly(self.loop)
        self.assertFalse(self.protocol.connection_lost.called)

    def test_close_buffer(self):
        tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
        tr._buffer = [b'data']
        tr.close()
        test_utils.run_briefly(self.loop)
        self.assertFalse(self.protocol.connection_lost.called)

    @mock.patch('asyncio.base_events.logger')
    def test_fatal_error(self, m_logging):
        tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
        tr._force_close = mock.Mock()
        tr._fata