#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
#
from txamqp.client import Closed
from txamqp.queue import Empty
from txamqp.content import Content
from txamqp.testlib import TestBase, supportedBrokers, QPID, OPENAMQ

from twisted.internet.defer import inlineCallbacks

class BasicTests(TestBase):
    """Tests for 'methods' on the amqp basic 'class'"""

    @supportedBrokers(QPID, OPENAMQ)
    @inlineCallbacks
    def test_consume_no_local(self):
        """
        Test that the no_local flag is honoured in the consume method
        """
        channel = self.channel
        #setup, declare two queues:
        yield channel.queue_declare(queue="test-queue-1a", exclusive=True)
        yield channel.queue_declare(queue="test-queue-1b", exclusive=True)
        #establish two consumers one of which excludes delivery of locally sent messages
        yield channel.basic_consume(consumer_tag="local_included", queue="test-queue-1a")
        yield channel.basic_consume(consumer_tag="local_excluded", queue="test-queue-1b", no_local=True)

        #send a message
        channel.basic_publish(routing_key="test-queue-1a", content=Content("consume_no_local"))
        channel.basic_publish(routing_key="test-queue-1b", content=Content("consume_no_local"))

        #check the queues of the two consumers
        excluded = yield self.client.queue("local_excluded")
        included = yield self.client.queue("local_included")
        msg = yield included.get()
        self.assertEqual("consume_no_local", msg.content.body)
        try:
            yield excluded.get(timeout=1)
            self.fail("Received locally published message though no_local=true")
        except Empty: None

    @inlineCallbacks
    def test_consume_exclusive(self):
        """
        Test that the exclusive flag is honoured in the consume method
        """
        channel = self.channel
        #setup, declare a queue:
        yield channel.queue_declare(queue="test-queue-2", exclusive=True)

        #check that an exclusive consumer prevents other consumer being created:
        yield channel.basic_consume(consumer_tag="first", queue="test-queue-2", exclusive=True)
        try:
            yield channel.basic_consume(consumer_tag="second", queue="test-queue-2")
            self.fail("Expected consume request to fail due to previous exclusive consumer")
        except Closed, e:
            self.assertChannelException(403, e.args[0])

        #open new channel and cleanup last consumer:
        channel = yield self.client.channel(2)
        yield channel.channel_open()

        #check that an exclusive consumer cannot be created if a consumer already exists:
        yield channel.basic_consume(consumer_tag="first", queue="test-queue-2")
        try:
            yield channel.basic_consume(consumer_tag="second", queue="test-queue-2", exclusive=True)
            self.fail("Expected exclusive consume request to fail due to previous consumer")
        except Closed, e:
            self.assertChannelException(403, e.args[0])

    @inlineCallbacks
    def test_consume_queue_not_found(self):
        """
        C{basic_consume} fails with a channel exception with a C{404} code when
        the specified queue doesn't exist.
        """
        channel = self.channel
        try:
            #queue specified but doesn't exist:
            yield channel.basic_consume(queue="invalid-queue")
            self.fail("Expected failure when consuming from non-existent queue")
        except Closed, e:
            self.assertChannelException(404, e.args[0])

    @supportedBrokers(QPID, OPENAMQ)
    @inlineCallbacks
    def test_consume_queue_unspecified(self):
        """
        C{basic_consume} fails with a connection exception with a C{503} code
        when no queue is specified.
        """
        channel = self.channel
        try:
            #queue not specified and none previously declared for channel:
            yield channel.basic_consume(queue="")
            self.fail("Expected failure when consuming from unspecified queue")
        except Closed, e:
            self.assertConnectionException(530, e.args[0])

    @inlineCallbacks
    def test_consume_unique_consumers(self):
        """
        Ensure unique consumer tags are enforced
        """
        channel = self.channel
        #setup, declare a queue:
        yield channel.queue_declare(queue="test-queue-3", exclusive=True)

        #check that attempts to use duplicate tags are detected and prevented:
        yield channel.basic_consume(consumer_tag="first", queue="test-queue-3")
        try:
            yield channel.basic_consume(consumer_tag="first", queue="test-queue-3")
            self.fail("Expected consume request to fail due to non-unique tag")
        except Closed, e:
            self.assertConnectionException(530, e.args[0])

    @inlineCallbacks
    def test_cancel(self):
        """
        Test compliance of the basic.cancel method
        """
        channel = self.channel