73 lines
2.7 KiB
Python
73 lines
2.7 KiB
Python
#!/usr/bin/env python3
|
||
"""Детерминированная проверка DLQ-переноса (без гонок тайминга):
|
||
после каждой доставки ждём истечения visibility + тик PeriodicTasks (2.5с).
|
||
Если 5/5 PASS — сервис корректен, флап fifo_dlq_probe — гонка теста."""
|
||
import json
|
||
import time
|
||
import uuid
|
||
|
||
import boto3
|
||
from botocore.config import Config
|
||
|
||
ENDPOINT = "https://sqs.containerk8s.dev.nubes.ru"
|
||
REGION = "us-east-1"
|
||
sqs = boto3.client("sqs", endpoint_url=ENDPOINT, region_name=REGION,
|
||
config=Config(connect_timeout=10, read_timeout=30, retries={"max_attempts": 0}))
|
||
|
||
|
||
def drain(url, n=10):
|
||
msgs = []
|
||
while True:
|
||
b = sqs.receive_message(QueueUrl=url, MaxNumberOfMessages=n).get("Messages", [])
|
||
if not b:
|
||
break
|
||
msgs.extend(b)
|
||
return msgs
|
||
|
||
|
||
def run(i):
|
||
uid = str(uuid.uuid4())[:8]
|
||
dlq_name = "dlqdet-%s-%d" % (uid, i)
|
||
main_name = "maindet-%s-%d" % (uid, i)
|
||
dlq_url = sqs.create_queue(QueueName=dlq_name)["QueueUrl"]
|
||
tenant = dlq_url.split("/")[-2]
|
||
arn = "arn:aws:sqs:%s:%s:%s" % (REGION, tenant, dlq_name)
|
||
policy = json.dumps({"deadLetterTargetArn": arn, "maxReceiveCount": "2"})
|
||
main_url = sqs.create_queue(
|
||
QueueName=main_name,
|
||
Attributes={"VisibilityTimeout": "1", "RedrivePolicy": policy},
|
||
)["QueueUrl"]
|
||
sqs.send_message(QueueUrl=main_url, MessageBody="to-dlq")
|
||
|
||
r1 = sqs.receive_message(QueueUrl=main_url, MaxNumberOfMessages=1).get("Messages", [])
|
||
time.sleep(2.5) # visibility истёк + тик вернул (Retry=1)
|
||
r2 = sqs.receive_message(QueueUrl=main_url, MaxNumberOfMessages=1).get("Messages", [])
|
||
time.sleep(2.5) # visibility истёк + тик: Retry=2 >= maxReceiveCount=2 -> DLQ
|
||
r3 = sqs.receive_message(QueueUrl=main_url, MaxNumberOfMessages=1).get("Messages", [])
|
||
time.sleep(2.0)
|
||
dlq_msgs = drain(dlq_url)
|
||
main_left = drain(main_url)
|
||
|
||
ok = (len(r1) == 1 and len(r2) == 1 and len(r3) == 0
|
||
and len(dlq_msgs) == 1 and dlq_msgs[0]["Body"] == "to-dlq"
|
||
and len(main_left) == 0)
|
||
print("run %d: r1=%d r2=%d r3=%d dlq=%d main_left=%d -> %s"
|
||
% (i, len(r1), len(r2), len(r3), len(dlq_msgs), len(main_left),
|
||
"PASS" if ok else "FAIL"), flush=True)
|
||
try:
|
||
sqs.delete_queue(QueueUrl=main_url)
|
||
sqs.delete_queue(QueueUrl=dlq_url)
|
||
except Exception:
|
||
pass
|
||
return ok
|
||
|
||
|
||
ok = 0
|
||
for i in range(1, 6):
|
||
try:
|
||
if run(i):
|
||
ok += 1
|
||
except Exception as e:
|
||
print("run %d: EXC %s %s" % (i, type(e).__name__, str(e)[:120]), flush=True)
|
||
print("TOTAL: %d/5 PASS" % ok, flush=True)
|