import os
import sys
import unittest
import pickle
import time

from itertools import count

sys.path.insert(0, os.pardir)
sys.path.append(os.getcwd())

from carrot.messaging import Consumer, Publisher, ConsumerSet
from carrot import serialization
from tests.utils import establish_test_connection


class AdvancedDataType(object):

    def __init__(self, something):
        self.data = something


def fetch_next_message(consumer):
    while True:
        message = consumer.fetch()
        if message:
            return message


class BackendMessagingCase(unittest.TestCase):
    nextq = count(1).next

    def setUp(self):
        self.conn = establish_test_connection()
        self.queue = TEST_QUEUE
        self.exchange = TEST_EXCHANGE
        self.routing_key = TEST_ROUTING_KEY

    def create_consumer(self, **options):
        queue = "%s%s" % (self.queue, self.nextq())
        return Consumer(connection=self.conn,
                        queue=queue, exchange=self.exchange,
                        routing_key=self.routing_key, **options)

    def create_consumerset(self, queues={}, consumers=[], **options):
        return ConsumerSet(connection=self.conn,
                           from_dict=queues, consumers=consumers, **options)

    def create_publisher(self, exchange=None, routing_key=None, **options):
        exchange = exchange or self.exchange
        routing_key = routing_key or self.routing_key
        return Publisher(connection=self.conn,
                        exchange=exchange, routing_key=routing_key,
                        **options)

    def test_regression_implied_auto_delete(self):
        consumer = self.create_consumer(exclusive=True, auto_declare=False)
        self.assertTrue(consumer.auto_delete, "exclusive implies auto_delete")
        consumer.close()

        consumer = self.create_consumer(durable=True, auto_delete=False,
                                        auto_declare=False)
        self.assertFalse(consumer.auto_delete,
            """durable does *not* imply auto_delete.
            regression: http://github.com/ask/carrot/issues/closed#issue/2""")
        consumer.close()

    def test_consumer_options(self):
        opposite_defaults = {
                "queue": "xyxyxyxy",
                "exchange": "xyxyxyxy",
                "routing_key": "xyxyxyxy",
                "durable": False,
                "exclusive": True,
                "auto_delete": True,
                "exchange_type": "topic",
        }
        consumer = Consumer(connection=self.conn, **opposite_defaults)
        for opt_name, opt_value in opposite_defaults.items():
            self.assertEquals(getattr(consumer, opt_name), opt_value)
        consumer.close()

    def test_consumer_backend(self):
        consumer = self.create_consumer()
        self.assertTrue(consumer.backend.connection is self.conn)
        consumer.close()

    def test_consumer_queue_declared(self):
        consumer = self.create_consumer()
        self.assertTrue(consumer.backend.queue_exists(consumer.queue))
        consumer.close()

    def test_consumer_callbacks(self):
        consumer = self.create_consumer()
        publisher = self.create_publisher()

        # raises on no callbacks
        self.assertRaises(NotImplementedError, consumer.receive, {}, {})

        callback1_scratchpad = {}

        def callback1(message_data, message):
            callback1_scratchpad["message_data"] = message_data

        callback2_scratchpad = {}

        def callback2(message_data, message):
            callback2_scratchpad.update({"delivery_tag": message.delivery_tag,
                                         "message_body": message.body})

        self.assertFalse(consumer.callbacks, "no default callbacks")
        consumer.register_callback(callback1)
        consumer.register_callback(callback2)
        self.assertEquals(len(consumer.callbacks), 2, "callbacks registered")

        self.assertTrue(consumer.callbacks[0] is callback1,
                "callbacks are ordered")
        self.assertTrue(consumer.callbacks[1] is callback2,
                "callbacks are ordered")

        body = {"foo": "bar"}

        message = self.create_raw_message(publisher, body, "Elaine was here")
        consumer._receive_callback(message)

        self.assertEquals(callback1_scratchpad.get("message_data"), body,
                "callback1 was called")
        self.assertEquals(callback2_scratchpad.get("delivery_tag"),
                "Elaine was here")

        consumer.close()
        publisher.close()

    def create_raw_message(self, publisher, body, delivery_tag):
        raw_message = publisher.create_message(body)
        raw_message.delivery_tag = delivery_tag
        return raw_message

    def test_empty_queue_returns_None(self):
        consumer = self.create_consumer()
        consumer.discard_all()
        self.assertFalse(consumer.fetch())
        consumer.close()

    def test_custom_serialization_scheme(self):
        serialization.registry.register('custom_test',
                pickle.dumps, pickle.loads,
                content_type='application/x-custom-test',
                content_encoding='binary')

        consumer = self.create_consumer()
        publisher = self.create_publisher()
        consumer.discard_all()

        data = {"string": "The quick brown fox jumps over the lazy dog",
                "int": 10,
                "float": 3.14159265,
                "unicode": u"The quick brown fox jumps over the lazy dog",
                "advanced": AdvancedDataType("something"),
                "set": set(["george", "jerry", "elaine", "cosmo"]),
                "exception": Exception("There was an error"),
        }

        publisher.send(data, serializer='custom_test')
        message = fetch_next_message(consumer)
        backend = self.conn.create_backend()
        self.assertTrue(isinstance(message, backend.Message))
        self.assertEquals(message.payload.get("int"), 10)
        self.assertEquals(message.content_type, 'application/x-custom-test')
        self.assertEquals(message.content_encoding, 'binary')

        decoded_data = message.decode()

        self.assertEquals(decoded_data.get("string"),
                "The quick brown fox jumps over the lazy dog")
        self.assertEquals(decoded_data.get("int"), 10)
        self.assertEquals(decoded_data.get("float"), 3.14159265)
        self.assertEquals(decoded_data.get("unicode"),
                u"The quick brown fox jumps over the lazy dog")
        self.assertEquals(decoded_data.get("set"),
            set(["george", "jerry", "elaine", "cosmo"]))
        self.assertTrue(isinstance(decoded_data.get("exception"), Exception))
        self.assertEquals(decoded_data.get("exception").args[0],
            "There was an error")
        self.assertTrue(isinstance(decoded_data.get("advanced"),
            AdvancedDataType))
        self.assertEquals(decoded_data["advanced"].data, "something")

        consumer.close()
        publisher.close()

    def test_consumer_fetch(self):
        consumer = self.create_consumer()
        publisher = self.create_publisher()
        consumer.discard_all()

        data = {"string": "The quick brown fox jumps over the lazy dog",
                "int": 10,
                "float": 3.14159265,
                "unicode": u"The quick brown fox jumps over the lazy dog",
        }

        publisher.send(data)
        message = fetch_next_message(consumer)
        backend = self.conn.create_backend()
        self.assertTrue(isinstance(message, backend.Message))

        self.assertEquals(message.decode(), data)

        consumer.close()
        publisher.close()

    def test_consumer_process_next(self):
        consumer = self.create_consumer()
        publisher = self.create_publisher()
        consumer.discard_all()

        scratchpad = {}

        def callback(message_data, message):
            scratchpad["delivery_tag"] = message.delivery_tag
        consumer.register_callback(callback)

        publisher.send({"name_discovered": {
                            "first_name": "Cosmo",
                            "last_name": "Kramer"}})

        while True:
            message = consumer.fetch(enable_callbacks=True)
            if message:
                break

        self.assertEquals(scratchpad.get("delivery_tag"),
                message.delivery_tag)

        consumer.close()
        publisher.close()

    def test_consumer_discard_all(self):
        consumer = self.create_consumer()
        publisher = self.create_publisher()
        consumer.discard_all()

        for i in xrange(100):
            publisher.send({"foo": "bar"})
        time.sleep(0.5)

        self.assertEquals(consumer.discard_all(), 100)

        consumer.close()
        publisher.close()

    def test_iterqueue(self):
        consumer = self.create_consumer()
        publisher = self.create_publisher()
        num = consumer.discard_all()

        it = consumer.iterqueue(limit=100)
        consumer.register_callback(lambda *args: args)

        for i in xrange(100):
            publisher.send({"foo%d" % i: "bar%d" % i})
        time.sleep(0.5)

        for i in xrange(100):
            try:
                message = it.next()
                data = message.decode()
                self.assertTrue("foo%d" % i in data, "foo%d not in data" % i)
                self.assertEquals(data.get("foo%d" % i), "bar%d" % i)
            except StopIteration:
                self.assertTrue(False, "iterqueue fails StopIteration")

        self.assertRaises(StopIteration, it.next)

        # no messages on queue raises StopIteration if infinite=False
        it = consumer.iterqueue()
        self.assertRaises(StopIteration, it.next)

        it = consumer.iterqueue(infinite=True)
        self.assertTrue(it.next() is None,
                "returns None if no messages and inifite=True")

        consumer.close()
        publisher.close()

    def test_publisher_message_priority(self):
        consumer = self.create_consumer()
        publisher = self.create_publisher()
        consumer.discard_all()

        m = publisher.create_message("foo", priority=9)

        publisher.send({"foo": "bar"}, routing_key="nowhere", priority=9,
                mandatory=False, immediate=False)

        consumer.discard_all()

        consumer.close()
        publisher.close()

    def test_backend_survives_channel_close_regr17(self):
        """
        test that a backend instance is still functional after
        a method that results in a channel closure.
        """
        backend = self.create_publisher().backend
        assert not backend.queue_exists('notaqueue')
        # after calling this once, the channel seems to close, but the
        # backend may be holding a reference to it...
        assert not backend.queue_exists('notaqueue')

    def disabled_publisher_mandatory_flag_regr16(self):
        """
        Test that the publisher "mandatory" flag
        raises exceptions at appropriate times.
        """
        routing_key = 'black_hole'

        assert self.conn.connection is not None

        message = {'foo': 'mandatory'}

        # sanity check cleanup from last test
        assert not self.create_consumer().backend.queue_exists(routing_key)

        publisher = self.create_publisher()

        # this should just get discarded silently, it's not mandatory
        publisher.send(message, routing_key=routing_key, mandatory=False)

        # This raises an unspecified exception because there is no queue to
        # deliver to
        self.assertRaises(Exception, publisher.send, message,
                          routing_key=routing_key, mandatory=True)

        # now bind a queue to it
        consumer = Consumer(connection=self.conn,
                            queue=routing_key, exchange=self.exchange,
                            routing_key=routing_key, durable=False,
                            exclusive=True)

        # check that it exists
        assert self.create_consumer().backend.queue_exists(routing_key)

        # this should now get routed to our consumer with no exception
        publisher.send(message, routing_key=routing_key, mandatory=True)

    def test_consumer_auto_ack(self):
        consumer = self.create_consumer(auto_ack=True)
        publisher = self.create_publisher()
        consumer.discard_all()

        publisher.send({"foo": "Baz"})
        message = fetch_next_message(consumer)
        self.assertEquals(message._state, "ACK")
        consumer.close()
        publisher.close()

        publisher = self.create_publisher()
        consumer = self.create_consumer(auto_ack=False)
        publisher.send({"foo": "Baz"})
        message = fetch_next_message(consumer)
        self.assertEquals(message._state, "RECEIVED")

        consumer.close()
        publisher.close()

    def test_consumer_consume(self):
        consumer = self.create_consumer(auto_ack=True)
        publisher = self.create_publisher()
        consumer.discard_all()

        data = {"foo": "Baz"}
        publisher.send(data)
        try:
            data2 = {"company": "Vandelay Industries"}
            publisher.send(data2)
            scratchpad = {}

            def callback(message_data, message):
                scratchpad["data"] = message_data
            consumer.register_callback(callback)

            it = consumer.iterconsume()
            it.next()
            self.assertEquals(scratchpad.get("data"), data)
            it.next()
            self.assertEquals(scratchpad.get("data"), data2)

            # Cancel consumer/close and restart.
            consumer.close()
            consumer = self.create_consumer(auto_ack=True)
            consumer.register_callback(callback)
            consumer.discard_all()
            scratchpad = {}

            # Test limits
            it = consumer.iterconsume(limit=4)
            publisher.send(data)
            publisher.send(data2)
            publisher.send(data)
            publisher.send(data2)
            publisher.send(data)

            it.next()
            self.assertEquals(scratchpad.get("data"), data)
            it.next()
            self.assertEquals(scratchpad.get("data"), data2)
            it.next()
            self.assertEquals(scratchpad.get("data"), data)
            it.next()
            self.assertEquals(scratchpad.get("data"), data2)
            self.assertRaises(StopIteration, it.next)


        finally:
            consumer.close()
            publisher.close()

    def test_consumerset_iterconsume(self):
        consumerset = self.create_consumerset(queues={
            "bar": {
                "exchange": "foo",
                "exchange_type": "direct",
                "routing_key": "foo.bar",
            },
            "baz": {
                "exchange": "foo",
                "exchange_type": "direct",
                "routing_key": "foo.baz",
            },
            "bam": {
                "exchange": "foo",
                "exchange_type": "direct",
                "routing_key": "foo.bam",
            },
            "xuzzy": {
                "exchange": "foo",
                "exchange_type": "direct",
                "routing_key": "foo.xuzzy",
            }})
        publisher = self.create_publisher(exchange="foo")
        consumerset.discard_all()

        scratchpad = {}

        def callback(message_data, message):
            scratchpad["data"] = message_data

        def assertDataIs(what):
            self.assertEquals(scratchpad.get("data"), what)

        try:
            consumerset.register_callback(callback)
            it = consumerset.iterconsume()
            publisher.send({"rkey": "foo.xuzzy"}, routing_key="foo.xuzzy")
            it.next()
            assertDataIs({"rkey": "foo.xuzzy"})

            publisher.send({"rkey": "foo.xuzzy"}, routing_key="foo.xuzzy")
            publisher.send({"rkey": "foo.bar"}, routing_key="foo.bar")
            publisher.send({"rkey": "foo.baz"}, routing_key="foo.baz")
            publisher.send({"rkey": "foo.bam"}, routing_key="foo.bam")

            it.next()
            assertDataIs({"rkey": "foo.xuzzy"})
            it.next()
            assertDataIs({"rkey": "foo.bar"})
            it.next()
            assertDataIs({"rkey": "foo.baz"})
            it.next()
            assertDataIs({"rkey": "foo.bam"})

        finally:
            consumerset.close()
            publisher.close()