def check_ring_lvq()

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

• 23 lines of code
• 9 McCabe index (conditional complexity)

    def check_ring_lvq(self, ring_size, keys, message_count):
        address = "lvq; {create: sender, delete: sender, node: {x-declare:{arguments:{'qpid.last_value_queue_key':lvq-key,'qpid.policy_type':'ring','qpid.max_count':%i}}}}" % ring_size
        snd = self.ssn.sender(address, durable=self.durable())
        counters = {}
        for k in keys:
            counters[k] = 0
        messages = []
        for i in range(message_count):
            k = random.choice(keys)
            counters[k] += 1
            messages.append(create_message(k, "%s-%i" % (k, counters[k])))
        # make sure we have sent at least one message for every key
        for k, v in counters.items():
            if v == 0:
                counters[k] += 1
                messages.append(create_message(k, "%s-%i" % (k, counters[k])))

        for m in messages:
            snd.send(m)

        rcv = self.ssn.receiver("lvq; {mode: browse}")
        retrieved = fetch_all_as_tuples(rcv)
        print([v for k, v in retrieved])

        for k, v in retrieved:
            assert v == "%s-%i" % (k, counters[k])
        assert len(retrieved) <= ring_size