34 lines
1.3 KiB
Python
34 lines
1.3 KiB
Python
"""Topology constants + routing key builders (REQ-021)."""
|
|
from __future__ import annotations
|
|
|
|
from trex_core.adapters.messaging import topology
|
|
|
|
|
|
def test_topology_constants():
|
|
assert topology.SNAPSHOT_REQ_QUEUE == "trex.snapshot.req"
|
|
assert topology.INTENT_REQ_QUEUE == "trex.intent.req"
|
|
assert topology.EVENTS_EXCHANGE == "trex.events"
|
|
assert topology.EVENTS_DLQ_EXCHANGE == "trex.events.dlq"
|
|
|
|
|
|
def test_event_routing_key_format():
|
|
assert topology.event_routing_key("slot_filled_buy", "A", "005930") == \
|
|
"slot_filled_buy.A.005930"
|
|
# dots/wildcards in inputs are sanitized to underscores
|
|
assert topology.event_routing_key("e.x", "y*z", "a#b") == "e_x.y_z.a_b"
|
|
# None / empty strategy → "_"
|
|
assert topology.event_routing_key("e", None, "S") == "e._.S"
|
|
|
|
|
|
def test_strategy_queue_name_and_binding_pattern():
|
|
assert topology.strategy_event_queue_name("A") == "trex.events.strategy.A"
|
|
assert topology.strategy_event_binding_pattern("A") == "*.A.#"
|
|
|
|
|
|
def test_strategy_queue_args_have_ttl_and_max_length():
|
|
args = topology.STRATEGY_QUEUE_ARGS
|
|
assert args["x-message-ttl"] > 0
|
|
assert args["x-max-length"] > 0
|
|
assert args["x-overflow"] == "drop-head"
|
|
assert args["x-dead-letter-exchange"] == topology.EVENTS_DLQ_EXCHANGE
|