from __future__ import absolute_import
from __future__ import with_statement

import warnings

from mock import patch

from kombu import Connection
from kombu.exceptions import StdChannelError
from kombu.transport import virtual
from kombu.utils import uuid

from kombu.tests.compat import catch_warnings
from kombu.tests.utils import TestCase
from kombu.tests.utils import Mock, redirect_stdouts


def client(**kwargs):
    return Connection(transport='kombu.transport.virtual.Transport', **kwargs)


def memory_client():
    return Connection(transport='memory')


class test_BrokerState(TestCase):

    def test_constructor(self):
        s = virtual.BrokerState()
        self.assertTrue(hasattr(s, 'exchanges'))
        self.assertTrue(hasattr(s, 'bindings'))

        t = virtual.BrokerState(exchanges=16, bindings=32)
        self.assertEqual(t.exchanges, 16)
        self.assertEqual(t.bindings, 32)


class test_QoS(TestCase):

    def setUp(self):
        self.q = virtual.QoS(client().channel(), prefetch_count=10)

    def tearDown(self):
        self.q._on_collect.cancel()

    def test_constructor(self):
        self.assertTrue(self.q.channel)
        self.assertTrue(self.q.prefetch_count)
        self.assertFalse(self.q._delivered.restored)
        self.assertTrue(self.q._on_collect)

    @redirect_stdouts
    def test_can_consume(self, stdout, stderr):
        _restored = []

        class RestoreChannel(virtual.Channel):
            do_restore = True

            def _restore(self, message):
                _restored.append(message)

        self.assertTrue(self.q.can_consume())
        for i in range(self.q.prefetch_count - 1):
            self.q.append(i, uuid())
            self.assertTrue(self.q.can_consume())
        self.q.append(i + 1, uuid())
        self.assertFalse(self.q.can_consume())

        tag1 = iter(self.q._delivered).next()
        self.q.ack(tag1)
        self.assertTrue(self.q.can_consume())

        tag2 = uuid()
        self.q.append(i + 2, tag2)
        self.assertFalse(self.q.can_consume())
        self.q.reject(tag2)
        self.assertTrue(self.q.can_consume())

        self.q.channel = RestoreChannel(self.q.channel.connection)
        tag3 = uuid()
        self.q.append(i + 3, tag3)
        self.q.reject(tag3, requeue=True)
        self.q._flush()
        self.q.restore_unacked_once()
        self.assertListEqual(_restored, [11, 9, 8, 7, 6, 5, 4, 3, 2, 1])
        self.assertTrue(self.q._delivered.restored)
        self.assertFalse(self.q._delivered)

        self.q.restore_unacked_once()
        self.q._delivered.restored = False
        self.q.restore_unacked_once()

        self.assertTrue(stderr.getvalue())
        self.assertFalse(stdout.getvalue())

        self.q.restore_at_shutdown = False
        self.q.restore_unacked_once()

    def test_get(self):
        self.q._delivered['foo'] = 1
        self.assertEqual(self.q.get('foo'), 1)


class test_Message(TestCase):

    def test_create(self):
        c = client().channel()
        data = c.prepare_message('the quick brown fox...')
        tag = data['properties']['delivery_tag'] = uuid()
        message = c.message_to_python(data)
        self.assertIsInstance(message, virtual.Message)
        self.assertIs(message, c.message_to_python(message))

        self.assertEqual(message.body,
                         'the quick brown fox...'.encode('utf-8'))
        self.assertTrue(message.delivery_tag, tag)

    def test_create_no_body(self):
        virtual.Message(Mock(), {
            'body': None,
            'properties': {'delivery_tag': 1}})

    def test_serializable(self):
        c = client().channel()
        data = c.prepare_message('the quick brown fox...')
        tag = data['properties']['delivery_tag'] = uuid()
        message = c.message_to_python(data)
        dict_ = message.serializable()
        self.assertEqual(dict_['body'],
                         'the quick brown fox...'.encode('utf-8'))
        self.assertEqual(dict_['properties']['delivery_tag'], tag)


class test_AbstractChannel(TestCase):

    def test_get(self):
        with self.assertRaises(NotImplementedError):
            virtual.AbstractChannel()._get('queue')

    def test_put(self):
        with self.assertRaises(NotImplementedError):
            virtual.AbstractChannel()._put('queue', 'm')

    def test_size(self):
        self.assertEqual(virtual.AbstractChannel()._size('queue'), 0)

    def test_purge(self):
        with self.assertRaises(NotImplementedError):
            virtual.AbstractChannel()._purge('queue')

    def test_delete(self):
        with self.assertRaises(NotImplementedError):
            virtual.AbstractChannel()._delete('queue')

    def test_new_queue(self):
        self.assertIsNone(virtual.AbstractChannel()._new_queue('queue'))

    def test_has_queue(self):
        self.assertTrue(virtual.AbstractChannel()._has_queue('queue'))

    def test_poll(self):

        class Cycle(object):
            called = False

            def get(self):
                self.called = True
                return True

        cycle = Cycle()
        self.assertTrue(virtual.AbstractChannel()._poll(cycle))
        self.assertTrue(cycle.called)


class test_Channel(TestCase):

    def setUp(self):
        self.channel = client().channel()

    def tearDown(self):
        if self.channel._qos is not None:
            self.channel._qos._on_collect.cancel()

    def test_exchange_bind_interface(self):
        with self.assertRaises(NotImplementedError):
            self.channel.exchange_bind('dest', 'src', 'key')

    def test_exchange_unbind_interface(self):
        with self.assertRaises(NotImplementedError):
            self.channel.exchange_unbind('dest', 'src', 'key')

    def test_queue_unbind_interface(self):
        with self.assertRaises(NotImplementedError):
            self.channel.queue_unbind('dest', 'ex', 'key')

    def test_management(self):
        m = self.channel.connection.client.get_manager()
        self.assertTrue(m)
        m.get_bindings()
        m.close()

    def test_exchange_declare(self):
        c = self.channel

        with self.assertRaises(StdChannelError):
            c.exchange_declare('test_exchange_declare', 'direct',
                               durable=True, auto_delete=True, passive=True)
        c.exchange_declare('test_exchange_declare', 'direct',
                           durable=True, auto_delete=True)
        c.exchange_declare('test_exchange_declare', 'direct',
                           durable=True, auto_delete=True, passive=True)
        self.assertIn('test_exchange_declare', c.state.exchanges)
        # can declare again with same values
        c.exchange_declare('test_exchange_declare', 'direct',
                           durable=True, auto_delete=True)
        self.assertIn('test_exchange_declare', c.state.exchanges)

        # using different values raises NotEquivalentError
        with self.assertRaises(virtual.NotEquivalentError):
            c.exchange_declare('test_exchange_declare', 'direct',
                               durable=False, auto_delete=True)

    def test_exchange_delete(self, ex='test_exchange_delete'):

        class PurgeChannel(virtual.Channel):
            purged = []

            def _purge(self, queue):
                self.purged.append(queue)

        c = PurgeChannel(self.channel.connection)

        c.exchange_declare(ex, 'direct', durable=True, auto_delete=True)
        self.assertIn(ex, c.state.exchanges)
        self.assertNotIn(ex, c.state.bindings)  # no bindings yet
        c.exchange_delete(ex)
        self.assertNotIn(ex, c.state.exchanges)

        c.exchange_declare(ex, 'direct', durable=True, auto_delete=True)
        c.queue_declare(ex)
        c.queue_bind(ex, ex, ex)
        self.assertTrue(c.state.bindings[ex])
        c.exchange_delete(ex)
        self.assertNotIn(ex, c.state.bindings)
        self.assertIn(ex, c.purged)

    def test_queue_delete__if_empty(self, n='test_queue_delete__if_empty'):
        class PurgeChannel(virtual.Channel):
            purged = []
            size = 30

            def _purge(self, queue):
                self.purged.append(queue)

            def _size(self, queue):
                return self.size

        c = PurgeChannel(self.channel.connection)
        c.exchange_declare(n)
        c.queue_declare(n)
        c.queue_bind(n, n, n)
        c.queue_bind(n, n, n)   # tests code path that returns
                                # if queue already bound.

        c.queue_delete(n, if_empty=True)
        self.assertIn(n, c.state.bindings)

        c.size = 0
        c.queue_delete(n, if_empty=True)
        self.assertNotIn(n, c.state.bindings)
        self.assertIn(n, c.purged)

    def test_queue_purge(self, n='test_queue_purge'):

        class PurgeChannel(virtual.Channel):
            purged = []

            def _purge(self, queue):
                self.purged.append(queue)

        c = PurgeChannel(self.channel.connection)
        c.exchange_declare(n)
        c.queue_declare(n)
        c.queue_bind(n, n, n)
        c.queue_purge(n)
        self.assertIn(n, c.purged)

    def test_basic_publish__get__consume__restore(self,
            n='test_basic_publish'):
        c = memory_client().channel()

        c.exchange_declare(n)
        c.queue_declare(n)
        c.queue_bind(n, n, n)
        c.queue_declare(n + '2')
        c.queue_bind(n + '2', n, n)

        m = c.prepare_message('nthex quick brown fox...')
        c.basic_publish(m, n, n)

        r1 = c.message_to_python(c.basic_get(n))
        self.assertTrue(r1)
        self.assertEqual(r1.body,
                         'nthex quick brown fox...'.encode('utf-8'))
        self.assertIsNone(c.basic_get(n))

        consumer_tag = uuid()

        c.basic_consume(n + '2', False, consumer_tag=consumer_tag,
                                        callback=lambda *a: None)
        self.assertIn(n + '2', c._active_queues)
        r2, _ = c.drain_events()
        r2 = c.message_to_python(r2)
        self.assertEqual(r2.body,
                         'nthex quick brown fox...'.encode('utf-8'))
        self.assertEqual(r2.delivery_info['exchange'], n)
        self.assertEqual(r2.delivery_info['routing_key'], n)
        with self.assertRaises(virtual.Empty):
            c.drain_events()
        c.basic_cancel(consumer_tag)

        c._restore(r2)
        r3 = c.message_to_python(c.basic_get(n))
        self.assertTrue(r3)
        self.assertEqual(r3.body, 'nthex quick brown fox...'.encode('utf-8'))
        self.assertIsNone(c.basic_get(n))

    def test_basic_ack(self):

        class MockQoS(virtual.QoS):
            was_acked = False

            def ack(self, delivery_tag):
                self.was_acked = True

        self.channel._qos = MockQoS(self.channel)
        self.channel.basic_ack('foo')
        self.assertTrue(self.channel._qos.was_acked)

    def test_basic_recover__requeue(self):

        class MockQoS(virtual.QoS):
            was_restored = False

            def restore_unacked(self):
                self.was_restored = True

        self.channel._qos = MockQoS(self.channel)
        self.channel.basic_recover(requeue=True)
        self.assertTrue(self.channel._qos.was_restored)

    def test_restore_unacked_raises_BaseException(self):
        q = self.channel.qos
        q._flush = Mock()
        q._delivered = {1: 1}

        q.channel._restore = Mock()
        q.channel._restore.side_effect = SystemExit

        errors = q.restore_unacked()
        self.assertIsInstance(errors[0][0], SystemExit)
        self.assertEqual(errors[0][1], 1)
        self.assertFalse(q._delivered)

    @patch('kombu.transport.virtual.emergency_dump_state')
    @patch('kombu.transport.virtual.say')
    def test_restore_unacked_once_when_unrestored(self, say,
            emergency_dump_state):
        q = self.channel.qos
        q._flush = Mock()

        class State(dict):
            restored = False

        q._delivered = State({1: 1})
        ru = q.restore_unacked = Mock()
        exc = None
        try:
            raise KeyError()
        except KeyError, exc_:
            exc = exc_
        ru.return_value = [(exc, 1)]

        self.channel.do_restore = True
        q.restore_unacked_once()
        self.assertTrue(say.called)
        self.assertTrue(emergency_dump_state.called)

    def test_basic_recover(self):
        with self.assertRaises(NotImplementedError):
            self.channel.basic_recover(requeue=False)

    def test_basic_reject(self):

        class MockQoS(virtual.QoS):
            was_rejected = False

            def reject(self, delivery_tag, requeue=False):
                self.was_rejected = True

        self.channel._qos = MockQoS(self.channel)
        self.channel.basic_reject('foo')
        self.assertTrue(self.channel._qos.was_rejected)

    def test_basic_qos(self):
        self.channel.basic_qos(prefetch_count=128)
        self.assertEqual(self.channel._qos.prefetch_count, 128)

    def test_lookup__undeliverable(self, n='test_lookup__undeliverable'):
        warnings.resetwarnings()
        with catch_warnings(record=True) as log:
            self.assertListEqual(self.channel._lookup(n, n, 'ae.undeliver'),
                                                      ['ae.undeliver'])
            self.assertTrue(log)
            self.assertIn('could not be delivered', log[0].message.args[0])

    def test_context(self):
        x = self.channel.__enter__()
        self.assertIs(x, self.channel)
        x.__exit__()
        self.assertTrue(x.closed)

    def test_cycle_property(self):
        self.assertTrue(self.channel.cycle)

    def test_flow(self):
        with self.assertRaises(NotImplementedError):
            self.channel.flow(False)

    def test_close_when_no_connection(self):
        self.channel.connection = None
        self.channel.close()
        self.assertTrue(self.channel.closed)

    def test_drain_events_has_get_many(self):
        c = self.channel
        c._get_many = Mock()
        c._poll = Mock()
        c._consumers = [1]
        c._qos = Mock()
        c._qos.can_consume.return_value = True

        c.drain_events(timeout=10.0)
        c._get_many.assert_called_with(c._active_queues, timeout=10.0)

    def test_get_exchanges(self):
        self.channel.exchange_declare(exchange='foo')
        self.assertTrue(self.channel.get_exchanges())

    def test_basic_cancel_not_in_active_queues(self):
        c = self.channel
        c._consumers.add('x')
        c._tag_to_queue['x'] = 'foo'
        c._active_queues = Mock()
        c._active_queues.remove.side_effect = ValueError()

        c.basic_cancel('x')
        c._active_queues.remove.assert_called_with('foo')

    def test_basic_cancel_unknown_ctag(self):
        self.assertIsNone(self.channel.basic_cancel('unknown-tag'))

    def test_list_bindings(self):
        c = self.channel
        c.exchange_declare(exchange='foo')
        c.queue_declare(queue='q')
        c.queue_bind(queue='q', exchange='foo', routing_key='rk')

        self.assertIn(('q', 'foo', 'rk'), list(c.list_bindings()))

    def test_after_reply_message_received(self):
        c = self.channel
        c.queue_delete = Mock()
        c.after_reply_message_received('foo')
        c.queue_delete.assert_called_with('foo')

    def test_queue_delete_unknown_queue(self):
        self.assertIsNone(self.channel.queue_delete('xiwjqjwel'))

    def test_queue_declare_passive(self):
        has_queue = self.channel._has_queue = Mock()
        has_queue.return_value = False
        with self.assertRaises(StdChannelError):
            self.channel.queue_declare(queue='21wisdjwqe', passive=True)


class test_Transport(TestCase):

    def setUp(self):
        self.transport = client().transport

    def test_custom_polling_interval(self):
        x = client(transport_options=dict(polling_interval=32.3))
        self.assertEqual(x.transport.polling_interval, 32.3)

    def test_close_connection(self):
        c1 = self.transport.create_channel(self.transport)
        c2 = self.transport.create_channel(self.transport)
        self.assertEqual(len(self.transport.channels), 2)
        self.transport.close_connection(self.transport)
        self.assertFalse(self.transport.channels)
        del(c1)  # so pyflakes doesn't complain
        del(c2)

    def test_drain_channel(self):
        channel = self.transport.create_channel(self.transport)
        with self.assertRaises(virtual.Empty):
            self.transport._drain_channel(channel)
