def test_query()

in qpid_tests/broker_0_10/msg_groups.py [0:0]

• 45 lines of code
• 4 McCabe index (conditional complexity)

    def test_query(self):
        """ Verify the queue query method against message groups
        """
        snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," +
                              " node: {x-declare: {arguments:" +
                              " {'qpid.group_header_key':'THE-GROUP'," +
                              "'qpid.shared_msg_group':1}}}}")

        groups = ["A","B","C","A","B","C","A"]
        messages = [Message(content={}, properties={"THE-GROUP": g}) for g in groups]
        index = 0
        for m in messages:
            m.content['index'] = index
            index += 1
            snd.send(m)

        s1 = self.setup_session()
        c1 = s1.receiver("msg-group-q", options={"capacity":0})
        s2 = self.setup_session()
        c2 = s2.receiver("msg-group-q", options={"capacity":0})

        m1 = c1.fetch(0)
        m2 = c2.fetch(0)

        # at this point, group A should be owned by C1, group B by C2, and
        # group C should be available

        # now setup a QMF session, so we can call methods
        self.qmf_session = qmf.console.Session()
        self.qmf_broker = self.qmf_session.addBroker(str(self.broker))
        brokers = self.qmf_session.getObjects(_class="broker")
        assert len(brokers) == 1
        broker = brokers[0]

        # verify the query method call's group information
        rc = broker.query("queue", "msg-group-q")
        assert rc.status == 0
        assert rc.text == "OK"
        results = rc.outArgs['results']
        assert 'qpid.message_group_queue' in results
        q_info = results['qpid.message_group_queue']
        assert 'group_header_key' in q_info and q_info['group_header_key'] == "THE-GROUP"
        assert 'group_state' in q_info and len(q_info['group_state']) == 3
        for g_info in q_info['group_state']:
            assert 'group_id' in g_info
            if g_info['group_id'] == "A":
                assert g_info['msg_count'] == 3
                assert g_info['consumer'] != ""
            elif g_info['group_id'] == "B":
                assert g_info['msg_count'] == 2
                assert g_info['consumer'] != ""
            elif g_info['group_id'] == "C":
                assert g_info['msg_count'] == 2
                assert g_info['consumer'] == ""
            else:
                assert(False)    # should never get here
        self.qmf_session.delBroker(self.qmf_broker)