"""Multi-instance topology tests (REQ-049/050 / BAT-013). Builder/routing-key tests run always. declare_topology + directed/broadcast routing tests are gated on TREX_RABBITMQ_TEST_URL. """ from __future__ import annotations import os import time import uuid import pytest from trex_core.adapters.messaging import topology from trex_core.adapters.messaging.rabbitmq import RabbitMQTransport # ---- builders (always run) -------------------------------------------------- def test_named_instance_suffix(): """REQ-049 AC-1.""" assert topology.intent_queue("inst-1") == "trex.intent.req.inst-1" assert topology.snapshot_queue("inst-1") == "trex.snapshot.req.inst-1" assert topology.control_queue("inst-1") == "trex.control.req.inst-1" def test_default_is_literal(): """REQ-049 AC-2: default keeps legacy literal names (backward-compat).""" assert topology.intent_queue("default") == "trex.intent.req" assert topology.snapshot_queue(None) == "trex.snapshot.req" assert topology.control_queue() == "trex.control.req" def test_direct_routing_key(): """REQ-050 AC-3.""" assert topology.direct_routing_key("inst-1", "control") == "inst-1.control" assert topology.direct_routing_key("inst-2", "intent") == "inst-2.intent" # ---- RabbitMQ topology (gated) ---------------------------------------------- @pytest.fixture def mq_url() -> str: url = os.environ.get("TREX_RABBITMQ_TEST_URL") if not url: pytest.skip("TREX_RABBITMQ_TEST_URL not set; skipping RabbitMQ integration test") return url def _delete_instance_queues(url, instance_key): import pika conn = pika.BlockingConnection(pika.URLParameters(url)) try: ch = conn.channel() for q in (topology.intent_queue(instance_key), topology.snapshot_queue(instance_key), topology.control_queue(instance_key)): try: ch.queue_delete(queue=q) except Exception: # noqa: BLE001 pass finally: conn.close() def test_directed_routing_isolation(mq_url): """REQ-050 AC-4 (S-2): inst-1 routing key reaches only inst-1's queue.""" import pika k1 = "it1-" + uuid.uuid4().hex[:6] k2 = "it2-" + uuid.uuid4().hex[:6] t1 = RabbitMQTransport(mq_url) t1.declare_topology(k1) t2 = RabbitMQTransport(mq_url) t2.declare_topology(k2) try: conn = pika.BlockingConnection(pika.URLParameters(mq_url)) ch = conn.channel() ch.basic_publish(exchange=topology.DIRECT_EXCHANGE, routing_key=topology.direct_routing_key(k1, "intent"), body=b"x") time.sleep(0.4) m1 = ch.basic_get(topology.intent_queue(k1), auto_ack=True) m2 = ch.basic_get(topology.intent_queue(k2), auto_ack=True) conn.close() assert m1[0] is not None, "inst-1 intent queue should receive the directed message" assert m2[0] is None, "inst-2 intent queue must NOT receive it" finally: t1.close() t2.close() _delete_instance_queues(mq_url, k1) _delete_instance_queues(mq_url, k2) def test_broadcast_dual_binding(mq_url): """REQ-050 AC-5 (S-3): broadcast reaches every instance's control queue.""" import pika k1 = "bc1-" + uuid.uuid4().hex[:6] k2 = "bc2-" + uuid.uuid4().hex[:6] t1 = RabbitMQTransport(mq_url) t1.declare_topology(k1) t2 = RabbitMQTransport(mq_url) t2.declare_topology(k2) try: conn = pika.BlockingConnection(pika.URLParameters(mq_url)) ch = conn.channel() ch.basic_publish(exchange=topology.BROADCAST_EXCHANGE, routing_key="", body=b"halt") time.sleep(0.4) m1 = ch.basic_get(topology.control_queue(k1), auto_ack=True) m2 = ch.basic_get(topology.control_queue(k2), auto_ack=True) conn.close() assert m1[0] is not None and m2[0] is not None, "both control queues should receive broadcast" finally: t1.close() t2.close() _delete_instance_queues(mq_url, k1) _delete_instance_queues(mq_url, k2)