def ring_queue_check()

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

• 12 lines of code
• 12 McCabe index (conditional complexity)

    def ring_queue_check(self, msgs, count=10):
        """
        Ensure that a ring queue removes lowest priority messages first.
        """
        snd = self.ssn.sender(address("priority-ring-queue", arguments="x-qpid-priorities:10, 'qpid.policy_type':ring, 'qpid.max_count':%s" % count),
                              durable=self.durable())
        for m in msgs: snd.send(m)

        rcv = self.ssn.receiver(snd.target)
        received = []
        try:
            while True: received.append(rcv.fetch(0))
        except Empty: None

        expected = sorted_(msgs, key=lambda x: priority_level(x.priority,10))[len(msgs)-count:]
        expected = sorted_(expected, key=lambda x: priority_level(x.priority,10), reverse=True)
        #print "sent %s; expected %s; got %s" % ([m.priority for m in msgs], [m.priority for m in expected], [m.priority for m in received])
        #print "sent %s; expected %s; got %s" % ([m.content for m in msgs], [m.content for m in expected], [m.content for m in received])
        assert [m.content for m in expected] == [m.content for m in received]