Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_xq_name_returns_delay_names(given, expected):
assert xq_name(given) == expected
def flush(self, queue_name):
"""Drop all the messages from a queue.
Parameters:
queue_name(str): The queue to flush.
"""
for name in (queue_name, dq_name(queue_name), xq_name(queue_name)):
self.channel.queue_purge(name)
def _declare_xq_queue(self, queue_name):
return self.channel.queue_declare(queue=xq_name(queue_name), durable=True, arguments={
# This HAS to be a static value since messages are expired
# in order inside of RabbitMQ (head-first).
"x-message-ttl": DEAD_MESSAGE_TTL,
})
def tab_from_q_name(name):
if name == dq_name(name):
return "delayed"
elif name == xq_name(name):
return "failed"
else:
return "standard"
def queue_for_tab(name, tab):
return {
"standard": name,
"delayed": dq_name(name),
"failed": xq_name(name),
}[tab]