# /// script # requires-python = ">=3.10" # dependencies = [ # "azure-servicebus==7.15.0", # "azure-storage-queue==12.18.1", # "boto3==1.44.3", # "google-cloud-pubsub==2.44.0", # "pika==1.4.4", # ] # [tool.uv] # exclude-newer = "2026-11-15T00:00:00Z" # /// """powerqueue-link checks the free message queue at powerqueue.link through every interface it offers, with each protocol's official Python client library, exactly as an application would (REQ-6314): AMQP 0-9-1 over TLS (pika), Amazon SQS (boto3), Google Cloud Pub/Sub (google-cloud-pubsub), Azure Queue Storage (azure-storage-queue) and Azure Service Bus, over AMQP 1.0 and its HTTPS administration API (azure-servicebus). For each interface it sends messages of random content and sizes, receives them, and verifies that every message came back exactly once with the same bytes and attributes (in order on AMQP 0-9-1), that nothing is left over, and that it deleted what it created. Then it sends through every interface and receives through every other one (20 pairs), and checks each conversion between interfaces against the rule it follows (REQ-6150). Run it with uv, which fetches Python and the libraries itself: uv run https://www.powerqueue.link/downloads/powerqueue_link.py It is the Python version of the Go program in examples/powerqueue-link, and behaves the same; the website offers this one. """ from __future__ import annotations import argparse import base64 import contextlib import dataclasses import json import os import platform import re import secrets import socket import sys import tempfile import threading import time from collections.abc import Callable from pathlib import Path from typing import Any VERSION = "2026.11.15" DEFAULT_HOST = "powerqueue.link" # Time allowed for each interface's whole check: connecting, creating its # entities, the messages each way, the emptiness check and the cleanup. DEFAULT_TIMEOUT_SECS = 60.0 # Exit statuses. EXIT_OK = 0 EXIT_FAILED = 1 EXIT_USAGE = 2 EXIT_REFUSED = 3 EXIT_INTERRUPTED = 130 # The service's tenant identifier format (DEC-6165): a lowercase letter, # then 23 lowercase letters or digits. TENANT_LENGTH = 24 TENANT_PATTERN = re.compile(r"^[a-z][a-z0-9]{23}$") FIRST_ALPHABET = "abcdefghijklmnopqrstuvwxyz" REST_ALPHABET = "abcdefghijklmnopqrstuvwxyz0123456789" # The random part of this run's entity names. RUN_ID_LENGTH = 8 # The tenant file, under the user's configuration directory, and its mode: # the identifier is the tenant's only secret. TENANT_DIR_NAME = "powerqueue-link" TENANT_FILE_NAME = "tenant" TENANT_FILE_MODE = 0o600 # The ports the service listens on. HTTPS_PORT = 443 AMQPS_PORT = 5673 # AMQP 0-9-1 over TLS (5671 is Service Bus's) SERVICE_BUS_PORT = 5671 # Service Bus AMQP 1.0 over TLS, as its SDKs use SERVICE_SUBDOMAINS = ("sqs", "pubsub", "azure", "amqp") # The only addresses a process's own listener may have for the guard to # allow connecting to it (pika's wake-up socket pair). LOOPBACK_ADDRESSES = frozenset({"127.0.0.1", "::1"}) # Request pacing (REQ-6308): the service allows each client address 100 # requests a second with bursts of 200 and bans addresses that keep # exceeding that; the program stays a tenth of the way there. REQUESTS_PER_SECOND = 10.0 # Body sizes each interface sends, one message each: a byte, small, a page, # and two large enough to cross frame and buffer boundaries while inside # every protocol's limit (Azure Queue Storage's 64 KiB is the smallest). MESSAGE_SIZES = (1, 37, 4096, 16384, 60000) # Text bodies: ASCII, punctuation XML and JSON must escape, and characters # outside ASCII, so encodings are checked as well as lengths. Every one is # allowed in an SQS message and in an Azure queue message (XML 1.0 text). TEXT_ALPHABET = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789 <>&\"'\\/{}é✓日本\n\t" ATTR_ID = "demo-id" ATTR_SEQ = "demo-seq" # Protocol limits and waits the checks use. SQS_REGION = "us-east-1" # any: the service ignores it; the SDK signs with one SQS_SECRET = "anything" # signatures are not checked SQS_WAIT_SECS = 5 # ReceiveMessage WaitTimeSeconds: 0 to 20 SQS_EMPTY_WAIT_SECS = 2 SQS_MAX_MESSAGES = 10 # ReceiveMessage MaxNumberOfMessages: 1 to 10 AZURE_ACCOUNT_KEY = "YW55dGhpbmc=" # any base64 value: signatures are not checked AZURE_BATCH = 32 # Get Messages numofmessages: 1 to 32 SERVICE_BUS_KEY = "anything" # signs the tokens; not checked SERVICE_BUS_BATCH = 10 SERVICE_BUS_WAIT_SECS = 5 SERVICE_BUS_LOCK_SECS = 45 # Service Bus: 5 seconds to 5 minutes PUBSUB_BATCH = 10 PUBSUB_PULL_SECS = 5.0 PUBSUB_QUIET_SECS = 3.0 # silence required after every message was acknowledged POLL_SECS = 0.2 # spacing of polls on interfaces without a long poll SDK_CALL_TIMEOUT_SECS = 30 # How long cleanups may take after a failure or Ctrl-C. CLEANUP_TIMEOUT_SECS = 20 PUBSUB_TOPIC_SUFFIX = "-topic" # The markers REQ-6150 reserves. MARKER_BODY_ENCODING = "x-pq-body-encoding" MARKER_BASE64_ATTRIBUTES = "x-pq-base64-attributes" ENCODING_BASE64 = "base64" ENCODING_JSON = "json" SQS_JSON_TYPE = "String.json" # an AMQP table header read through SQS (rule 7) BINARY_BODY_SIZE = 256 # random bytes: certainly not text CLIP_BYTES = 48 # how much of a body a failure message shows # Attribute types, as SQS names them; the other protocols' typed values # are described with the same words. TYPE_STRING = "String" TYPE_NUMBER = "Number" TYPE_BINARY = "Binary" TYPE_TABLE = "Table" # an AMQP 0-9-1 field table, which no other protocol has # Interface names, as the cross-protocol ends name themselves. NAME_AMQP = "AMQP 0-9-1" NAME_SQS = "SQS" NAME_PUBSUB = "Pub/Sub" NAME_AZURE = "Azure Queue" NAME_SERVICE_BUS = "Service Bus" CROSS_CODES = {NAME_AMQP: "amqp", NAME_SQS: "sqs", NAME_PUBSUB: "ps", NAME_AZURE: "az", NAME_SERVICE_BUS: "sb"} # SQS error codes for throttling and quotas, which the service answers when # a limit is reached; HTTP statuses and gRPC codes likewise. SQS_REFUSAL_CODES = frozenset( {"Throttling", "ThrottlingException", "RequestThrottled", "OverLimit", "AWS.SimpleQueueService.OverLimit"} ) HTTP_REFUSAL_STATUSES = frozenset({429, 503}) GRPC_REFUSAL_CODES = frozenset({"RESOURCE_EXHAUSTED", "UNAVAILABLE"}) SERVICE_BUS_BUSY = "com.microsoft:server-busy" # Environment variables that would make a client library read the user's # files or cloud configuration, or route through a proxy the guard cannot # vouch for; they are removed for this process before any library loads. ENV_PREFIXES_REMOVED = ("AWS_", "AZURE_", "GOOGLE_", "CLOUDSDK_", "GRPC_") ENV_NAMES_REMOVED = frozenset( { "HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "NO_PROXY", "http_proxy", "https_proxy", "all_proxy", "no_proxy", "PUBSUB_EMULATOR_HOST", "REQUESTS_CA_BUNDLE", "CURL_CA_BUNDLE", } ) class SetupError(Exception): """A problem before any check runs: the tenant, the arguments.""" class CheckFailed(Exception): """A check found the service behaving otherwise than it should.""" class Refused(Exception): """The service refused a request for a limit, a ban or being busy: the run stops instead of retrying.""" class DestinationRefused(OSError): """The guard refused a connection to somewhere other than the service.""" # -------------------------------------------------------------------------- # Tenant identifier and its file # -------------------------------------------------------------------------- def random_identifier(length: int) -> str: """A lowercase letter followed by length-1 lowercase letters or digits.""" return secrets.choice(FIRST_ALPHABET) + "".join(secrets.choice(REST_ALPHABET) for _ in range(length - 1)) def new_tenant() -> str: return random_identifier(TENANT_LENGTH) def valid_tenant(tenant: str) -> bool: return bool(TENANT_PATTERN.fullmatch(tenant)) def config_dir() -> Path: """The user's configuration directory, as each platform defines it (the same one the Go program uses, so both share the tenant).""" system = platform.system() if system == "Windows": appdata = os.environ.get("APPDATA") if not appdata: raise SetupError("cannot find your configuration directory: APPDATA is not set; give --tenant") return Path(appdata) if system == "Darwin": return Path.home() / "Library" / "Application Support" xdg = os.environ.get("XDG_CONFIG_HOME") if xdg: return Path(xdg) return Path.home() / ".config" def tenant_file() -> Path: return config_dir() / TENANT_DIR_NAME / TENANT_FILE_NAME def load_tenant(path: Path, fresh: bool) -> tuple[str, bool]: """The saved tenant, or a new one saved there when none exists or fresh is set; also whether it was created. A file that cannot be read or holds no valid identifier is an error, never silently replaced.""" if not fresh: try: text = path.read_text(encoding="ascii").strip() except FileNotFoundError: pass except (OSError, UnicodeDecodeError) as e: raise SetupError(f"cannot read the tenant file {path}: {e}; give --tenant or --new-tenant") from e else: if not valid_tenant(text): raise SetupError( f"the tenant file {path} holds no valid tenant identifier; " "give --new-tenant to replace it or --tenant to use another" ) return text, False tenant = new_tenant() try: save_tenant(path, tenant) except OSError as e: raise SetupError(f"cannot save the tenant identifier in {path}: {e}; give --tenant to name one") from e return tenant, True def save_tenant(path: Path, tenant: str) -> None: """Writes the file atomically, readable only by the user.""" path.parent.mkdir(parents=True, exist_ok=True) fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=".tenant-") try: os.chmod(tmp, TENANT_FILE_MODE) with os.fdopen(fd, "w", encoding="ascii") as f: f.write(tenant + "\n") os.replace(tmp, path) except BaseException: with contextlib.suppress(OSError): os.unlink(tmp) raise # -------------------------------------------------------------------------- # The network guard # -------------------------------------------------------------------------- class Guard: """Allows connections to the service's interfaces only: sqs., pubsub., azure. and amqp. on 443, and amqp. on 5673 and 5671. Installed by wrapping socket.getaddrinfo, socket.create_connection and socket.socket.connect/connect_ex, which every client library here uses except one: Google's gRPC library (Pub/Sub) connects from its C core, which Python cannot intercept, so the Pub/Sub connection is not guarded; its destination is fixed by the endpoint the program passes, pubsub.:443. A name is checked before it is resolved; a connection to an address is allowed only to an address an allowed name resolved to in this run, on that name's allowed port. The one other connection allowed is to a loopback listener this process itself opened: pika wakes its own event loop through a socket pair it makes that way.""" def __init__(self, host: str) -> None: host = canonical_host(host) self.allowed: set[tuple[str, int]] = {(f"{sub}.{host}", HTTPS_PORT) for sub in SERVICE_SUBDOMAINS} self.allowed |= {(f"amqp.{host}", port) for port in (AMQPS_PORT, SERVICE_BUS_PORT)} self.resolved: set[tuple[str, int]] = set() # Loopback (address, port) pairs this process listens on. self.own_listeners: set[tuple[str, int]] = set() self._lock = threading.Lock() self._originals: dict[str, Any] = {} def note_listening(self, sock: socket.socket) -> None: """Records a loopback listener of this process, which it may connect to (pika's wake-up socket pair); any other listener is not recorded.""" if sock.family not in (socket.AF_INET, socket.AF_INET6): return address, port = sock.getsockname()[:2] if address in LOOPBACK_ADDRESSES: with self._lock: self.own_listeners.add((address, port)) def allowed_list(self) -> list[str]: return sorted(f"{h}:{p}" for h, p in self.allowed) def allows_name(self, host: str, port: int) -> bool: return (canonical_host(host), port) in self.allowed def note_resolved(self, host: str, port: int | None, infos: list) -> None: """Records the addresses an allowed name resolved to, for its allowed ports (all of them when the call named no port).""" name = canonical_host(host) ports = [p for h, p in self.allowed if h == name and (port is None or p == port)] with self._lock: for info in infos: address = info[4][0] for p in ports: self.resolved.add((address, p)) def allows_address(self, address: str, port: int) -> bool: with self._lock: return (address, port) in self.resolved or (address, port) in self.own_listeners def check_connect(self, address: Any, family: int) -> None: if family not in (socket.AF_INET, socket.AF_INET6): raise DestinationRefused(f"refused: this program connects only to the PowerQueue service, not {address!r}") host, port = address[0], int(address[1]) if not self.allows_address(host, port): raise DestinationRefused( f"refused: this program connects only to the PowerQueue service, not {host}:{port}" ) def install(self) -> None: guard = self original_getaddrinfo = socket.getaddrinfo original_create_connection = socket.create_connection original_connect = socket.socket.connect original_connect_ex = socket.socket.connect_ex self._originals = { "getaddrinfo": original_getaddrinfo, "create_connection": original_create_connection, "connect": original_connect, "connect_ex": original_connect_ex, } def getaddrinfo(host, port, *args, **kwargs): infos = original_getaddrinfo(host, port, *args, **kwargs) if isinstance(host, (str, bytes)): name = host.decode() if isinstance(host, bytes) else host number = port_number(port) if any(h == canonical_host(name) and (number is None or p == number) for h, p in guard.allowed): guard.note_resolved(name, number, infos) return infos def create_connection(address, *args, **kwargs): host, port = address[0], port_number(address[1]) if port is None or not (guard.allows_name(host, port) or guard.allows_address(host, port)): raise DestinationRefused( f"refused: this program connects only to the PowerQueue service, not {host}:{address[1]}" ) return original_create_connection(address, *args, **kwargs) def connect(sock, address): guard.check_connect(address, sock.family) return original_connect(sock, address) def connect_ex(sock, address): guard.check_connect(address, sock.family) return original_connect_ex(sock, address) original_listen = socket.socket.listen self._originals["listen"] = original_listen def listen(sock, *args): original_listen(sock, *args) guard.note_listening(sock) socket.getaddrinfo = getaddrinfo socket.create_connection = create_connection socket.socket.connect = connect socket.socket.connect_ex = connect_ex socket.socket.listen = listen def uninstall(self) -> None: if not self._originals: return socket.getaddrinfo = self._originals["getaddrinfo"] socket.create_connection = self._originals["create_connection"] socket.socket.connect = self._originals["connect"] socket.socket.connect_ex = self._originals["connect_ex"] socket.socket.listen = self._originals["listen"] self._originals = {} def canonical_host(host: str) -> str: return host.lower().rstrip(".") def port_number(port: Any) -> int | None: if port is None: return None if isinstance(port, bytes): port = port.decode() try: return int(port) except (TypeError, ValueError): return None def sanitise_environment() -> None: """Keeps the client libraries away from the user's files, cloud configuration and proxies: their configuration files point at the null device, instance metadata is off, and every variable that could steer them is removed for this process.""" for name in list(os.environ): if name in ENV_NAMES_REMOVED or name.startswith(ENV_PREFIXES_REMOVED): del os.environ[name] os.environ["AWS_CONFIG_FILE"] = os.devnull os.environ["AWS_SHARED_CREDENTIALS_FILE"] = os.devnull os.environ["AWS_EC2_METADATA_DISABLED"] = "true" # requests (under the Azure libraries) would read ~/.netrc. os.environ["NETRC"] = os.devnull # -------------------------------------------------------------------------- # Pacing and refusals # -------------------------------------------------------------------------- class Pacer: """Spaces requests at REQUESTS_PER_SECOND across every interface.""" def __init__(self, per_second: float) -> None: self.interval = 1.0 / per_second self.next = time.monotonic() self.lock = threading.Lock() def wait(self) -> None: with self.lock: now = time.monotonic() delay = self.next - now self.next = max(now, self.next) + self.interval if delay > 0: time.sleep(delay) def refusal_reason(e: BaseException) -> str | None: """Why e is the service refusing a request for a limit, a ban or being busy, whichever protocol it came through; None when it is not one. The exceptions are recognised by their shape, so this needs no library.""" if isinstance(e, Refused): return str(e) response = getattr(e, "response", None) if isinstance(response, dict): # botocore ClientError code = response.get("Error", {}).get("Code", "") status = response.get("ResponseMetadata", {}).get("HTTPStatusCode") if code in SQS_REFUSAL_CODES: return f"SQS {code}" if status in HTTP_REFUSAL_STATUSES: return f"HTTP {status}" status_code = getattr(e, "status_code", None) # azure.core HttpResponseError if isinstance(status_code, int) and status_code in HTTP_REFUSAL_STATUSES: return f"HTTP {status_code}" grpc_code = getattr(e, "grpc_status_code", None) # google.api_core exceptions name = getattr(grpc_code, "name", None) if name in GRPC_REFUSAL_CODES: return f"gRPC {name}" if type(e).__name__ == "ServiceBusServerBusyError" or SERVICE_BUS_BUSY in str(e): return "Service Bus server busy" return None class Env: """What every check shares: where it connects, the pacing of requests, and the refusal that stops the whole run.""" def __init__(self, host: str, tenant: str, guard: Guard) -> None: self.host = host self.tenant = tenant self.guard = guard self.pacer = Pacer(REQUESTS_PER_SECOND) def call(self, fn: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: """One request: paced, and a refusal stops the run.""" self.pacer.wait() try: return fn(*args, **kwargs) except Exception as e: reason = refusal_reason(e) if reason is not None: raise Refused(reason) from e raise # The clients, configured explicitly: endpoint, dummy credentials, no # retries (the run stops at a refusal instead). def sqs(self): import botocore.config import botocore.session session = botocore.session.Session() return session.create_client( "sqs", region_name=SQS_REGION, endpoint_url=f"https://sqs.{self.host}", aws_access_key_id=self.tenant, aws_secret_access_key=SQS_SECRET, config=botocore.config.Config( retries={"max_attempts": 1, "mode": "standard"}, connect_timeout=SDK_CALL_TIMEOUT_SECS, read_timeout=SDK_CALL_TIMEOUT_SECS, proxies={}, ), ) def amqp(self): import ssl import pika host = f"amqp.{self.host}" params = pika.ConnectionParameters( host=host, port=AMQPS_PORT, virtual_host=self.tenant, credentials=pika.PlainCredentials("user", "password"), ssl_options=pika.SSLOptions(ssl.create_default_context(), server_hostname=host), connection_attempts=1, socket_timeout=SDK_CALL_TIMEOUT_SECS, blocked_connection_timeout=SDK_CALL_TIMEOUT_SECS, ) return self.call(pika.BlockingConnection, params) def pubsub(self): """The Pub/Sub publisher and subscriber, over one TLS channel to pubsub.:443, gRPC's own retries off. The channel is gRPC's C core, which the guard cannot see.""" import grpc from google.pubsub_v1 import PublisherClient, SubscriberClient from google.pubsub_v1.services.publisher.transports import PublisherGrpcTransport from google.pubsub_v1.services.subscriber.transports import SubscriberGrpcTransport target = f"pubsub.{self.host}:{HTTPS_PORT}" options = [("grpc.enable_retries", 0)] publisher_channel = grpc.secure_channel(target, grpc.ssl_channel_credentials(), options=options) subscriber_channel = grpc.secure_channel(target, grpc.ssl_channel_credentials(), options=options) publisher = PublisherClient(transport=PublisherGrpcTransport(channel=publisher_channel)) subscriber = SubscriberClient(transport=SubscriberGrpcTransport(channel=subscriber_channel)) return publisher, subscriber def azure_queue(self, queue: str): from azure.storage.queue import QueueClient return QueueClient( account_url=f"https://azure.{self.host}/{self.tenant}", queue_name=queue, credential={"account_name": self.tenant, "account_key": AZURE_ACCOUNT_KEY}, retry_total=0, connection_timeout=SDK_CALL_TIMEOUT_SECS, read_timeout=SDK_CALL_TIMEOUT_SECS, ) def connection_string(self) -> str: return f"Endpoint=sb://amqp.{self.host}/;SharedAccessKeyName={self.tenant};SharedAccessKey={SERVICE_BUS_KEY}" def service_bus(self): from azure.servicebus import ServiceBusClient return ServiceBusClient.from_connection_string(self.connection_string(), retry_total=0) def service_bus_admin(self): from azure.servicebus.management import ServiceBusAdministrationClient return ServiceBusAdministrationClient.from_connection_string(self.connection_string(), retry_total=0) # -------------------------------------------------------------------------- # Messages and verification # -------------------------------------------------------------------------- @dataclasses.dataclass class Message: """One message as sent, and what a receive must give back.""" id: str body: bytes attrs: dict[str, str] @dataclasses.dataclass class Received: body: bytes attrs: dict[str, str] def random_text(size: int) -> bytes: """Random text of exactly size bytes in UTF-8.""" out = bytearray() while len(out) < size: ch = secrets.choice(TEXT_ALPHABET).encode() if len(out) + len(ch) > size: ch = b"x" # one byte, to land on size exactly out += ch return bytes(out) def new_messages(run_id: str, text: bool) -> list[Message]: """One message per MESSAGE_SIZES entry: random bytes, or random text for interfaces whose bodies are text.""" out = [] for i, size in enumerate(MESSAGE_SIZES): body = random_text(size) if text else secrets.token_bytes(size) mid = f"{run_id}-{i}" out.append(Message(mid, body, {ATTR_ID: mid, ATTR_SEQ: str(i)})) return out def verify(sent: list[Message], got: list[Received], ordered: bool) -> None: """Each sent message exactly once, with the same body and attributes and, when ordered, in send order; extra attributes are the protocol's own.""" by_id = {m.id: m for m in sent} seen: set[str] = set() order = [] for r in got: mid = r.attrs.get(ATTR_ID, "") want = by_id.get(mid) if want is None: raise CheckFailed(f"received a message that was not sent: id {mid!r}") if mid in seen: raise CheckFailed(f"received a message twice: {mid}") seen.add(mid) order.append(mid) if r.body != want.body: raise CheckFailed( f"body differs from what was sent: {mid} ({len(want.body)} B sent, {len(r.body)} received)" ) for k, v in want.attrs.items(): if r.attrs.get(k) != v: raise CheckFailed(f"attribute differs from what was sent: {mid} {k} = {r.attrs.get(k)!r}, sent {v!r}") if len(seen) != len(sent): raise CheckFailed(f"sent messages were not received: {len(seen)} of {len(sent)} received") if ordered and order != [m.id for m in sent]: raise CheckFailed(f"messages arrived out of send order: {order}") def verify_cross(sent: list[Message], got: list[Received], with_attrs: bool) -> None: """A cross-protocol delivery of text messages, matched by body (unique random text): each exactly once with identical bytes and, when both interfaces carry attributes, the same attribute values.""" by_body = {m.body: m for m in sent} seen: set[str] = set() for r in got: want = by_body.get(r.body) if want is None: raise CheckFailed(f"body differs from what was sent: a {len(r.body)}-byte body matching nothing sent") if want.id in seen: raise CheckFailed(f"received a message twice: {want.id}") seen.add(want.id) if with_attrs: for k, v in want.attrs.items(): if r.attrs.get(k) != v: raise CheckFailed(f"attribute differs: {want.id} {k} = {r.attrs.get(k)!r}, sent {v!r}") if len(seen) != len(sent): raise CheckFailed(f"sent messages were not received: {len(seen)} of {len(sent)} received") # -------------------------------------------------------------------------- # Narration # -------------------------------------------------------------------------- class Narrator: """Prints what the program is doing, one indented line per step, with the time since the check began; quiet narrates nothing.""" def __init__(self, quiet: bool) -> None: self.quiet = quiet self.start = time.monotonic() def step(self, text: str) -> None: if not self.quiet: print(f" {time.monotonic() - self.start:6.2f}s {text}", flush=True) def sizes_text(msgs: list[Message]) -> str: return ", ".join(f"{len(m.body)} B" for m in msgs) def entity_name(run_id: str) -> str: """This run's queue or topic name: a letter first, as Pub/Sub requires, and only characters every protocol allows.""" return f"plk-{run_id}" class Cleanups: """What a check created, deleted however the check ends, newest first.""" def __init__(self, say: Narrator) -> None: self.say = say self.actions: list[tuple[str, Callable[[], Any]]] = [] def add(self, what: str, action: Callable[[], Any]) -> None: self.actions.append((what, action)) def run(self) -> list[str]: errors = [] while self.actions: what, action = self.actions.pop() try: action() except Exception as e: # noqa: BLE001 - a failed cleanup is reported, and the rest still run errors.append(f"delete {what}: {e}") return errors def describe(e: BaseException) -> str: """A failure for the report: a check's own message as it is, anything else (a library's exception) with its type.""" if isinstance(e, CheckFailed): return str(e) return f"{type(e).__name__}: {e}" def deadline_after(secs: float) -> float: return time.monotonic() + secs def expired(deadline: float) -> bool: return time.monotonic() >= deadline # -------------------------------------------------------------------------- # The six interface checks # -------------------------------------------------------------------------- def amqp_headers_text(headers: dict | None) -> dict[str, str]: out = {} for k, v in (headers or {}).items(): if isinstance(v, bytes): with contextlib.suppress(UnicodeDecodeError): out[k] = v.decode() elif isinstance(v, str): out[k] = v return out def check_amqp(env: Env, run_id: str, say: Narrator, timeout: float, cleanups: Cleanups) -> None: import pika queue = entity_name(run_id) sent = new_messages(run_id, text=False) deadline = deadline_after(timeout) say.step(f"connecting to amqps://amqp.{env.host}:{AMQPS_PORT}, virtual host {env.tenant}") conn = env.amqp() cleanups.add("the AMQP 0-9-1 connection", conn.close) ch = conn.channel() say.step(f"declaring durable queue {queue}") env.call(ch.queue_declare, queue, durable=True) cleanups.add(f"queue {queue}", lambda: ch.queue_delete(queue)) say.step("turning on publisher confirms, so the broker acknowledges each message it has stored") ch.confirm_delivery() for m in sent: props = pika.BasicProperties( content_type="application/octet-stream", delivery_mode=2, message_id=m.id, headers=dict(m.attrs) ) env.call(ch.basic_publish, "", queue, m.body, props, mandatory=True) say.step(f"published {m.id} ({len(m.body)} B, persistent, with headers and message-id); broker confirmed it") say.step("reading the messages back with basic.get, acknowledging each") got = [] while len(got) < len(sent): method, props, body = env.call(ch.basic_get, queue, auto_ack=False) if method is None: if expired(deadline): raise CheckFailed(f"sent messages were not received: {len(got)} of {len(sent)} in time") time.sleep(POLL_SECS) continue ch.basic_ack(method.delivery_tag) attrs = amqp_headers_text(props.headers) if props.message_id != attrs.get(ATTR_ID): raise CheckFailed(f"message-id {props.message_id!r} differs from header {attrs.get(ATTR_ID)!r}") got.append(Received(body, attrs)) say.step(f"received {len(got)} messages; verifying bodies, headers, message-ids and send order") verify(sent, got, ordered=True) say.step(f"all {len(sent)} match, in order; checking the queue is now empty") method, _, _ = env.call(ch.basic_get, queue, auto_ack=False) if method is not None: raise CheckFailed("the queue still held messages after all were settled") say.step(f"empty; deleting queue {queue}") def sqs_attrs(attrs: dict[str, str]) -> dict: return {k: {"DataType": TYPE_STRING, "StringValue": v} for k, v in attrs.items()} def check_sqs(env: Env, run_id: str, say: Narrator, timeout: float, cleanups: Cleanups) -> None: sent = new_messages(run_id, text=True) deadline = deadline_after(timeout) sqs = env.sqs() queue = entity_name(run_id) say.step(f"creating queue {queue} at https://sqs.{env.host}") url = env.call(sqs.create_queue, QueueName=queue)["QueueUrl"] cleanups.add(f"queue {queue}", lambda: sqs.delete_queue(QueueUrl=url)) for m in sent: env.call(sqs.send_message, QueueUrl=url, MessageBody=m.body.decode(), MessageAttributes=sqs_attrs(m.attrs)) say.step(f"sent {m.id} ({len(m.body)} B of text, with string message attributes)") say.step("long-polling ReceiveMessage, deleting each message received") got = [] while len(got) < len(sent): if expired(deadline): raise CheckFailed(f"sent messages were not received: {len(got)} of {len(sent)} in time") out = env.call( sqs.receive_message, QueueUrl=url, MaxNumberOfMessages=SQS_MAX_MESSAGES, WaitTimeSeconds=SQS_WAIT_SECS, MessageAttributeNames=["All"], ) for m in out.get("Messages", []): attrs = {k: v.get("StringValue", "") for k, v in m.get("MessageAttributes", {}).items()} got.append(Received(m["Body"].encode(), attrs)) env.call(sqs.delete_message, QueueUrl=url, ReceiptHandle=m["ReceiptHandle"]) say.step(f"received {len(got)} messages; verifying bodies and message attributes") verify(sent, got, ordered=False) say.step(f"all {len(sent)} match; a {SQS_EMPTY_WAIT_SECS} s long poll must now find the queue empty") left = env.call(sqs.receive_message, QueueUrl=url, WaitTimeSeconds=SQS_EMPTY_WAIT_SECS) if left.get("Messages"): raise CheckFailed("the queue still held messages after all were settled") say.step("empty; deleting the queue") def pubsub_pull(env: Env, subscriber, subscription: str, max_messages: int) -> list: """One pull, acknowledging what it got; an answer of nothing in time is an empty list.""" from google.api_core import exceptions try: resp = env.call( subscriber.pull, subscription=subscription, max_messages=max_messages, retry=None, timeout=PUBSUB_PULL_SECS, ) except exceptions.DeadlineExceeded: return [] got = list(resp.received_messages) if got: env.call( subscriber.acknowledge, subscription=subscription, ack_ids=[m.ack_id for m in got], retry=None, timeout=SDK_CALL_TIMEOUT_SECS, ) return [m.message for m in got] def check_pubsub(env: Env, run_id: str, say: Narrator, timeout: float, cleanups: Cleanups) -> None: from google.pubsub_v1.types import PubsubMessage sent = new_messages(run_id, text=False) deadline = deadline_after(timeout) say.step(f"connecting to pubsub.{env.host}:{HTTPS_PORT} (gRPC over TLS), project {env.tenant}, no credentials") publisher, subscriber = env.pubsub() project = f"projects/{env.tenant}" topic = f"{project}/topics/{entity_name(run_id)}" subscription = f"{project}/subscriptions/{entity_name(run_id)}-sub" say.step(f"creating topic {topic} and subscription {subscription}") env.call(publisher.create_topic, name=topic, retry=None, timeout=SDK_CALL_TIMEOUT_SECS) cleanups.add(f"topic {topic}", lambda: publisher.delete_topic(topic=topic, retry=None)) env.call(subscriber.create_subscription, name=subscription, topic=topic, retry=None, timeout=SDK_CALL_TIMEOUT_SECS) cleanups.add(f"subscription {subscription}", lambda: subscriber.delete_subscription(subscription=subscription)) for m in sent: env.call( publisher.publish, topic=topic, messages=[PubsubMessage(data=m.body, attributes=m.attrs)], retry=None, timeout=SDK_CALL_TIMEOUT_SECS, ) say.step( f"published {len(sent)} messages with attributes ({sizes_text(sent)}); each publish acknowledged with a message id" ) say.step("pulling from the subscription, acknowledging each") got = [] while len(got) < len(sent): if expired(deadline): raise CheckFailed(f"sent messages were not received: {len(got)} of {len(sent)} in time") for m in pubsub_pull(env, subscriber, subscription, PUBSUB_BATCH): got.append(Received(bytes(m.data), dict(m.attributes))) say.step(f"received {len(got)} messages; verifying bodies and attributes") verify(sent, got, ordered=False) say.step(f"all {len(sent)} match; pulling for {PUBSUB_QUIET_SECS:g}s more to check nothing is redelivered") quiet_until = deadline_after(PUBSUB_QUIET_SECS) while not expired(quiet_until): if pubsub_pull(env, subscriber, subscription, PUBSUB_BATCH): raise CheckFailed("the subscription redelivered messages after all were acknowledged") time.sleep(POLL_SECS) say.step("nothing redelivered; deleting the subscription and topic") def check_azure_queue(env: Env, run_id: str, say: Narrator, timeout: float, cleanups: Cleanups) -> None: sent = new_messages(run_id, text=True) by_body = {m.body: m for m in sent} deadline = deadline_after(timeout) queue = entity_name(run_id) q = env.azure_queue(queue) say.step(f"creating queue https://azure.{env.host}/{env.tenant}/{queue}") env.call(q.create_queue) cleanups.add(f"queue {queue}", q.delete_queue) for m in sent: env.call(q.send_message, m.body.decode()) say.step(f"enqueued {m.id} ({len(m.body)} B of text)") say.step("dequeuing, deleting each message; each must be on its first dequeue") got = [] while len(got) < len(sent): batch = env.call(lambda: list(q.receive_messages(messages_per_page=AZURE_BATCH, max_messages=AZURE_BATCH))) for m in batch: if m.dequeue_count != 1: raise CheckFailed(f"received a message twice: dequeue count {m.dequeue_count}") body = m.content.encode() want = by_body.get(body) got.append(Received(body, dict(want.attrs) if want else {})) env.call(q.delete_message, m) if not batch: if expired(deadline): raise CheckFailed(f"sent messages were not received: {len(got)} of {len(sent)} in time") time.sleep(POLL_SECS) say.step(f"received {len(got)} messages; verifying their text") verify(sent, got, ordered=False) say.step(f"all {len(sent)} match; peeking to check the queue is empty") if env.call(q.peek_messages): raise CheckFailed("the queue still held messages after all were settled") say.step("empty; deleting the queue") def check_service_bus_admin(env: Env, run_id: str, say: Narrator, timeout: float, cleanups: Cleanups) -> None: import datetime admin = env.service_bus_admin() name = entity_name(run_id) + "-admin" lock = datetime.timedelta(seconds=SERVICE_BUS_LOCK_SECS) say.step(f"creating queue {name} with lock duration {SERVICE_BUS_LOCK_SECS} s") env.call(admin.create_queue, name, lock_duration=lock) cleanups.add(f"queue {name}", lambda: admin.delete_queue(name)) say.step("reading it back; its name and lock duration must be what was set") got = env.call(admin.get_queue, name) if got.name != name or got.lock_duration != lock: raise CheckFailed( f"the administration API returned a queue unlike the one created: {got.name} {got.lock_duration}" ) say.step("listing queues; it must be there") names = env.call(lambda: [q.name for q in admin.list_queues()]) if name not in names: raise CheckFailed("the administration API did not list the queue created") say.step("found; deleting it") def service_bus_body(m) -> bytes: """A received message's body as bytes (a data body is a sequence of byte sections).""" body = m.body if isinstance(body, (bytes, bytearray)): return bytes(body) return b"".join(body) def text_or_none(v: Any) -> str | None: if isinstance(v, str): return v if isinstance(v, (bytes, bytearray)): try: return bytes(v).decode() except UnicodeDecodeError: return None return None def service_bus_props_text(props: dict | None) -> dict[str, str]: """azure-servicebus returns received property names and string values as bytes; as text.""" out = {} for k, v in (props or {}).items(): key = text_or_none(k) value = text_or_none(v) if key is not None and value is not None: out[key] = value return out def check_service_bus(env: Env, run_id: str, say: Narrator, timeout: float, cleanups: Cleanups) -> None: from azure.servicebus import ServiceBusMessage sent = new_messages(run_id, text=False) deadline = deadline_after(timeout) queue = entity_name(run_id) client = env.service_bus() cleanups.add("the Service Bus client", client.close) admin = env.service_bus_admin() cleanups.add(f"queue {queue}", lambda: admin.delete_queue(queue)) say.step(f"connecting over AMQP 1.0 with TLS to amqp.{env.host}:{SERVICE_BUS_PORT}; sending to queue {queue}") sender = env.call(client.get_queue_sender, queue) with sender: for m in sent: env.call( sender.send_messages, ServiceBusMessage(m.body, message_id=m.id, application_properties=dict(m.attrs)), ) say.step(f"sent {m.id} ({len(m.body)} B, with MessageID and application properties)") say.step("receiving, completing each; each must be on its first delivery") got = [] receiver = env.call(client.get_queue_receiver, queue, max_wait_time=SERVICE_BUS_WAIT_SECS) with receiver: while len(got) < len(sent): if expired(deadline): raise CheckFailed(f"sent messages were not received: {len(got)} of {len(sent)} in time") for m in env.call(receiver.receive_messages, max_message_count=SERVICE_BUS_BATCH): attrs = service_bus_props_text(m.application_properties) if m.message_id != attrs.get(ATTR_ID): raise CheckFailed(f"MessageID {m.message_id!r} differs from property {attrs.get(ATTR_ID)!r}") # The AMQP header's delivery-count: deliveries that failed # before this one (AMQP 1.0 part 3 section 3.2.1). if m.delivery_count: raise CheckFailed(f"received a message twice: delivery count {m.delivery_count}") got.append(Received(service_bus_body(m), attrs)) env.call(receiver.complete_message, m) say.step(f"received {len(got)} messages; verifying bodies, properties and MessageIDs") verify(sent, got, ordered=False) say.step(f"all {len(sent)} match; peeking to check the queue is empty") if env.call(receiver.peek_messages, max_message_count=1): raise CheckFailed("the queue still held messages after all were settled") say.step(f"empty; deleting queue {queue} through the administration API") # -------------------------------------------------------------------------- # Cross-protocol: one queue, several interfaces # -------------------------------------------------------------------------- @dataclasses.dataclass class PayloadAttr: """One attribute to send, typed.""" name: str type: str text: str = "" data: bytes = b"" table: dict | None = None @dataclasses.dataclass class Payload: """One message to send through any interface; amqp_value, when set, is sent by Service Bus as an AMQP value body (nothing else can send one).""" body: bytes = b"" attrs: list[PayloadAttr] = dataclasses.field(default_factory=list) amqp_value: Any = None @dataclasses.dataclass class Attr: """One attribute as an interface delivered it.""" type: str text: str = "" data: bytes | None = None @dataclasses.dataclass class Delivery: body: bytes attrs: dict[str, Attr] class Unsupported(CheckFailed): """The interface cannot send that.""" def number_value(text: str) -> int | float: try: return int(text) except ValueError: return float(text) def typed_attr(v: Any) -> Attr: """A typed value (an AMQP 0-9-1 header, a Service Bus property) as an Attr. azure-servicebus returns strings as bytes, so bytes that are text carry both forms.""" if isinstance(v, bool): return Attr(TYPE_STRING, "true" if v else "false") if isinstance(v, str): return Attr(TYPE_STRING, v) if isinstance(v, (bytes, bytearray)): data = bytes(v) return Attr(TYPE_BINARY, text_or_none(data) or "", data) if isinstance(v, (int, float)): return Attr(TYPE_NUMBER, str(v)) try: import decimal if isinstance(v, decimal.Decimal): return Attr(TYPE_NUMBER, str(v)) except ImportError: pass return Attr(TYPE_TABLE, json.dumps(v, default=str)) class CrossEnd: """One interface as a sender (creating the queue) and a receiver.""" name = "" attrs = True def send(self, queue: str, msgs: list[Payload]) -> None: raise NotImplementedError def receive(self, queue: str, want: int, deadline: float) -> list[Delivery]: raise NotImplementedError def cleanup(self, queue: str) -> None: raise NotImplementedError class AmqpEnd(CrossEnd): name = NAME_AMQP def __init__(self, env: Env) -> None: self.env = env self.conn = env.amqp() self.ch = self.conn.channel() self.ch.confirm_delivery() def close(self) -> None: with contextlib.suppress(Exception): self.conn.close() def send(self, queue, msgs): import pika self.env.call(self.ch.queue_declare, queue, durable=True) for m in msgs: if m.amqp_value is not None: raise Unsupported("this interface cannot send an AMQP 1.0 value body") headers = {} for a in m.attrs: if a.type == TYPE_STRING: headers[a.name] = a.text elif a.type == TYPE_NUMBER: headers[a.name] = number_value(a.text) elif a.type == TYPE_BINARY: headers[a.name] = a.data elif a.type == TYPE_TABLE: headers[a.name] = a.table props = pika.BasicProperties(delivery_mode=2, headers=headers) self.env.call(self.ch.basic_publish, "", queue, m.body, props, mandatory=True) def receive(self, queue, want, deadline): got = [] while len(got) < want: method, props, body = self.env.call(self.ch.basic_get, queue, auto_ack=False) if method is None: if expired(deadline): raise CheckFailed(f"{len(got)} of {want} arrived in time") time.sleep(POLL_SECS) continue self.ch.basic_ack(method.delivery_tag) got.append(Delivery(body, {k: typed_attr(v) for k, v in (props.headers or {}).items()})) return got def cleanup(self, queue): self.ch.queue_delete(queue) class SqsEnd(CrossEnd): name = NAME_SQS def __init__(self, env: Env) -> None: self.env = env self.sqs = env.sqs() def url(self, queue: str) -> str: return self.env.call(self.sqs.get_queue_url, QueueName=queue)["QueueUrl"] def send(self, queue, msgs): url = self.env.call(self.sqs.create_queue, QueueName=queue)["QueueUrl"] for m in msgs: if m.amqp_value is not None: raise Unsupported("this interface cannot send an AMQP 1.0 value body") attrs = {} for a in m.attrs: if a.type in (TYPE_STRING, TYPE_NUMBER): attrs[a.name] = {"DataType": a.type, "StringValue": a.text} elif a.type == TYPE_BINARY: attrs[a.name] = {"DataType": TYPE_BINARY, "BinaryValue": a.data} else: raise Unsupported(f"this interface cannot send an attribute of type {a.type}") self.env.call(self.sqs.send_message, QueueUrl=url, MessageBody=m.body.decode(), MessageAttributes=attrs) def receive(self, queue, want, deadline): url = self.url(queue) got = [] while len(got) < want: if expired(deadline): raise CheckFailed(f"{len(got)} of {want} arrived in time") out = self.env.call( self.sqs.receive_message, QueueUrl=url, MaxNumberOfMessages=SQS_MAX_MESSAGES, WaitTimeSeconds=SQS_WAIT_SECS, MessageAttributeNames=["All"], ) for m in out.get("Messages", []): attrs = { k: Attr(v.get("DataType", ""), v.get("StringValue", ""), v.get("BinaryValue")) for k, v in m.get("MessageAttributes", {}).items() } got.append(Delivery(m["Body"].encode(), attrs)) self.env.call(self.sqs.delete_message, QueueUrl=url, ReceiptHandle=m["ReceiptHandle"]) return got def cleanup(self, queue): self.sqs.delete_queue(QueueUrl=self.sqs.get_queue_url(QueueName=queue)["QueueUrl"]) class PubSubEnd(CrossEnd): name = NAME_PUBSUB def __init__(self, env: Env) -> None: self.env = env self.publisher, self.subscriber = env.pubsub() self.project = f"projects/{env.tenant}" def topic(self, queue: str) -> str: return f"{self.project}/topics/{queue}{PUBSUB_TOPIC_SUFFIX}" def subscription(self, queue: str) -> str: return f"{self.project}/subscriptions/{queue}" def send(self, queue, msgs): from google.pubsub_v1.types import PubsubMessage topic = self.topic(queue) self.env.call(self.publisher.create_topic, name=topic, retry=None, timeout=SDK_CALL_TIMEOUT_SECS) self.env.call( self.subscriber.create_subscription, name=self.subscription(queue), topic=topic, retry=None, timeout=SDK_CALL_TIMEOUT_SECS, ) for m in msgs: if m.amqp_value is not None: raise Unsupported("this interface cannot send an AMQP 1.0 value body") attrs = {} for a in m.attrs: if a.type != TYPE_STRING: raise Unsupported(f"Pub/Sub attributes are strings, not {a.type}") attrs[a.name] = a.text self.env.call( self.publisher.publish, topic=topic, messages=[PubsubMessage(data=m.body, attributes=attrs)], retry=None, timeout=SDK_CALL_TIMEOUT_SECS, ) def receive(self, queue, want, deadline): got = [] while len(got) < want: if expired(deadline): raise CheckFailed(f"{len(got)} of {want} arrived in time") for m in pubsub_pull(self.env, self.subscriber, self.subscription(queue), want - len(got)): got.append(Delivery(bytes(m.data), {k: Attr(TYPE_STRING, v) for k, v in m.attributes.items()})) return got def cleanup(self, queue): self.subscriber.delete_subscription(subscription=self.subscription(queue), retry=None) self.publisher.delete_topic(topic=self.topic(queue), retry=None) class AzureEnd(CrossEnd): name = NAME_AZURE attrs = False def __init__(self, env: Env) -> None: self.env = env def send(self, queue, msgs): q = self.env.azure_queue(queue) self.env.call(q.create_queue) for m in msgs: if m.amqp_value is not None: raise Unsupported("this interface cannot send an AMQP 1.0 value body") # Azure queue messages have no attributes; only the text goes. self.env.call(q.send_message, m.body.decode()) def receive(self, queue, want, deadline): q = self.env.azure_queue(queue) got = [] while len(got) < want: # The library requires max_messages >= messages_per_page. remaining = min(AZURE_BATCH, want - len(got)) batch = self.env.call(lambda n=remaining: list(q.receive_messages(messages_per_page=n, max_messages=n))) for m in batch: got.append(Delivery((m.content or "").encode(), {})) self.env.call(q.delete_message, m) if not batch: if expired(deadline): raise CheckFailed(f"{len(got)} of {want} arrived in time") time.sleep(POLL_SECS) return got def cleanup(self, queue): self.env.azure_queue(queue).delete_queue() class ServiceBusEnd(CrossEnd): name = NAME_SERVICE_BUS def __init__(self, env: Env) -> None: self.env = env self.client = env.service_bus() self.admin = env.service_bus_admin() def close(self) -> None: with contextlib.suppress(Exception): self.client.close() def send(self, queue, msgs): from azure.servicebus import ServiceBusMessage from azure.servicebus.amqp import AmqpAnnotatedMessage self.env.call(self.admin.create_queue, queue) with self.env.call(self.client.get_queue_sender, queue) as sender: for m in msgs: props: dict[str, Any] = {} for a in m.attrs: if a.type == TYPE_STRING: props[a.name] = a.text elif a.type == TYPE_NUMBER: props[a.name] = number_value(a.text) elif a.type == TYPE_BINARY: props[a.name] = a.data else: raise Unsupported(f"this interface cannot send an application property of type {a.type}") if m.amqp_value is not None: message = AmqpAnnotatedMessage(value_body=m.amqp_value, application_properties=props or None) else: message = ServiceBusMessage(m.body, application_properties=props or None) self.env.call(sender.send_messages, message) def receive(self, queue, want, deadline): got = [] with self.env.call(self.client.get_queue_receiver, queue, max_wait_time=SERVICE_BUS_WAIT_SECS) as receiver: while len(got) < want: if expired(deadline): raise CheckFailed(f"{len(got)} of {want} arrived in time") for m in self.env.call(receiver.receive_messages, max_message_count=SERVICE_BUS_BATCH): attrs = {} for k, v in (m.application_properties or {}).items(): key = text_or_none(k) if key is not None: attrs[key] = typed_attr(v) got.append(Delivery(service_bus_body(m), attrs)) self.env.call(receiver.complete_message, m) return got def cleanup(self, queue): self.admin.delete_queue(queue) # The conversion cases (REQ-6150, DEC-6306): a message read through any # interface arrives in that interface's natural form, marked where the # interface can carry a mark, never silently damaged. @dataclasses.dataclass class AttrCheck: """One attribute a reader must deliver: exact text, exact bytes, text that parses to the same JSON, or text whatever its type (a Number crossing to a protocol with typed values).""" name: str type: str = "" text: str | None = None data: bytes | None = None json: str | None = None any_text: str | None = None @dataclasses.dataclass class Expectation: """What one reader must deliver for a case's message, and the rule.""" why: str body: bytes | None = None body_json: str | None = None attrs: list[AttrCheck] = dataclasses.field(default_factory=list) def check(self, d: Delivery) -> None: if self.body_json is not None: if not json_equal(d.body, self.body_json.encode()): raise CheckFailed( f"conversion differs from REQ-6150: body {clip(d.body)!r}, want JSON {self.body_json}" ) elif d.body != self.body: raise CheckFailed( f"conversion differs from REQ-6150: body {clip(d.body)!r}, want {clip(self.body or b'')!r}" ) for c in self.attrs: got = d.attrs.get(c.name) if got is None: raise CheckFailed(f"conversion differs from REQ-6150: attribute {c.name} missing") if c.type and got.type != c.type: raise CheckFailed( f"conversion differs from REQ-6150: attribute {c.name} has type {got.type}, want {c.type}" ) if c.data is not None: if got.data != c.data: raise CheckFailed( f"conversion differs from REQ-6150: attribute {c.name} = {got.data!r}, want bytes {c.data!r}" ) elif c.json is not None: if not json_equal(got.text.encode(), c.json.encode()): raise CheckFailed( f"conversion differs from REQ-6150: attribute {c.name} = {got.text!r}, want JSON {c.json}" ) elif c.any_text is not None: if got.text != c.any_text: raise CheckFailed( f"conversion differs from REQ-6150: attribute {c.name} = {got.text!r}, want {c.any_text}" ) elif got.text != c.text: raise CheckFailed( f"conversion differs from REQ-6150: attribute {c.name} = {got.text!r}, want {c.text!r}" ) def json_equal(a: bytes, b: bytes) -> bool: try: return json.loads(a) == json.loads(b) except (ValueError, UnicodeDecodeError): return False def clip(b: bytes) -> bytes: return b[:CLIP_BYTES] def b64(b: bytes) -> bytes: return base64.b64encode(b) @dataclasses.dataclass class ConversionCase: name: str about: str senders: list[str] payload: Payload expect: Callable[[str], Expectation] def random_binary(size: int) -> bytes: """Random bytes that start with bytes UTF-8 forbids, so the body is not text whatever follows.""" return b"\xff\xfe" + secrets.token_bytes(size - 2) def conversion_cases(binary: bytes | None = None) -> list[ConversionCase]: """The cases, in the order REQ-6150 states its rules.""" binary = binary if binary is not None else random_binary(BINARY_BODY_SIZE) small_binary = b"\xff\x00\x01" table = {"k": "v", "n": 1} table_json = '{"k":"v","n":1}' def binary_body(reader: str) -> Expectation: if reader == NAME_SQS: return Expectation( "SQS bodies are text: base64, marked x-pq-body-encoding = base64 (rule 2)", body=b64(binary), attrs=[AttrCheck(MARKER_BODY_ENCODING, text=ENCODING_BASE64)], ) if reader == NAME_AZURE: return Expectation( "Azure Queue Storage bodies are text and carry no attributes: base64, unmarked (rule 3)", body=b64(binary), ) return Expectation("this interface carries bytes: the exact bytes (rule 1)", body=binary) def sqs_marked(reader: str) -> Expectation: if reader == NAME_AZURE: return Expectation("Azure Queue Storage text: base64 again, unmarked (rule 3)", body=b64(binary)) return Expectation("stored as the decoded bytes, delivered exactly (rules 1 and 2)", body=binary) def typed(reader: str) -> Expectation: body = b"typed attributes" if reader == NAME_PUBSUB: return Expectation( "Pub/Sub attributes are strings: the number as its text, the binary value base64, " "listed in x-pq-base64-attributes (rule 6)", body=body, attrs=[ AttrCheck("s", text="text"), AttrCheck("n", text="42"), AttrCheck("b", text=b64(small_binary).decode()), AttrCheck(MARKER_BASE64_ATTRIBUTES, text="b"), ], ) if reader == NAME_AZURE: return Expectation("Azure Queue Storage shows no attributes; the text crosses (rule 8)", body=body) return Expectation( "typed values stay typed: the text, the number, the bytes (rule 5)", body=body, attrs=[AttrCheck("s", text="text"), AttrCheck("n", any_text="42"), AttrCheck("b", data=small_binary)], ) def table_header(reader: str) -> Expectation: body = b"a table header" if reader == NAME_SQS: return Expectation( "as JSON text, SQS data type String.json (rule 7)", body=body, attrs=[AttrCheck("t", type=SQS_JSON_TYPE, json=table_json)], ) if reader == NAME_AZURE: return Expectation("Azure Queue Storage shows no attributes (rule 8)", body=body) return Expectation("as JSON text (rule 7)", body=body, attrs=[AttrCheck("t", json=table_json)]) return [ ConversionCase( "a binary body", "Sent through the interfaces that carry bytes; SQS and Azure Queue Storage carry only text.", [NAME_AMQP, NAME_PUBSUB, NAME_SERVICE_BUS], Payload(body=binary), binary_body, ), ConversionCase( "binary sent through SQS with the marker", "SQS sends binary as base64 text marked x-pq-body-encoding = base64; it is stored as the bytes.", [NAME_SQS], Payload(body=b64(binary), attrs=[PayloadAttr(MARKER_BODY_ENCODING, TYPE_STRING, text=ENCODING_BASE64)]), sqs_marked, ), ConversionCase( "typed attributes", "SQS sends a String, a Number and a Binary attribute.", [NAME_SQS], Payload( body=b"typed attributes", attrs=[ PayloadAttr("s", TYPE_STRING, text="text"), PayloadAttr("n", TYPE_NUMBER, text="42"), PayloadAttr("b", TYPE_BINARY, data=small_binary), ], ), typed, ), ConversionCase( "an AMQP table header", "AMQP 0-9-1 sends a header whose value is a field table, which no other protocol has.", [NAME_AMQP], Payload(body=b"a table header", attrs=[PayloadAttr("t", TYPE_TABLE, table=table)]), table_header, ), service_bus_value_case("a Service Bus integer value body", 42, "42"), service_bus_value_case("a Service Bus map value body", {"a": "b", "n": 1}, '{"a":"b","n":1}'), ] def service_bus_value_case(name: str, value: Any, want: str) -> ConversionCase: """A Service Bus AMQP value body, which every other interface must deliver as the JSON text want (rule 4).""" def expect(reader: str) -> Expectation: if reader == NAME_AZURE: return Expectation("as JSON text, unmarked (rules 3 and 4)", body_json=want) return Expectation( "as JSON text, marked x-pq-body-encoding = json (rule 4)", body_json=want, attrs=[AttrCheck(MARKER_BODY_ENCODING, text=ENCODING_JSON)], ) return ConversionCase( name, "Service Bus sends an AMQP 1.0 value body, which only AMQP 1.0 can carry as such.", [NAME_SERVICE_BUS], Payload(amqp_value=value), expect, ) @dataclasses.dataclass class CrossResult: check: str sender: str receiver: str error: str | None def text_payloads(msgs: list[Message]) -> list[Payload]: return [Payload(body=m.body, attrs=[PayloadAttr(k, TYPE_STRING, text=v) for k, v in m.attrs.items()]) for m in msgs] def deliveries_received(ds: list[Delivery]) -> list[Received]: return [Received(d.body, {k: a.text for k, a in d.attrs.items()}) for d in ds] def run_cross(env: Env, run_id: str, timeout: float, say: Narrator) -> list[CrossResult]: say.step("opening one client per protocol, shared by every pair") amqp_end = AmqpEnd(env) sb_end = ServiceBusEnd(env) try: ends: list[CrossEnd] = [amqp_end, SqsEnd(env), PubSubEnd(env), AzureEnd(env), sb_end] results = [] for sender in ends: for receiver in ends: if sender is receiver: continue queue = f"{entity_name(run_id)}-{CROSS_CODES[sender.name]}-{CROSS_CODES[receiver.name]}" results.append( CrossResult( "text", sender.name, receiver.name, text_pair(run_id, timeout, say, sender, receiver, queue) ) ) by_name = {e.name: e for e in ends} for ci, case in enumerate(conversion_cases()): say.step(f"conversion: {case.name}. {case.about}") for sender_name in case.senders: for receiver in ends: if receiver.name == sender_name: continue sender = by_name[sender_name] code = f"{CROSS_CODES[sender_name]}-{CROSS_CODES[receiver.name]}" queue = f"{entity_name(run_id)}-c{ci}-{code}" results.append( CrossResult( case.name, sender_name, receiver.name, conversion_pair(timeout, say, case, sender, receiver, queue), ) ) return results finally: amqp_end.close() sb_end.close() def delete_through(sender: CrossEnd, queue: str) -> str | None: try: sender.cleanup(queue) except Exception as e: # noqa: BLE001 - a failed deletion is reported with the pair return f"delete through {sender.name}: {e}" return None def text_pair( run_id: str, timeout: float, say: Narrator, sender: CrossEnd, receiver: CrossEnd, queue: str ) -> str | None: deadline = deadline_after(timeout) sent = new_messages(f"{run_id}-{queue}", text=True) with_attrs = sender.attrs and receiver.attrs what = "bytes and attributes" if with_attrs else "bytes" say.step( f"{sender.name} -> {receiver.name}: queue {queue}, sending {len(sent)} text messages through {sender.name}" ) error = None try: sender.send(queue, text_payloads(sent)) got = receiver.receive(queue, len(sent), deadline) verify_cross(sent, deliveries_received(got), with_attrs) say.step(f"{sender.name} -> {receiver.name}: received all {len(sent)} through {receiver.name}; {what} match") except Refused: raise except Exception as e: # noqa: BLE001 - a pair's failure, whatever raised it, is reported; the next pair runs error = describe(e) say.step(f"{sender.name} -> {receiver.name}: MISMATCH: {error}") finally: cleanup_error = delete_through(sender, queue) return error or cleanup_error def conversion_pair( timeout: float, say: Narrator, case: ConversionCase, sender: CrossEnd, receiver: CrossEnd, queue: str ) -> str | None: deadline = deadline_after(timeout) want = case.expect(receiver.name) say.step(f" {sender.name} -> {receiver.name}: expecting {want.why}") error = None try: sender.send(queue, [case.payload]) got = receiver.receive(queue, 1, deadline) want.check(got[0]) say.step(f" {sender.name} -> {receiver.name}: as expected") except Refused: raise except Exception as e: # noqa: BLE001 - a pair's failure, whatever raised it, is reported; the next pair runs error = describe(e) say.step(f" {sender.name} -> {receiver.name}: MISMATCH: {error}") finally: cleanup_error = delete_through(sender, queue) return error or cleanup_error # -------------------------------------------------------------------------- # The run # -------------------------------------------------------------------------- INTRO = """powerqueue-link checks the free message queue at {host} through each of its six interfaces, using each protocol's official Python client library, exactly as an application would. For every interface it: 1. connects, naming tenant {tenant} in the protocol's credential field; 2. creates a queue (or topic and subscription) named for this run; 3. sends {count} messages of random content, sized {sizes}; 4. receives them, and verifies that each came back exactly once with the same bytes and the same attributes or properties (and, on AMQP 0-9-1, in the order sent); 5. checks that nothing is left or redelivered once they are settled; 6. deletes what it created. Then, because every interface shares one set of queue names within a tenant, it sends through each interface and receives through every other one (20 pairs), checking the bytes, and the attributes wherever both interfaces carry them; and it checks each conversion between interfaces: binary bodies, typed attributes, an AMQP table header and Service Bus value bodies, each against the rule it should follow. What it does and does not do: - it connects only to {allowed}; every connection is checked against that list (the AMQP 0-9-1 library also connects to a loopback socket of its own, to wake itself), except the Pub/Sub one: Google's gRPC library connects from its C core, which Python cannot intercept, and its one destination is fixed by the endpoint the program passes, pubsub.{host}:{https}; - it sends only random test data, reads no files of yours (only its tenant file) and no cloud configuration, sends no telemetry and never updates itself; - it paces its requests to {rate:g} a second and stops at the first refusal for a limit rather than retrying; - unlike the Go version, it does not sandbox itself on Linux; - the tenant identifier is saved in {where}. """ INTERFACES = [ ( "AMQP 0-9-1 (amqps, port 5673)", lambda env: f"amqps://amqp.{env.host}:{AMQPS_PORT} with pika; the virtual host is the tenant", check_amqp, ), ( "Amazon SQS (https)", lambda env: f"https://sqs.{env.host} with boto3; the access key id is the tenant", check_sqs, ), ( "Google Cloud Pub/Sub (gRPC)", lambda env: f"pubsub.{env.host}:{HTTPS_PORT} with google-cloud-pubsub; the project id is the tenant", check_pubsub, ), ( "Azure Queue Storage (https)", lambda env: f"https://azure.{env.host}/{env.tenant} with azure-storage-queue; the account is the tenant", check_azure_queue, ), ( "Azure Service Bus administration (https)", lambda env: ( f"https://amqp.{env.host} with azure-servicebus (administration client); the key name is the tenant" ), check_service_bus_admin, ), ( "Azure Service Bus (AMQP 1.0, port 5671)", lambda env: f"sb://amqp.{env.host} (port {SERVICE_BUS_PORT}) with azure-servicebus; the key name is the tenant", check_service_bus, ), ] def print_table(rows: list[list[str]]) -> None: widths = [max(len(r[i]) for r in rows) for i in range(len(rows[0]))] for r in rows: print(" ".join(c.ljust(widths[i]) for i, c in enumerate(r)).rstrip()) def print_cross(results: list[CrossResult]) -> int: failed = sum(1 for r in results if r.error) text = [r for r in results if r.check == "text"] conversions = [r for r in results if r.check != "text"] if text: names = list(dict.fromkeys(r.sender for r in text)) cell = {(r.sender, r.receiver): "FAIL" if r.error else "ok" for r in text} print("\nCross-protocol text (rows send, columns receive)\n") rows = [["sent through \\ received through", *names]] for s in names: rows.append([s, *["-" if s == r else cell.get((s, r), "") for r in names]]) print_table(rows) if conversions: print("\nCross-protocol conversions\n") print_table([["FAIL" if r.error else "ok", r.check, f"{r.sender} -> {r.receiver}"] for r in conversions]) for r in results: if r.error: print(f" FAIL {r.check}, {r.sender} -> {r.receiver}: {r.error}") return failed def parse_args(argv: list[str]) -> argparse.Namespace: p = argparse.ArgumentParser( prog="powerqueue_link.py", description="Check the free message queue at powerqueue.link through every interface it offers.", ) p.add_argument("--host", default=DEFAULT_HOST, help="the service's domain (default %(default)s)") p.add_argument("--tenant", help="tenant identifier to use (not saved; default: the saved one)") p.add_argument("--new-tenant", action="store_true", help="make a new tenant identifier and save it") p.add_argument("--quiet", action="store_true", help="print only the summary, not each step") p.add_argument( "--skip-cross", action="store_true", help="skip the cross-protocol checks (send through one interface, receive through each other)", ) p.add_argument( "--timeout", type=float, default=DEFAULT_TIMEOUT_SECS, help="seconds allowed for each interface (default %(default)g)", ) p.add_argument("--version", action="store_true", help="print the version and exit") return p.parse_args(argv) def run(argv: list[str]) -> int: args = parse_args(argv) if args.version: print( f"powerqueue-link {VERSION} (Python {platform.python_version()}, {platform.system()}/{platform.machine()})" ) return EXIT_OK if args.timeout <= 0: print("--timeout must be positive", file=sys.stderr) return EXIT_USAGE try: if args.tenant and args.new_tenant: raise SetupError("give --tenant or --new-tenant, not both") if args.tenant: if not valid_tenant(args.tenant): raise SetupError( f"{args.tenant!r}: a tenant identifier is a lowercase letter followed by " "23 lowercase letters or digits" ) tenant, where = args.tenant, "nowhere (given with --tenant)" else: path = tenant_file() tenant, created = load_tenant(path, args.new_tenant) where = f"{path}{' (new)' if created else ''}" except SetupError as e: print(e, file=sys.stderr) return EXIT_USAGE # Before any client library loads or connects. sanitise_environment() guard = Guard(args.host) guard.install() env = Env(args.host, tenant, guard) run_id = random_identifier(RUN_ID_LENGTH) if not args.quiet: print( INTRO.format( host=args.host, tenant=tenant, count=len(MESSAGE_SIZES), sizes=", ".join(f"{n} B" for n in MESSAGE_SIZES), allowed=", ".join(guard.allowed_list()), https=HTTPS_PORT, rate=REQUESTS_PER_SECOND, where=where, ) ) failed = 0 refused: str | None = None rows = [] try: for i, (name, about, check) in enumerate(INTERFACES, 1): say = Narrator(args.quiet) if not args.quiet: print(f"[{i}/{len(INTERFACES)}] {name}\n {about(env)}") cleanups = Cleanups(say) started = time.monotonic() error = None try: check(env, run_id, say, args.timeout, cleanups) except Refused: raise except Exception as e: # noqa: BLE001 - a check's failure, whatever raised it, is reported; the next runs error = describe(e) finally: cleanup_errors = cleanups.run() if error is None and cleanup_errors: error = "; ".join(cleanup_errors) took = f"{time.monotonic() - started:.3f}s" if error: failed += 1 say.step(f"FAILED: {error}") rows.append(["FAIL", name, took, error]) else: say.step("passed") rows.append(["ok", name, took, ""]) if not args.quiet: print() print("Summary\n") print_table(rows) if not args.skip_cross: if not args.quiet: print("\nCross-protocol: send through one interface, receive through another") try: failed += print_cross(run_cross(env, run_id, args.timeout, Narrator(args.quiet))) except Refused: raise except Exception as e: # noqa: BLE001 - reported as a failure in the summary print(f"\ncross-protocol checks stopped: {describe(e)}") failed += 1 except Refused as e: refused = str(e) except KeyboardInterrupt: print("\nInterrupted; what this run created was deleted.") return EXIT_INTERRUPTED finally: guard.uninstall() if refused is not None: print( f"\nStopped: the service refused a request for exceeding a limit or being busy: {refused}.\n" "The service limits each address's requests and bans addresses that keep\n" "exceeding them; this program stops at the first refusal instead of retrying.\n" "Wait a few minutes before running it again." ) return EXIT_REFUSED if failed: print(f"\n{failed} checks failed") return EXIT_FAILED print("\nall checks verified") return EXIT_OK def main() -> None: try: sys.exit(run(sys.argv[1:])) except KeyboardInterrupt: print("\nInterrupted.") sys.exit(EXIT_INTERRUPTED) if __name__ == "__main__": main()