"""Base test classes for async_adapter_tests.py."""

from __future__ import annotations

import functools
import logging
import select
import sys
import unittest
import uuid
from datetime import datetime, timezone
from unittest import mock

import pika
import pika._utils
from pika import adapters
from pika.adapters import select_connection
from pika.exchange_type import ExchangeType
from tests.wrappers.threaded_test_wrapper import create_run_in_thread_decorator

TEST_TIMEOUT = 15

# Decorator for running our tests in threads with timeout
# NOTE: we give it a little more time to give our I/O loop-based timeout logic
# sufficient time to mop up.
run_test_in_thread_with_timeout = create_run_in_thread_decorator(TEST_TIMEOUT *
                                                                 1.1)


def make_stop_on_error_with_self(the_self=None):
    """
    Create a decorator that stops test if the decorated method exits with exception and causes the
    test to fail by re-raising that exception after ioloop exits.

    :param None | AsyncTestCase the_self: if None, will use the first arg of     decorated method if
    it is an instance of AsyncTestCase, raising     exception otherwise.
    """

    def stop_on_error_with_self_decorator(fun):

        @functools.wraps(fun)
        def stop_on_error_wrapper(*args, **kwargs):
            this = the_self
            if this is None and args and isinstance(args[0], AsyncTestCase):
                this = args[0]
            if not isinstance(this, AsyncTestCase):
                raise TypeError('Decorated method is not an AsyncTestCase '
                                f'instance method: {fun!r}')
            try:
                return fun(*args, **kwargs)
            except Exception as error:
                this.logger.exception('Stopping test due to failure in %r', fun)
                this.stop(error)

        return stop_on_error_wrapper

    return stop_on_error_with_self_decorator


# Decorator that stops test if AsyncTestCase-based method exits with
# exception and causes the test to fail by re-raising that exception after
# ioloop exits.
#
# NOTE: only use it to decorate instance methods where self arg is a
#    AsyncTestCase instance.
stop_on_error_in_async_test_case_method = make_stop_on_error_with_self()


def enable_tls():
    return '--use-tls' in sys.argv


class AsyncTestCase(unittest.TestCase):
    DESCRIPTION = ''
    ADAPTER = None
    TIMEOUT = TEST_TIMEOUT

    def setUp(self):
        self.logger = logging.getLogger(self.__class__.__name__)
        self.parameters = self.new_connection_params()
        self._timed_out = False
        self._conn_open_error = None
        self._public_stop_requested = False
        self._conn_closed_reason = None
        self._public_stop_error_in = None  # exception passed to our stop()
        super().setUp()

    def new_connection_params(
            self) -> pika.ConnectionParameters | pika.URLParameters:
        if enable_tls():
            return self._new_tls_connection_params()
        return self._new_plaintext_connection_params()

    def _new_tls_connection_params(self) -> pika.URLParameters:
        self.logger.info('testing using TLS/SSL connection to port 5671')
        url = 'amqps://localhost:5671/%2F?ssl_options=%7B%27ca_certs%27%3A%27tests%2Fcerts%2Fca_certificate.pem%27%2C%27keyfile%27%3A%27tests%2Fcerts%2Fclient_key.pem%27%2C%27certfile%27%3A%27tests%2Fcerts%2Fclient_certificate.pem%27%7D'
        return pika.URLParameters(url)

    @staticmethod
    def _new_plaintext_connection_params() -> pika.ConnectionParameters:
        return pika.ConnectionParameters(host='127.0.0.1', port=5672)

    def tearDown(self):
        self._stop()

    def shortDescription(self):
        method_desc = super().shortDescription()
        if self.DESCRIPTION:
            return f'{self.DESCRIPTION} ({method_desc})'
        return method_desc

    def begin(self, channel) -> None:
        """Extend to start the actual tests on the channel."""
        self.fail('AsyncTestCase.begin_test not extended')

    def start(self, adapter_class, ioloop_factory):
        self.logger.info('start at %s', datetime.now(timezone.utc))
        self.adapter = adapter_class or self.ADAPTER

        self.connection = self.adapter(self.parameters,
                                       self.on_open,
                                       self.on_open_error,
                                       self.on_closed,
                                       custom_ioloop=ioloop_factory())
        try:
            self.timeout = self.connection._adapter_call_later(
                self.TIMEOUT, self.on_timeout)
            self._run_ioloop()

            self.assertFalse(self._timed_out)
            self.assertIsNone(self._conn_open_error)
            # Catch unexpected loss of connection
            self.assertTrue(
                self._public_stop_requested,
                'Unexpected end of test; connection close reason: '
                f'{self._conn_closed_reason!r}')
            if self._public_stop_error_in is not None:
                raise self._public_stop_error_in
        finally:
            self.connection._nbio.close()
            self.connection = None

    def stop_ioloop_only(self):
        """Request stopping of the connection's ioloop to end the test without closing the
        connection.
        """
        self._safe_remove_test_timeout()
        self.connection._nbio.stop()

    def stop(self, error=None):
        """
        Close the connection and stop the ioloop.

        :param None | Exception error: if not None, will raise the given     exception after ioloop
        exits.
        """
        if error is not None:
            if self._public_stop_error_in is None:
                self.logger.error('stop(): stopping with error=%r.', error)
            else:
                self.logger.error('stop(): replacing pending error=%r with %r',
                                  self._public_stop_error_in, error)
            self._public_stop_error_in = error

        self.logger.info('Stopping test')
        self._public_stop_requested = True
        assert self.connection is not None
        if self.connection.is_open:
            self.connection.close()  # NOTE: on_closed() will stop the ioloop
        elif self.connection.is_closed:
            self.logger.info(
                'Connection already closed, so just stopping ioloop')
            self._stop()

    def _run_ioloop(self):
        """Some tests need to subclass this in order to bootstrap their test
        logic after we instantiate the connection and assign it to
        `self.connection`, but before we run the ioloop
        """
        self.connection._nbio.run()

    def _safe_remove_test_timeout(self):
        if hasattr(self, 'timeout') and self.timeout is not None:
            self.logger.info('Removing timeout')
            assert self.connection is not None
            self.connection._adapter_remove_timeout(self.timeout)
            self.timeout = None

    def _stop(self):
        if hasattr(self, 'connection') and self.connection is not None:
            self._safe_remove_test_timeout()
            self.logger.info('Stopping ioloop')
            self.connection._nbio.stop()

    def on_closed(self, connection, error):
        """Called when the connection has finished closing."""
        self.logger.info('on_closed: %r %r', connection, error)
        self._conn_closed_reason = error
        self._stop()

    def on_open(self, connection):
        self.logger.debug('on_open: %r', connection)
        self.channel = connection.channel(
            on_open_callback=self.on_channel_opened)

    def on_open_error(self, connection, error):
        self._conn_open_error = error
        self.logger.error('on_open_error: %r %r', connection, error)
        self._stop()

    def on_channel_opened(self, channel):
        self.begin(channel)

    def on_timeout(self):
        """Called when stuck waiting for connection to close."""
        self.logger.error('%s timed out; on_timeout called at %s', self,
                          datetime.now(timezone.utc))
        self.timeout = None  # the dispatcher should have removed it
        self._timed_out = True
        # initiate cleanup
        self.stop()


class BoundQueueTestCase(AsyncTestCase):

    def start(self, adapter_class, ioloop_factory):
        # Encoding
        self.exchange = 'e-' + self.__class__.__name__ + ':' + uuid.uuid1().hex
        self.queue = 'q-' + self.__class__.__name__ + ':' + uuid.uuid1().hex
        self.routing_key = self.__class__.__name__
        super().start(adapter_class, ioloop_factory)

    def begin(self, channel):
        self.channel.exchange_declare(self.exchange,
                                      exchange_type=ExchangeType.direct,
                                      passive=False,
                                      durable=False,
                                      auto_delete=True,
                                      callback=self.on_exchange_declared)

    def on_exchange_declared(self, frame):
        self.channel.queue_declare(self.queue,
                                   passive=False,
                                   durable=False,
                                   exclusive=True,
                                   auto_delete=True,
                                   arguments={'x-expires': self.TIMEOUT * 1000},
                                   callback=self.on_queue_declared)

    def on_queue_declared(self, frame):
        self.channel.queue_bind(self.queue,
                                self.exchange,
                                self.routing_key,
                                callback=self.on_ready)

    def on_ready(self, frame):
        raise NotImplementedError


#
# In order to write test cases that will tested using all the Async Adapters
# write a class that inherits both from one of TestCase classes above and
# from the AsyncAdapters class below. This allows you to avoid duplicating the
# test methods for each adapter in each test class.
#


class AsyncAdapters:

    def start(self, adapter_class, ioloop_factory):
        """

        :param adapter_class: pika connection adapter class to test.
        :param ioloop_factory: to be called without args to instantiate a
           non-shared ioloop to be passed as the `custom_ioloop` arg to the
           `adapter_class` constructor. This is needed because some of the
           adapters default to using a singleton ioloop, which results in
           tests errors after prior tests close the ioloop to release resources,
           in order to eliminate ResourceWarning warnings concerning unclosed
           sockets from our adapters.
        :return:
        """
        raise NotImplementedError

    @run_test_in_thread_with_timeout
    def test_with_select_default(self):
        """SelectConnection:DefaultPoller."""
        with mock.patch.multiple(select_connection, SELECT_TYPE=None):
            self.start(adapters.SelectConnection, select_connection.IOLoop)

    @run_test_in_thread_with_timeout
    def test_with_select_select(self):
        """SelectConnection:select."""
        with mock.patch.multiple(select_connection, SELECT_TYPE='select'):
            self.start(adapters.SelectConnection, select_connection.IOLoop)

    @unittest.skipIf(
        not hasattr(select, 'poll') or not hasattr(select.poll(), 'modify'),
        'poll not supported')
    @run_test_in_thread_with_timeout
    def test_with_select_poll(self):
        """SelectConnection:poll."""
        with mock.patch.multiple(select_connection, SELECT_TYPE='poll'):
            self.start(adapters.SelectConnection, select_connection.IOLoop)

    @unittest.skipIf(not hasattr(select, 'epoll'), 'epoll not supported')
    @run_test_in_thread_with_timeout
    def test_with_select_epoll(self):
        """SelectConnection:epoll."""
        with mock.patch.multiple(select_connection, SELECT_TYPE='epoll'):
            self.start(adapters.SelectConnection, select_connection.IOLoop)

    @unittest.skipIf(not hasattr(select, 'kqueue'), 'kqueue not supported')
    @run_test_in_thread_with_timeout
    def test_with_select_kqueue(self):
        """SelectConnection:kqueue."""
        with mock.patch.multiple(select_connection, SELECT_TYPE='kqueue'):
            self.start(adapters.SelectConnection, select_connection.IOLoop)

    @unittest.skipIf(pika._utils.ON_WINDOWS, 'Windows not supported')
    @run_test_in_thread_with_timeout
    def test_with_gevent(self):
        """GeventConnection."""
        import gevent

        from pika.adapters.gevent_connection import GeventConnection, _GeventSelectorIOLoop

        def ioloop_factory():
            return _GeventSelectorIOLoop(gevent.get_hub())

        self.start(GeventConnection, ioloop_factory)

    @run_test_in_thread_with_timeout
    def test_with_tornado(self):
        """TornadoConnection."""
        import tornado.ioloop

        from pika.adapters.tornado_connection import TornadoConnection
        ioloop_factory = tornado.ioloop.IOLoop
        self.start(TornadoConnection, ioloop_factory)

    @run_test_in_thread_with_timeout
    def test_with_asyncio(self):
        """AsyncioConnection."""
        from pika.adapters.asyncio_connection import AsyncioConnection
        from tests.base.asyncio_loop import new_pika_asyncio_loop
        ioloop_factory = new_pika_asyncio_loop
        self.start(AsyncioConnection, ioloop_factory)