espressif/esp-mqtt
1
0
Fork 0
mirror of https://github.com/espressif/esp-mqtt.git synced 2026-10-11 18:27:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

ci: Add QoS flow and large-payload conformance cases

Browse Source 
Witness Client sender QoS 0/1/2 on the wire (DUP, packet id, PUBREL) and
assert esp-mqtt splits oversized inbound payloads into DATA chunks.
This commit is contained in:
Euripedes Rocha Filho
2026-08-28 07:41:40 +02:00
parent 4d36e70732
commit 71442e5e9c
1 changed files with 164 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/apps/mqtt_conformance/pytest_mqtt_conformance.py
+164
View File
@@ -157,6 +157,14 @@ def mqtt_packet_type_of(packet: object) -> MqttPacketType:
return MqttPacketType(raw)
def mqtt_fixed_header_qos(packet: object) -> int:
return int(getattr(getattr(packet, "fh"), "QoS"))
def mqtt_fixed_header_dup(packet: object) -> bool:
return bool(getattr(getattr(packet, "fh"), "DUP"))
class BrokerInterface(Protocol):
uri: str
@@ -192,6 +200,8 @@ class BrokerInterface(Protocol):
def inbound_packet_identifiers(self, packet_type: MqttPacketType) -> list[int]: ...
def inbound_packets(self, packet_type: MqttPacketType) -> list[object]: ...
def set_receive_maximum(self, receive_maximum: int) -> None: ...
def disconnect_clients(self) -> None: ...
@@ -379,6 +389,10 @@ def _start_paho_broker(
with inbound_condition:
return [mqtt_packet_identifier_of(packet) for packet in inbound_packets.get(packet_type, ())]
def inbound_packets(self, packet_type: MqttPacketType) -> list[object]:
with inbound_condition:
return list(inbound_packets.get(packet_type, ()))
def set_receive_maximum(self, receive_maximum: int) -> None:
if not 1 <= receive_maximum <= 0xFFFF:
raise ValueError("Receive Maximum must be in the range 1..65535")
@@ -1559,3 +1573,153 @@ def test_client_pubrel_follows_pubrec_receive_order__mqtt_4_6_0_4(dut: Dut, prot
assert pubrel_ids == pubrec_release_order, (
f"MQTT-4.6.0-4: PUBREL order {pubrel_ids} does not match PUBREC receive order {pubrec_release_order}"
)
@pytest.mark.eth_ip101
@pytest.mark.timeout(
case_timeout(
connect_operations=1,
publish_operations=1,
event_wait_operations=2,
)
)
@idf_parametrize("target", ["esp32"], indirect=["target"])
@pytest.mark.parametrize(
"protocol_ver",
[MQTT_PROTOCOL_V_3_1_1, MQTT_PROTOCOL_V_5],
ids=["v311", "v5"],
)
@pytest.mark.parametrize("qos", [0, 1, 2], ids=["qos0", "qos1", "qos2"])
def test_mqtt_publish_subscribe_at_all_qos_levels(dut: Dut, protocol_ver: int, qos: int) -> None:
"""DUT PUBLISH on the wire uses the requested QoS flow (sender role).
Claim ID: MQTT-4.3.1-1 (qos 0); MQTT-4.3.2-1 / MQTT-4.3.2-2 (qos 1);
MQTT-4.3.3-1 / MQTT-4.3.3-2 / MQTT-4.3.3-4 (qos 2)
Version: 3.1.1, 5.0
Section: 4.3 Quality of Service levels and protocol flows
Party: Client (sender)
Quote: QoS 0 sender MUST send PUBLISH with QoS 0 and DUP 0 [MQTT-4.3.1-1].
QoS 1 sender MUST assign an unused Packet Identifier and send PUBLISH with
QoS 1 and DUP 0 [MQTT-4.3.2-1] [MQTT-4.3.2-2].
QoS 2 sender MUST assign an unused Packet Identifier, send PUBLISH with
QoS 2 and DUP 0, and send PUBREL with that identifier after PUBREC
[MQTT-4.3.3-1] [MQTT-4.3.3-2] [MQTT-4.3.3-4].
DUT observable: inbound PUBLISH fh.QoS and DUP; qos>0 unused packet id;
qos 2 inbound PUBREL with the same packet id
Violating DUT still passes?: no — a DUT that sends the wrong QoS, sets DUP
on a new publish, omits a packet id, or skips PUBREL fails the wire asserts
"""
topic = build_topic()
record_types: tuple[MqttPacketType, ...] = (MqttPacketType.PUBLISH,)
if qos == 2:
record_types = (MqttPacketType.PUBLISH, MqttPacketType.PUBREL)
with (
broker_started(record_inbound_types=record_types) as broker,
initialized_mqtt_client(dut, broker.uri, protocol_ver=protocol_ver) as client,
started_client(client),
):
publish_from_dut(client, topic, qos, payload_prefix=f"qos{qos}", message_count=1)
expect_n(client, {b"Publish requested, msg_id=": 1}, timeout=DUT_CMD_TIMEOUT)
broker.wait_for_inbound_packets(MqttPacketType.PUBLISH, 1, timeout=DUT_EVENT_TIMEOUT)
publish_pkt = broker.inbound_packets(MqttPacketType.PUBLISH)[0]
assert mqtt_fixed_header_qos(publish_pkt) == qos, (
f"MQTT-4.3: inbound PUBLISH QoS {mqtt_fixed_header_qos(publish_pkt)} != {qos}"
)
assert mqtt_fixed_header_dup(publish_pkt) is False, "MQTT-4.3: new PUBLISH MUST have DUP 0"
if qos == 0:
return
packet_id = mqtt_packet_identifier_of(publish_pkt)
assert packet_id != 0, "MQTT-4.3.2/4.3.3: QoS > 0 PUBLISH MUST carry a Packet Identifier"
if qos == 1:
expect_n(client, {b"MQTT_EVENT_PUBLISHED": 1}, timeout=DUT_EVENT_TIMEOUT)
return
broker.wait_for_inbound_packets(MqttPacketType.PUBREL, 1, timeout=DUT_EVENT_TIMEOUT)
pubrel_id = broker.inbound_packet_identifiers(MqttPacketType.PUBREL)[0]
assert pubrel_id == packet_id, (
f"MQTT-4.3.3-4: PUBREL id {pubrel_id} != PUBLISH id {packet_id}"
)
expect_n(client, {b"MQTT_EVENT_PUBLISHED": 1}, timeout=DUT_EVENT_TIMEOUT)
@pytest.mark.eth_ip101
@pytest.mark.timeout(
case_timeout(
connect_operations=1,
subscribe_operations=1,
event_wait_operations=2,
)
)
@idf_parametrize("target", ["esp32"], indirect=["target"])
@pytest.mark.parametrize(
"protocol_ver",
[MQTT_PROTOCOL_V_3_1_1, MQTT_PROTOCOL_V_5],
ids=["v311", "v5"],
)
def test_mqtt_large_payload_fragmentation(dut: Dut, protocol_ver: int) -> None:
"""Inbound payload larger than the rx buffer is delivered as offset chunks.
Claim ID: API (not an MQTT SHALL) — esp-mqtt MQTT_EVENT_DATA reassembly
Version: 3.1.1, 5.0
Section: n/a (client rx buffer MQTT_BUFFER_SIZE_BYTE, default 1024)
Party: Client (application delivery)
Quote: n/a — MQTT delivers one Application Message; this asserts the DUT
splits it into MQTT_EVENT_DATA chunks and reports MQTT_EVENT_DATA_COMPLETE
DUT observable: two or more MQTT_EVENT_DATA lines with offset 0, 0+len, …
covering total=5000, then MQTT_EVENT_DATA_COMPLETE total=5000
Violating DUT still passes?: no — a single DATA event with total=5000 fails
the chunk-count assert; gaps or a wrong total fail the offset walk
"""
topic = build_topic()
payload_size = 5000
payload = b"X" * payload_size
data_line = re.compile(rb"MQTT_EVENT_DATA .*qos=\d+ len=(\d+) offset=(\d+) total=(\d+)")
with (
broker_started() as broker,
initialized_mqtt_client(dut, broker.uri, protocol_ver=protocol_ver) as client,
started_client(client),
):
subscribed_to(client, topic, 1)
host, _, port = broker.uri.removeprefix("mqtt://").partition(":")
protocol = "mqtt5" if protocol_ver == MQTT_PROTOCOL_V_5 else "mqtt311"
with host_publisher(
protocol=protocol,
client_id="paho-large-payload-pub",
host=host,
port=int(port),
) as pub:
pub.publish(topic, payload, 1)
chunks: list[tuple[int, int, int]] = []
deadline = time.monotonic() + DUT_EVENT_TIMEOUT
while True:
remaining = deadline - time.monotonic()
if remaining <= 0:
raise pexpect.TIMEOUT(f"Timed out collecting DATA chunks; got {chunks}")
match = client.expect(data_line, timeout=remaining)
length = int(match.group(1))
offset = int(match.group(2))
total = int(match.group(3))
chunks.append((offset, length, total))
if offset + length == total:
break
assert len(chunks) >= 2, f"expected multiple DATA chunks, got {chunks}"
assert chunks[0][0] == 0, f"first DATA offset must be 0, got {chunks}"
assert all(total == payload_size for _, _, total in chunks), (
f"DATA total must stay {payload_size}, got {chunks}"
)
for index in range(len(chunks) - 1):
offset, length, _ = chunks[index]
next_offset = chunks[index + 1][0]
assert offset + length == next_offset, f"DATA offsets not contiguous: {chunks}"
last_offset, last_length, _ = chunks[-1]
assert last_offset + last_length == payload_size
client.expect(
re.compile(rb"MQTT_EVENT_DATA_COMPLETE msg_id=\d+ total=" + str(payload_size).encode()),
timeout=DUT_CMD_TIMEOUT,
)