This is an automated email from the ASF dual-hosted git repository. kpvdr pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/qpid-interop-test.git
commit 6496068acb1c6c6da19b5d55cb1738c00a92b952 Author: QIT Development Team <[email protected]> AuthorDate: Fri Aug 7 11:34:03 2026 -0400 Phase 2d: Add AMQP message header section interop tests (275 tests) Add --message-header CLI arg to all 5 AMQP shims and test that durable, priority, ttl, first-acquirer, and delivery-count round-trip correctly across the full 5×5 client matrix through Artemis. Co-Authored-By: Claude Opus 4.6 <[email protected]> --- shims/cpp-proton/include/qit_shim.hpp | 5 +- shims/cpp-proton/src/main.cpp | 5 +- shims/cpp-proton/src/receiver.cpp | 9 + shims/cpp-proton/src/sender.cpp | 29 +- shims/dotnet-proton/src/Program.cs | 5 +- shims/dotnet-proton/src/Receiver.cs | 10 + shims/dotnet-proton/src/Sender.cs | 19 +- .../main/java/org/apache/qpid/qit/Receiver.java | 9 + .../src/main/java/org/apache/qpid/qit/Sender.java | 17 ++ shims/javascript-rhea/shim.js | 18 +- shims/python-proton/shim.py | 34 ++- tests/test_amqp_headers.py | 291 +++++++++++++++++++++ 12 files changed, 444 insertions(+), 7 deletions(-) diff --git a/shims/cpp-proton/include/qit_shim.hpp b/shims/cpp-proton/include/qit_shim.hpp index 01e8131..bd12625 100644 --- a/shims/cpp-proton/include/qit_shim.hpp +++ b/shims/cpp-proton/include/qit_shim.hpp @@ -33,7 +33,8 @@ public: const std::string& test_data_json, bool jms_mode = false, const std::string& headers_json = "", - const std::string& properties_json = ""); + const std::string& properties_json = "", + const std::string& message_header_json = ""); void on_container_start(proton::container& c) override; void on_sendable(proton::sender& s) override; @@ -51,10 +52,12 @@ private: bool jms_mode_; Json::Value headers_; Json::Value properties_; + Json::Value message_header_; int8_t get_jms_message_type(const std::string& amqp_type) const; void apply_headers(proton::message& msg); void apply_properties(proton::message& msg); + void apply_message_header(proton::message& msg); }; // Receiver handler - receives messages diff --git a/shims/cpp-proton/src/main.cpp b/shims/cpp-proton/src/main.cpp index 93ea161..8b4af79 100644 --- a/shims/cpp-proton/src/main.cpp +++ b/shims/cpp-proton/src/main.cpp @@ -43,6 +43,7 @@ struct CommandLineArgs { std::string data; std::string headers; std::string properties; + std::string message_header; int count = 0; int timeout = 30; bool jms_mode = false; @@ -93,6 +94,8 @@ struct CommandLineArgs { headers = val; } else if (opt == "--properties") { properties = val; + } else if (opt == "--message-header") { + message_header = val; } else if (opt == "--large-content") { large_content = val; } else if (opt == "--size") { @@ -193,7 +196,7 @@ int main(int argc, char** argv) { return 0; } } else if (args.command == "send") { - qit::Sender sender(args.broker, args.queue, args.amqp_type, args.data, args.jms_mode, args.headers, args.properties); + qit::Sender sender(args.broker, args.queue, args.amqp_type, args.data, args.jms_mode, args.headers, args.properties, args.message_header); proton::container(sender).run(); return 0; } else if (args.command == "receive") { diff --git a/shims/cpp-proton/src/receiver.cpp b/shims/cpp-proton/src/receiver.cpp index dfd48fc..39edcfa 100644 --- a/shims/cpp-proton/src/receiver.cpp +++ b/shims/cpp-proton/src/receiver.cpp @@ -185,6 +185,15 @@ void Receiver::on_message(proton::delivery& d, proton::message& m) { msg_data["properties"] = props; } + // Extract AMQP Header section fields + Json::Value msg_header(Json::objectValue); + msg_header["durable"] = m.durable(); + msg_header["priority"] = static_cast<int>(m.priority()); + msg_header["ttl"] = static_cast<Json::UInt64>(m.ttl().milliseconds()); + msg_header["first_acquirer"] = m.first_acquirer(); + msg_header["delivery_count"] = m.delivery_count(); + msg_data["message_header"] = msg_header; + received_messages_.append(msg_data); received_count_++; diff --git a/shims/cpp-proton/src/sender.cpp b/shims/cpp-proton/src/sender.cpp index 93a40c4..59c8ba9 100644 --- a/shims/cpp-proton/src/sender.cpp +++ b/shims/cpp-proton/src/sender.cpp @@ -4,6 +4,7 @@ #include "qit_shim.hpp" #include <json/json.h> +#include <proton/duration.hpp> #include <proton/message_id.hpp> #include <proton/transport.hpp> #include <proton/annotation_key.hpp> @@ -27,7 +28,8 @@ Sender::Sender(const std::string& broker_url, const std::string& test_data_json, bool jms_mode, const std::string& headers_json, - const std::string& properties_json) + const std::string& properties_json, + const std::string& message_header_json) : broker_url_(broker_url), queue_name_(queue_name), amqp_type_(amqp_type), @@ -70,6 +72,16 @@ Sender::Sender(const std::string& broker_url, throw std::runtime_error("Failed to parse properties JSON: " + perrors); } } + + // Parse message header JSON if provided + if (!message_header_json.empty()) { + Json::CharReaderBuilder mbuilder; + std::istringstream miss(message_header_json); + std::string merrors; + if (!Json::parseFromStream(mbuilder, miss, &message_header_, &merrors)) { + throw std::runtime_error("Failed to parse message header JSON: " + merrors); + } + } } int8_t Sender::get_jms_message_type(const std::string& amqp_type) const { @@ -155,6 +167,10 @@ void Sender::on_sendable(proton::sender& s) { apply_properties(msg); } + if (!message_header_.isNull()) { + apply_message_header(msg); + } + s.send(msg); sent_count_++; } @@ -227,6 +243,17 @@ void Sender::apply_properties(proton::message& msg) { } } +void Sender::apply_message_header(proton::message& msg) { + if (message_header_.isMember("durable")) + msg.durable(message_header_["durable"].asBool()); + if (message_header_.isMember("priority")) + msg.priority(static_cast<uint8_t>(message_header_["priority"].asInt())); + if (message_header_.isMember("ttl")) + msg.ttl(proton::duration(message_header_["ttl"].asUInt())); + if (message_header_.isMember("first_acquirer")) + msg.first_acquirer(message_header_["first_acquirer"].asBool()); +} + void Sender::on_tracker_accept(proton::tracker& t) { confirmed_count_++; diff --git a/shims/dotnet-proton/src/Program.cs b/shims/dotnet-proton/src/Program.cs index 57dc07a..dbc032f 100644 --- a/shims/dotnet-proton/src/Program.cs +++ b/shims/dotnet-proton/src/Program.cs @@ -26,6 +26,7 @@ namespace Qit.Shim var sendJmsModeOption = new Option<bool>("--jms-mode", () => false, "Enable JMS emulation mode"); var sendHeadersOption = new Option<string>("--headers", () => null, "JSON JMS headers"); var sendPropertiesOption = new Option<string>("--properties", () => null, "JSON JMS application properties"); + var sendMessageHeaderOption = new Option<string>("--message-header", () => null, "JSON AMQP Header section fields"); var sendLargeContentOption = new Option<string>("--large-content", () => null, "Large content type (binary or string)"); var sendSizeOption = new Option<int>("--size", () => 0, "Large content size"); var sendSeedOption = new Option<int>("--seed", () => 0, "PRNG seed"); @@ -40,6 +41,7 @@ namespace Qit.Shim sendCommand.AddOption(sendJmsModeOption); sendCommand.AddOption(sendHeadersOption); sendCommand.AddOption(sendPropertiesOption); + sendCommand.AddOption(sendMessageHeaderOption); sendCommand.AddOption(sendLargeContentOption); sendCommand.AddOption(sendSizeOption); sendCommand.AddOption(sendSeedOption); @@ -69,7 +71,8 @@ namespace Qit.Shim var data = context.ParseResult.GetValueForOption(sendDataOption); var headers = context.ParseResult.GetValueForOption(sendHeadersOption); var properties = context.ParseResult.GetValueForOption(sendPropertiesOption); - Sender.Send(broker, queue, type, data, jmsMode, headers, properties); + var messageHeaderStr = context.ParseResult.GetValueForOption(sendMessageHeaderOption); + Sender.Send(broker, queue, type, data, jmsMode, headers, properties, messageHeaderStr); } } catch (Exception ex) diff --git a/shims/dotnet-proton/src/Receiver.cs b/shims/dotnet-proton/src/Receiver.cs index aa77438..e6d34f6 100644 --- a/shims/dotnet-proton/src/Receiver.cs +++ b/shims/dotnet-proton/src/Receiver.cs @@ -238,6 +238,16 @@ namespace Qit.Shim msgResult.Properties = props; } + // Extract AMQP Header section fields + msgResult.MessageHeader = new Dictionary<string, object> + { + { "durable", message.Durable }, + { "priority", (int)message.Priority }, + { "ttl", (long)message.TimeToLive }, + { "first_acquirer", message.FirstAcquirer }, + { "delivery_count", (int)message.DeliveryCount }, + }; + messages.Add(msgResult); } diff --git a/shims/dotnet-proton/src/Sender.cs b/shims/dotnet-proton/src/Sender.cs index 2bccbc9..74d0eb1 100644 --- a/shims/dotnet-proton/src/Sender.cs +++ b/shims/dotnet-proton/src/Sender.cs @@ -56,7 +56,7 @@ namespace Qit.Shim public static class Sender { - public static void Send(string broker, string queue, string type, string data, bool jmsMode = false, string headersJson = null, string propertiesJson = null) + public static void Send(string broker, string queue, string type, string data, bool jmsMode = false, string headersJson = null, string propertiesJson = null, string messageHeaderJson = null) { try { @@ -193,6 +193,20 @@ namespace Qit.Shim } } + // Apply AMQP Header section fields + if (!string.IsNullOrEmpty(messageHeaderJson)) + { + var mh = JsonConvert.DeserializeObject<Dictionary<string, object>>(messageHeaderJson); + if (mh.ContainsKey("durable")) + message.Durable = Convert.ToBoolean(mh["durable"]); + if (mh.ContainsKey("priority")) + message.Priority = (byte)Convert.ToInt32(mh["priority"]); + if (mh.ContainsKey("ttl")) + message.TimeToLive = Convert.ToUInt32(mh["ttl"]); + if (mh.ContainsKey("first_acquirer")) + message.FirstAcquirer = Convert.ToBoolean(mh["first_acquirer"]); + } + sender.Send(message); messages.Add(new MessageResult @@ -359,5 +373,8 @@ namespace Qit.Shim [JsonProperty("properties", NullValueHandling = NullValueHandling.Ignore)] public Dictionary<string, object> Properties { get; set; } + + [JsonProperty("message_header", NullValueHandling = NullValueHandling.Ignore)] + public Dictionary<string, object> MessageHeader { get; set; } } } diff --git a/shims/java-protonj2/src/main/java/org/apache/qpid/qit/Receiver.java b/shims/java-protonj2/src/main/java/org/apache/qpid/qit/Receiver.java index 4a10133..841fb2d 100644 --- a/shims/java-protonj2/src/main/java/org/apache/qpid/qit/Receiver.java +++ b/shims/java-protonj2/src/main/java/org/apache/qpid/qit/Receiver.java @@ -399,6 +399,15 @@ public class Receiver { msgResult.add("properties", propsJson); } + // Extract AMQP Header section fields + JsonObject msgHeader = new JsonObject(); + msgHeader.addProperty("durable", message.durable()); + msgHeader.addProperty("priority", (int) message.priority()); + msgHeader.addProperty("ttl", message.timeToLive()); + msgHeader.addProperty("first_acquirer", message.firstAcquirer()); + msgHeader.addProperty("delivery_count", (int) message.deliveryCount()); + msgResult.add("message_header", msgHeader); + messages.add(msgResult); } diff --git a/shims/java-protonj2/src/main/java/org/apache/qpid/qit/Sender.java b/shims/java-protonj2/src/main/java/org/apache/qpid/qit/Sender.java index ca3e137..dd413ee 100644 --- a/shims/java-protonj2/src/main/java/org/apache/qpid/qit/Sender.java +++ b/shims/java-protonj2/src/main/java/org/apache/qpid/qit/Sender.java @@ -28,6 +28,7 @@ public class Sender { String data = null; String headersJson = null; String propertiesJson = null; + String messageHeaderJson = null; boolean jmsMode = false; String largeContent = null; int size = 0; @@ -72,6 +73,9 @@ public class Sender { case "properties": propertiesJson = value; break; + case "message-header": + messageHeaderJson = value; + break; case "large-content": largeContent = value; break; @@ -313,6 +317,19 @@ public class Sender { } } + // Apply AMQP Header section fields + if (messageHeaderJson != null) { + JsonObject mh = new Gson().fromJson(messageHeaderJson, JsonObject.class); + if (mh.has("durable")) + message.durable(mh.get("durable").getAsBoolean()); + if (mh.has("priority")) + message.priority((byte) mh.get("priority").getAsInt()); + if (mh.has("ttl")) + message.timeToLive(mh.get("ttl").getAsLong()); + if (mh.has("first_acquirer")) + message.firstAcquirer(mh.get("first_acquirer").getAsBoolean()); + } + sender.send(message); // Record sent message diff --git a/shims/javascript-rhea/shim.js b/shims/javascript-rhea/shim.js index 272fbc9..eba813b 100755 --- a/shims/javascript-rhea/shim.js +++ b/shims/javascript-rhea/shim.js @@ -976,9 +976,10 @@ function receiveLargeContent(options) { // Sender function send(options) { - const { broker, queue, type: amqpType, data, 'jms-mode': jmsMode, headers: headersJson, properties: propsJson } = options; + const { broker, queue, type: amqpType, data, 'jms-mode': jmsMode, headers: headersJson, properties: propsJson, 'message-header': messageHeaderJson } = options; const headers = headersJson ? JSON.parse(headersJson) : null; const properties = propsJson ? JSON.parse(propsJson) : null; + const messageHeader = messageHeaderJson ? JSON.parse(messageHeaderJson) : null; const testData = JSON.parse(data); let sentCount = 0; @@ -1108,6 +1109,13 @@ function send(options) { message.application_properties = appProps; } + if (messageHeader) { + if (messageHeader.durable !== undefined) message.durable = messageHeader.durable; + if (messageHeader.priority !== undefined) message.priority = messageHeader.priority; + if (messageHeader.ttl !== undefined) message.ttl = messageHeader.ttl; + if (messageHeader.first_acquirer !== undefined) message.first_acquirer = messageHeader.first_acquirer; + } + context.sender.send(message); sentCount++; @@ -1396,6 +1404,14 @@ function receive(options) { } } + msgData.message_header = { + durable: context.message.durable || false, + priority: context.message.priority !== undefined ? context.message.priority : 4, + ttl: context.message.ttl || 0, + first_acquirer: context.message.first_acquirer || false, + delivery_count: context.message.delivery_count || 0, + }; + messages.push(msgData); if (messages.length >= expectedCount) { diff --git a/shims/python-proton/shim.py b/shims/python-proton/shim.py index 268ec48..e349664 100755 --- a/shims/python-proton/shim.py +++ b/shims/python-proton/shim.py @@ -225,6 +225,7 @@ class SenderHandler(MessagingHandler): jms_mode: bool = False, amqp_type: str = "string", headers: dict[str, Any] | None = None, properties: dict[str, Any] | None = None, + message_header: dict[str, Any] | None = None, ) -> None: super().__init__() self.url = url @@ -234,6 +235,7 @@ class SenderHandler(MessagingHandler): self.amqp_type = amqp_type self.headers = headers self.properties = properties + self.message_header = message_header self.sent_count = 0 self.confirmed_count = 0 @@ -281,6 +283,9 @@ class SenderHandler(MessagingHandler): if self.properties: self._apply_properties(msg) + if self.message_header: + self._apply_message_header(msg) + event.sender.send(msg) self.sent_count += 1 @@ -376,6 +381,18 @@ class SenderHandler(MessagingHandler): props[name] = str(pval) msg.properties = props + def _apply_message_header(self, msg: Message) -> None: + """Set AMQP Header section fields on the message.""" + if "durable" in self.message_header: + msg.durable = self.message_header["durable"] + if "priority" in self.message_header: + msg.priority = int(self.message_header["priority"]) + if "ttl" in self.message_header: + # Proton Python msg.ttl is in seconds; input is AMQP ms + msg.ttl = float(self.message_header["ttl"]) / 1000.0 + if "first_acquirer" in self.message_header: + msg.first_acquirer = self.message_header["first_acquirer"] + def _encode_value(self, amqp_type: str, value: Any) -> Any: """Encode test value to AMQP type.""" if amqp_type == "null": @@ -534,6 +551,8 @@ class ReceiverHandler(MessagingHandler): if properties: msg_data["properties"] = properties + msg_data["message_header"] = self._extract_message_header(msg) + self.received_messages.append(msg_data) # Close when all messages received @@ -624,6 +643,17 @@ class ReceiverHandler(MessagingHandler): properties[name] = prop return properties + def _extract_message_header(self, msg: Message) -> dict[str, Any]: + """Extract AMQP Header section fields.""" + return { + "durable": msg.durable, + "priority": int(msg.priority), + # Proton Python msg.ttl is in seconds; output as AMQP ms + "ttl": int(float(msg.ttl) * 1000), + "first_acquirer": msg.first_acquirer, + "delivery_count": int(msg.delivery_count), + } + def _decode_jms_message(self, msg: Message, jms_msg_type: int) -> dict[str, Any]: """Decode JMS message based on message type annotation.""" # JMS message type constants @@ -1038,7 +1068,8 @@ def send_messages(args: argparse.Namespace) -> None: jms_mode = getattr(args, "jms_mode", False) headers = json.loads(args.headers) if args.headers else None properties = json.loads(args.properties) if getattr(args, "properties", None) else None - handler = SenderHandler(args.broker, args.queue, messages, jms_mode, args.type, headers, properties) + message_header = json.loads(args.message_header) if getattr(args, "message_header", None) else None + handler = SenderHandler(args.broker, args.queue, messages, jms_mode, args.type, headers, properties, message_header) Container(handler).run() # Output result @@ -1096,6 +1127,7 @@ def main() -> None: ) send_parser.add_argument("--headers", default=None, help="JSON JMS headers") send_parser.add_argument("--properties", default=None, help="JSON JMS application properties") + send_parser.add_argument("--message-header", default=None, help="JSON AMQP Header section fields") send_parser.add_argument("--large-content", default=None, help="Large content type (binary, string, list, array, map, described)") send_parser.add_argument("--size", type=int, default=None, help="Large content size in bytes (binary/string)") send_parser.add_argument("--seed", type=int, default=None, help="PRNG seed for large content") diff --git a/tests/test_amqp_headers.py b/tests/test_amqp_headers.py new file mode 100644 index 0000000..f42a59c --- /dev/null +++ b/tests/test_amqp_headers.py @@ -0,0 +1,291 @@ +""" +AMQP Message Header Section Interoperability Tests (Phase 2d) + +Tests AMQP 1.0 Header section fields across the 5×5 AMQP client matrix: + durable, priority, ttl, first-acquirer, delivery-count + +Test Pairs: +- AMQP N×N (25 pairs): all 5 AMQP clients against each other (no JMS) + +275 total tests: + durable: 25 pairs × 2 values = 50 + priority: 25 pairs × 4 values = 100 + ttl: 25 pairs × 2 values = 50 + first_acquirer: 25 pairs × 2 values = 50 + delivery_count: 25 pairs × 1 (verify 0) = 25 +""" + +import itertools +import json +import os +import subprocess +from pathlib import Path +from typing import Any + +import pytest + + +# ============================================================================= +# Client Configurations +# ============================================================================= + +AMQP_CLIENTS = [ + "python-proton", + "javascript-rhea", + "cpp-proton", + "dotnet-proton", + "java-protonj2", +] + +AMQP_PAIRS = [ + pytest.param(s, r, id=f"{s}->{r}") + for s, r in itertools.product(AMQP_CLIENTS, repeat=2) +] + +CLIENT_INFO = { + "python-proton": { + "name": "Python Proton", + "send_cmd": lambda path: ["python3", str(path / "shims/python-proton/shim.py"), "send"], + "recv_cmd": lambda path: ["python3", str(path / "shims/python-proton/shim.py"), "receive"], + "broker_prefix": "amqp://", + }, + "javascript-rhea": { + "name": "JavaScript Rhea", + "send_cmd": lambda path: ["node", str(path / "shims/javascript-rhea/shim.js"), "send"], + "recv_cmd": lambda path: ["node", str(path / "shims/javascript-rhea/shim.js"), "receive"], + "broker_prefix": "amqp://", + }, + "cpp-proton": { + "name": "C++ Proton", + "send_cmd": lambda path: [str(path / "shims/cpp-proton/build/qit-shim-cpp"), "send"], + "recv_cmd": lambda path: [str(path / "shims/cpp-proton/build/qit-shim-cpp"), "receive"], + "broker_prefix": "amqp://", + }, + "dotnet-proton": { + "name": ".NET Proton", + "send_cmd": lambda path: [str(path / "shims/dotnet-proton/shim.sh"), "send"], + "recv_cmd": lambda path: [str(path / "shims/dotnet-proton/shim.sh"), "receive"], + "broker_prefix": "amqp://", + }, + "java-protonj2": { + "name": "Java ProtonJ2", + "send_cmd": lambda path: [str(path / "shims/java-protonj2/shim.sh"), "send"], + "recv_cmd": lambda path: [str(path / "shims/java-protonj2/shim.sh"), "receive"], + "broker_prefix": "amqp://", + }, +} + + +# ============================================================================= +# Fixtures +# ============================================================================= + [email protected] +def broker_url(): + return os.environ.get("QIT_BROKER_URL", "localhost:5672") + + [email protected] +def test_queue(): + import random + import string + suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=8)) + return f"qit.test.amqp_header.{suffix}" + + [email protected] +def project_root(): + return Path(__file__).parent.parent + + +# ============================================================================= +# Shim Runners +# ============================================================================= + +def run_sender( + client: str, + broker_url: str, + queue: str, + project_root: Path, + message_header: dict[str, Any] | None = None, + timeout: int = 30, +) -> dict[str, Any]: + info = CLIENT_INFO[client] + broker = info["broker_prefix"] + broker_url + + messages = [{"index": 0, "type": "string", "value": "header-test"}] + + cmd = info["send_cmd"](project_root) + [ + "--broker", broker, + "--queue", queue, + "--type", "string", + "--count", "1", + "--data", json.dumps(messages), + ] + + if message_header: + cmd += ["--message-header", json.dumps(message_header)] + + result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) + if result.returncode != 0: + pytest.fail(f"{info['name']} sender failed: {result.stderr}") + + return json.loads(result.stdout) + + +def run_receiver( + client: str, + broker_url: str, + queue: str, + project_root: Path, + timeout: int = 30, +) -> dict[str, Any]: + info = CLIENT_INFO[client] + broker = info["broker_prefix"] + broker_url + + cmd = info["recv_cmd"](project_root) + [ + "--broker", broker, + "--queue", queue, + "--count", "1", + "--timeout", str(timeout), + ] + + result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout + 5) + if result.returncode != 0: + pytest.fail(f"{info['name']} receiver failed: {result.stderr}") + + return json.loads(result.stdout) + + +# ============================================================================= +# Comparison Helpers +# ============================================================================= + +def assert_message_header_field( + field: str, + sent_value: Any, + received_header: dict[str, Any], + sender: str, + receiver: str, +) -> None: + assert "message_header" in received_header, ( + f"{sender}->{receiver}: receiver output missing 'message_header' key" + ) + mh = received_header["message_header"] + assert field in mh, ( + f"{sender}->{receiver}: message_header missing '{field}' field" + ) + recv_value = mh[field] + + if field == "ttl": + assert recv_value > 0, ( + f"{sender}->{receiver}: TTL expired (received 0, sent {sent_value})" + ) + assert recv_value <= sent_value, ( + f"{sender}->{receiver}: TTL increased " + f"(received {recv_value}, sent {sent_value})" + ) + elif field == "priority": + assert int(recv_value) == int(sent_value), ( + f"{sender}->{receiver}: priority mismatch — " + f"sent {sent_value}, received {recv_value}" + ) + else: + assert recv_value == sent_value, ( + f"{sender}->{receiver}: {field} mismatch — " + f"sent {sent_value}, received {recv_value}" + ) + + +# ============================================================================= +# Test Data +# ============================================================================= + +DURABLE_VALUES = [True, False] +PRIORITY_VALUES = [0, 4, 7, 9] +TTL_VALUES = [60000, 300000] +FIRST_ACQUIRER_VALUES = [True, False] + + +# ============================================================================= +# Tests +# ============================================================================= + [email protected]("sender_client,receiver_client", AMQP_PAIRS) [email protected]("durable", DURABLE_VALUES, ids=lambda v: f"durable={v}") +def test_amqp_header_durable( + sender_client, receiver_client, durable, broker_url, test_queue, project_root +): + message_header = {"durable": durable} + run_sender(sender_client, broker_url, test_queue, project_root, + message_header=message_header) + recv_result = run_receiver(receiver_client, broker_url, test_queue, project_root) + received = recv_result["messages"] + assert len(received) == 1 + assert_message_header_field("durable", durable, received[0], + sender_client, receiver_client) + + [email protected]("sender_client,receiver_client", AMQP_PAIRS) [email protected]("priority", PRIORITY_VALUES, ids=lambda v: f"priority={v}") +def test_amqp_header_priority( + sender_client, receiver_client, priority, broker_url, test_queue, project_root +): + message_header = {"priority": priority} + run_sender(sender_client, broker_url, test_queue, project_root, + message_header=message_header) + recv_result = run_receiver(receiver_client, broker_url, test_queue, project_root) + received = recv_result["messages"] + assert len(received) == 1 + assert_message_header_field("priority", priority, received[0], + sender_client, receiver_client) + + [email protected]("sender_client,receiver_client", AMQP_PAIRS) [email protected]("ttl", TTL_VALUES, ids=lambda v: f"ttl={v}") +def test_amqp_header_ttl( + sender_client, receiver_client, ttl, broker_url, test_queue, project_root +): + message_header = {"ttl": ttl} + run_sender(sender_client, broker_url, test_queue, project_root, + message_header=message_header) + recv_result = run_receiver(receiver_client, broker_url, test_queue, project_root) + received = recv_result["messages"] + assert len(received) == 1 + assert_message_header_field("ttl", ttl, received[0], + sender_client, receiver_client) + + [email protected]("sender_client,receiver_client", AMQP_PAIRS) [email protected]("first_acquirer", FIRST_ACQUIRER_VALUES, + ids=lambda v: f"first_acquirer={v}") +def test_amqp_header_first_acquirer( + sender_client, receiver_client, first_acquirer, broker_url, test_queue, project_root +): + message_header = {"first_acquirer": first_acquirer} + run_sender(sender_client, broker_url, test_queue, project_root, + message_header=message_header) + recv_result = run_receiver(receiver_client, broker_url, test_queue, project_root) + received = recv_result["messages"] + assert len(received) == 1 + assert_message_header_field("first_acquirer", first_acquirer, received[0], + sender_client, receiver_client) + + [email protected]("sender_client,receiver_client", AMQP_PAIRS) +def test_amqp_header_delivery_count( + sender_client, receiver_client, broker_url, test_queue, project_root +): + run_sender(sender_client, broker_url, test_queue, project_root) + recv_result = run_receiver(receiver_client, broker_url, test_queue, project_root) + received = recv_result["messages"] + assert len(received) == 1 + assert "message_header" in received[0], ( + f"{sender_client}->{receiver_client}: " + "receiver output missing 'message_header' key" + ) + assert int(received[0]["message_header"]["delivery_count"]) == 0, ( + f"{sender_client}->{receiver_client}: " + f"delivery_count should be 0 on first delivery, " + f"got {received[0]['message_header']['delivery_count']}" + ) --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
