def test_binding_query_header()

in qpid_tests/broker_0_9/query.py [0:0]

• 43 lines of code
• 1 McCabe index (conditional complexity)

    def test_binding_query_header(self):
        """
        Test that the binding_query method works as expected with headers exchanges
        """
        channel = self.channel
        #setup
        channel.queue_declare(queue="used-queue", exclusive=True)
        channel.queue_declare(queue="unused-queue", exclusive=True)
        channel.queue_bind(exchange="amq.match", queue="used-queue", arguments={"x-match":"all", "a":"A"} )

        # test detection of any binding to specific queue
        response = channel.binding_query(exchange="amq.match", queue="used-queue")
        self.assertEqual(False, response.exchange_not_found)
        self.assertEqual(False, response.queue_not_found)
        self.assertEqual(False, response.queue_not_matched)        

        # test detection of specific binding to any queue
        response = channel.binding_query(exchange="amq.match", arguments={"x-match":"all", "a":"A"})
        self.assertEqual(False, response.exchange_not_found)
        self.assertEqual(False, response.queue_not_found)
        self.assertEqual(False, response.args_not_matched)        

        # test detection of specific binding to specific queue
        response = channel.binding_query(exchange="amq.match", queue="used-queue", arguments={"x-match":"all", "a":"A"})
        self.assertEqual(False, response.exchange_not_found)
        self.assertEqual(False, response.queue_not_found)
        self.assertEqual(False, response.queue_not_matched)        
        self.assertEqual(False, response.args_not_matched)        

        # test unmatched queue, unspecified binding
        response = channel.binding_query(exchange="amq.match", queue="unused-queue")
        self.assertEqual(False, response.exchange_not_found)
        self.assertEqual(False, response.queue_not_found)
        self.assertEqual(True, response.queue_not_matched)        

        # test unspecified queue, unmatched binding
        response = channel.binding_query(exchange="amq.match", arguments={"x-match":"all", "b":"B"})
        self.assertEqual(False, response.exchange_not_found)
        self.assertEqual(False, response.queue_not_found)
        self.assertEqual(True, response.args_not_matched)        

        # test matched queue, unmatched binding
        response = channel.binding_query(exchange="amq.match", queue="used-queue", arguments={"x-match":"all", "b":"B"})
        self.assertEqual(False, response.exchange_not_found)
        self.assertEqual(False, response.queue_not_found)
        self.assertEqual(False, response.queue_not_matched)        
        self.assertEqual(True, response.args_not_matched)        

        # test unmatched queue, matched binding
        response = channel.binding_query(exchange="amq.match", queue="unused-queue", arguments={"x-match":"all", "a":"A"})
        self.assertEqual(False, response.exchange_not_found)
        self.assertEqual(False, response.queue_not_found)
        self.assertEqual(True, response.queue_not_matched)        
        self.assertEqual(False, response.args_not_matched)        

        # test unmatched queue, unmatched binding
        response = channel.binding_query(exchange="amq.match", queue="unused-queue", arguments={"x-match":"all", "b":"B"})
        self.assertEqual(False, response.exchange_not_found)
        self.assertEqual(False, response.queue_not_found)
        self.assertEqual(True, response.queue_not_matched)        
        self.assertEqual(True, response.args_not_matched)        

        #test exchange not found
        self.assertEqual(True, channel.binding_query(exchange="unknown-exchange").exchange_not_found)

        #test queue not found
        self.assertEqual(True, channel.binding_query(exchange="amq.match", queue="unknown-queue").queue_not_found)