From fa88c3c64f9af56b362fe0b2d2a05ee2924edd7c Mon Sep 17 00:00:00 2001 From: Carter Tinney Date: Fri, 21 Aug 2026 16:08:26 -0700 Subject: [PATCH] fix: validate encoded telemetry message size Measure IoT Hub messages from their encoded payload and logical property contribution instead of CPython object layout, and make boundary coverage deterministic across supported Python versions. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../iot/device/iothub/aio/async_clients.py | 4 +- .../azure/iot/device/iothub/models/message.py | 86 +++++++++-- .../iothub/pipeline/mqtt_topic_iothub.py | 54 ++----- .../azure/iot/device/iothub/sync_clients.py | 4 +- tests/unit/iothub/aio/test_async_clients.py | 136 +++++++----------- tests/unit/iothub/models/test_message.py | 134 ++++++++++++++++- tests/unit/iothub/test_sync_clients.py | 130 +++++++---------- 7 files changed, 316 insertions(+), 232 deletions(-) diff --git a/azure-iot-device/azure/iot/device/iothub/aio/async_clients.py b/azure-iot-device/azure/iot/device/iothub/aio/async_clients.py index 6eaca9c1a..08cb3288d 100644 --- a/azure-iot-device/azure/iot/device/iothub/aio/async_clients.py +++ b/azure-iot-device/azure/iot/device/iothub/aio/async_clients.py @@ -648,11 +648,11 @@ async def send_message_to_output(self, message: Union[Message, str], output_name if not isinstance(message, Message): message = Message(message) + message.output_name = output_name + if message.get_size() > device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT: raise ValueError("Size of message can not exceed 256 KB.") - message.output_name = output_name - logger.info("Sending message to output:" + output_name + "...") send_output_message_async = async_adapter.emulate_async( self._mqtt_pipeline.send_output_message diff --git a/azure-iot-device/azure/iot/device/iothub/models/message.py b/azure-iot-device/azure/iot/device/iothub/models/message.py index 8332ec090..a7033ed96 100644 --- a/azure-iot-device/azure/iot/device/iothub/models/message.py +++ b/azure-iot-device/azure/iot/device/iothub/models/message.py @@ -5,8 +5,64 @@ # -------------------------------------------------------------------------- """This module contains a class representing messages that are sent or received. """ +from datetime import date + from azure.iot.device import constant -import sys + + +def _encode_message_data(data): + if isinstance(data, str): + return data.encode("utf-8") + if isinstance(data, (int, float)): + return str(data).encode("ascii") + if data is None: + return b"" + if not isinstance(data, (bytes, bytearray)): + raise TypeError("Message data must be a string, bytes, bytearray, int, float, or None.") + return data + + +def _get_system_properties(message): + properties = [] + if message.output_name: + properties.append(("$.on", str(message.output_name))) + if message.message_id: + properties.append(("$.mid", str(message.message_id))) + if message.correlation_id: + properties.append(("$.cid", str(message.correlation_id))) + if message.user_id: + properties.append(("$.uid", str(message.user_id))) + if message.content_type: + properties.append(("$.ct", str(message.content_type))) + if message.content_encoding: + properties.append(("$.ce", str(message.content_encoding))) + if message.iothub_interface_id: + properties.append(("$.ifid", str(message.iothub_interface_id))) + if message.expiry_time_utc: + expiry_time = ( + message.expiry_time_utc.isoformat() + if isinstance(message.expiry_time_utc, date) + else message.expiry_time_utc + ) + properties.append(("$.exp", str(expiry_time))) + return properties + + +def _get_custom_properties(message): + if not message.custom_properties: + return [] + + properties = [(str(key), str(value)) for key, value in message.custom_properties.items()] + properties.sort() + + keys = [key for key, _ in properties] + if len(keys) != len(set(keys)): + raise ValueError("Duplicate keys in custom properties!") + return properties + + +def _get_string_size(value): + return len(value.encode("utf-8")) class Message(object): @@ -65,14 +121,22 @@ def __str__(self): return str(self.data) def get_size(self) -> int: - total = 0 - total = total + sum( - sys.getsizeof(v) - for v in self.__dict__.values() - if v is not None and v is not self.custom_properties + """Return the message size in bytes as measured by IoT Hub. + + The size is the encoded body plus system property values and application property names + and values. Strings are measured as UTF-8, matching the MQTT transport; bytes and + bytearrays are measured as-is. MQTT topic and packet overhead are not included. + + :raises TypeError: If the message data is not a payload type supported by the MQTT + transport. + :raises ValueError: If custom property keys are duplicated after string conversion. + """ + payload_size = len(_encode_message_data(self.data)) + system_property_size = sum( + _get_string_size(value) for _, value in _get_system_properties(self) + ) + application_property_size = sum( + _get_string_size(key) + _get_string_size(value) + for key, value in _get_custom_properties(self) ) - if self.custom_properties: - total = total + sum( - sys.getsizeof(v) for v in self.custom_properties.values() if v is not None - ) - return total + return payload_size + system_property_size + application_property_size diff --git a/azure-iot-device/azure/iot/device/iothub/pipeline/mqtt_topic_iothub.py b/azure-iot-device/azure/iot/device/iothub/pipeline/mqtt_topic_iothub.py index 9db638882..3538625d3 100644 --- a/azure-iot-device/azure/iot/device/iothub/pipeline/mqtt_topic_iothub.py +++ b/azure-iot-device/azure/iot/device/iothub/pipeline/mqtt_topic_iothub.py @@ -5,9 +5,13 @@ # -------------------------------------------------------------------------- import logging -from datetime import date import urllib +from azure.iot.device.iothub.models.message import ( + _get_custom_properties, + _get_system_properties, +) + logger = logging.getLogger(__name__) # NOTE: Whenever using standard URL encoding via the urllib.parse.quote() API @@ -343,59 +347,17 @@ def encode_message_properties_in_topic(message_to_send, topic): "devices//modules//messages/events/ :return: The topic which has been uri-encoded """ - system_properties = [] - if message_to_send.output_name: - system_properties.append(("$.on", str(message_to_send.output_name))) - if message_to_send.message_id: - system_properties.append(("$.mid", str(message_to_send.message_id))) - - if message_to_send.correlation_id: - system_properties.append(("$.cid", str(message_to_send.correlation_id))) - - if message_to_send.user_id: - system_properties.append(("$.uid", str(message_to_send.user_id))) - - if message_to_send.content_type: - system_properties.append(("$.ct", str(message_to_send.content_type))) - - if message_to_send.content_encoding: - system_properties.append(("$.ce", str(message_to_send.content_encoding))) - - if message_to_send.iothub_interface_id: - system_properties.append(("$.ifid", str(message_to_send.iothub_interface_id))) - - if message_to_send.expiry_time_utc: - system_properties.append( - ( - "$.exp", - message_to_send.expiry_time_utc.isoformat() # returns string - if isinstance(message_to_send.expiry_time_utc, date) - else message_to_send.expiry_time_utc, - ) - ) - + system_properties = _get_system_properties(message_to_send) system_properties_encoded = urllib.parse.urlencode( system_properties, quote_via=urllib.parse.quote ) topic += system_properties_encoded - if message_to_send.custom_properties and len(message_to_send.custom_properties) > 0: + custom_prop_seq = _get_custom_properties(message_to_send) + if custom_prop_seq: if system_properties and len(system_properties) > 0: topic += "&" - # Convert the custom properties to a sorted list in order to ensure the - # resulting ordering in the topic string is consistent across versions of Python. - # Convert to the properties to strings for safety. - custom_prop_seq = [ - (str(i[0]), str(i[1])) for i in list(message_to_send.custom_properties.items()) - ] - custom_prop_seq.sort() - - # Validate that string conversion has not created duplicate keys - keys = [i[0] for i in custom_prop_seq] - if len(keys) != len(set(keys)): - raise ValueError("Duplicate keys in custom properties!") - user_properties_encoded = urllib.parse.urlencode( custom_prop_seq, quote_via=urllib.parse.quote ) diff --git a/azure-iot-device/azure/iot/device/iothub/sync_clients.py b/azure-iot-device/azure/iot/device/iothub/sync_clients.py index 4088e0b6f..6023f31e6 100644 --- a/azure-iot-device/azure/iot/device/iothub/sync_clients.py +++ b/azure-iot-device/azure/iot/device/iothub/sync_clients.py @@ -669,11 +669,11 @@ def send_message_to_output(self, message: Union[Message, str], output_name: str) if not isinstance(message, Message): message = Message(message) + message.output_name = output_name + if message.get_size() > device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT: raise ValueError("Size of message can not exceed 256 KB.") - message.output_name = output_name - logger.info("Sending message to output:" + output_name + "...") callback = EventedCallback() diff --git a/tests/unit/iothub/aio/test_async_clients.py b/tests/unit/iothub/aio/test_async_clients.py index 3796f494a..dfa0cc8e0 100644 --- a/tests/unit/iothub/aio/test_async_clients.py +++ b/tests/unit/iothub/aio/test_async_clients.py @@ -9,7 +9,6 @@ import asyncio import time import urllib -import sys from azure.iot.device import exceptions as client_exceptions from azure.iot.device.common.auth import sastoken as st from azure.iot.device.iothub.aio import IoTHubDeviceClient, IoTHubModuleClient @@ -659,10 +658,10 @@ def fail_send_message(message, callback): [ pytest.param("message", id="String input"), pytest.param(222, id="Integer input"), - pytest.param(object(), id="Object input"), + pytest.param(1.5, id="Float input"), + pytest.param(b"message", id="Bytes input"), + pytest.param(bytearray(b"message"), id="Bytearray input"), pytest.param(None, id="None input"), - pytest.param([1, "str"], id="List input"), - pytest.param({"a": 2}, id="Dictionary input"), ], ) async def test_wraps_data_in_message_and_calls_pipeline_send_message( @@ -674,44 +673,27 @@ async def test_wraps_data_in_message_and_calls_pipeline_send_message( assert isinstance(sent_message, Message) assert sent_message.data == message_input - @pytest.mark.it("Raises error when message data size is greater than 256 KB") - async def test_raises_error_when_message_data_greater_than_256(self, client, mqtt_pipeline): - data_input = "serpensortia" * 256000 - message = Message(data_input) - with pytest.raises(ValueError) as e_info: - await client.send_message(message) - assert "256 KB" in e_info.value.args[0] - assert mqtt_pipeline.send_message.call_count == 0 - - @pytest.mark.it("Raises error when message size is greater than 256 KB") - async def test_raises_error_when_message_size_greater_than_256(self, client, mqtt_pipeline): - data_input = "serpensortia" - message = Message(data_input) - message.custom_properties["spell"] = data_input * 256000 - with pytest.raises(ValueError) as e_info: - await client.send_message(message) - assert "256 KB" in e_info.value.args[0] - assert mqtt_pipeline.send_message.call_count == 0 - - @pytest.mark.skipif( - sys.version_info >= (3, 12), - reason="Python 3.12 appears to have an issue. Investigate further.", - ) - @pytest.mark.it("Does not raises error when message data size is equal to 256 KB") - async def test_raises_error_when_message_data_equal_to_256(self, client, mqtt_pipeline): - data_input = "a" * 262095 - message = Message(data_input) - # This check was put as message class may undergo the default content type encoding change - # and the above calculation will change. - if message.get_size() != device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT: - assert False - - await client.send_message(message) + @pytest.mark.it("Validates message size at the 256 KB boundary") + @pytest.mark.parametrize( + "size_delta, raises_error", + [ + pytest.param(-1, False, id="Below limit"), + pytest.param(0, False, id="At limit"), + pytest.param(1, True, id="Above limit"), + ], + ) + async def test_validates_message_size(self, client, mqtt_pipeline, size_delta, raises_error): + message_size = device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT + size_delta + message = Message(b"a" * message_size) - assert mqtt_pipeline.send_message.call_count == 1 - sent_message = mqtt_pipeline.send_message.call_args[0][0] - assert isinstance(sent_message, Message) - assert sent_message.data == data_input + if raises_error: + with pytest.raises(ValueError, match="256 KB"): + await client.send_message(message) + assert mqtt_pipeline.send_message.call_count == 0 + else: + await client.send_message(message) + assert mqtt_pipeline.send_message.call_count == 1 + assert mqtt_pipeline.send_message.call_args[0][0] is message class SharedClientReceiveMethodRequestTests(object): @@ -2061,10 +2043,10 @@ def fail_send_output_message(message, callback): [ pytest.param("message", id="String input"), pytest.param(222, id="Integer input"), - pytest.param(object(), id="Object input"), + pytest.param(1.5, id="Float input"), + pytest.param(b"message", id="Bytes input"), + pytest.param(bytearray(b"message"), id="Bytearray input"), pytest.param(None, id="None input"), - pytest.param([1, "str"], id="List input"), - pytest.param({"a": 2}, id="Dictionary input"), ], ) async def test_send_message_to_output_calls_pipeline_wraps_data_in_message( @@ -2077,53 +2059,31 @@ async def test_send_message_to_output_calls_pipeline_wraps_data_in_message( assert isinstance(sent_message, Message) assert sent_message.data == message_input - @pytest.mark.it("Raises error when message data size is greater than 256 KB") - async def test_raises_error_when_message_to_output_data_greater_than_256( - self, client, mqtt_pipeline - ): - output_name = "some_output" - data_input = "serpensortia" * 256000 - message = Message(data_input) - with pytest.raises(ValueError) as e_info: - await client.send_message_to_output(message, output_name) - assert "256 KB" in e_info.value.args[0] - assert mqtt_pipeline.send_output_message.call_count == 0 - - @pytest.mark.it("Raises error when message size is greater than 256 KB") - async def test_raises_error_when_message_to_output_size_greater_than_256( - self, client, mqtt_pipeline - ): - output_name = "some_output" - data_input = "serpensortia" - message = Message(data_input) - message.custom_properties["spell"] = data_input * 256000 - with pytest.raises(ValueError) as e_info: - await client.send_message_to_output(message, output_name) - assert "256 KB" in e_info.value.args[0] - assert mqtt_pipeline.send_output_message.call_count == 0 - - @pytest.mark.skipif( - sys.version_info >= (3, 12), - reason="Python 3.12 appears to have an issue. Investigate further.", + @pytest.mark.it("Validates message size at the 256 KB boundary") + @pytest.mark.parametrize( + "size_delta, raises_error", + [ + pytest.param(-1, False, id="Below limit"), + pytest.param(0, False, id="At limit"), + pytest.param(1, True, id="Above limit"), + ], ) - @pytest.mark.it("Does not raises error when message data size is equal to 256 KB") - async def test_raises_error_when_message_to_output_data_equal_to_256( - self, client, mqtt_pipeline - ): + async def test_validates_message_size(self, client, mqtt_pipeline, size_delta, raises_error): output_name = "some_output" - data_input = "a" * 262095 - message = Message(data_input) - # This check was put as message class may undergo the default content type encoding change - # and the above calculation will change. - if message.get_size() != device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT: - assert False - - await client.send_message_to_output(message, output_name) + message_size = device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT + size_delta + output_name_size = len(output_name.encode("utf-8")) + message = Message(b"a" * (message_size - output_name_size)) + + if raises_error: + with pytest.raises(ValueError, match="256 KB"): + await client.send_message_to_output(message, output_name) + assert mqtt_pipeline.send_output_message.call_count == 0 + else: + await client.send_message_to_output(message, output_name) + assert mqtt_pipeline.send_output_message.call_count == 1 + assert mqtt_pipeline.send_output_message.call_args[0][0] is message - assert mqtt_pipeline.send_output_message.call_count == 1 - sent_message = mqtt_pipeline.send_output_message.call_args[0][0] - assert isinstance(sent_message, Message) - assert sent_message.data == data_input + assert message.get_size() == message_size @pytest.mark.describe("IoTHubModuleClient (Asynchronous) - .receive_message_on_input()") diff --git a/tests/unit/iothub/models/test_message.py b/tests/unit/iothub/models/test_message.py index d444a55fd..c9aa15b0d 100644 --- a/tests/unit/iothub/models/test_message.py +++ b/tests/unit/iothub/models/test_message.py @@ -4,8 +4,11 @@ # license information. # -------------------------------------------------------------------------- -import pytest +import datetime import logging + +import pytest + from azure.iot.device.iothub.models import Message from azure.iot.device import constant @@ -16,6 +19,50 @@ data_obj = Message(data_str) +def _make_unicode_message(target_size): + multibyte_character = "\N{SNAKE}" + remaining_size = target_size - len(multibyte_character.encode("utf-8")) + return Message(multibyte_character + ("a" * remaining_size)) + + +def _make_custom_property_message(target_size): + key = "custom" + value = "property" + property_size = len(key.encode("utf-8")) + len(value.encode("utf-8")) + message = Message(b"a" * (target_size - property_size)) + message.custom_properties[key] = value + return message + + +def _set_system_properties(message): + message.output_name = "output" + message.message_id = 1234 + message.correlation_id = 5678 + message.user_id = 4000 + message.content_type = "application/json" + message.content_encoding = "utf-8" + message.expiry_time_utc = datetime.datetime(2026, 8, 21, 15, 33, 42) + message.set_as_security_message() + return [ + "output", + "1234", + "5678", + "4000", + "application/json", + "utf-8", + constant.SECURITY_MESSAGE_INTERFACE_ID, + "2026-08-21T15:33:42", + ] + + +def _make_system_property_message(target_size): + message = Message(b"") + values = _set_system_properties(message) + property_size = sum(len(value.encode("utf-8")) for value in values) + message.data = b"a" * (target_size - property_size) + return message + + @pytest.mark.describe("Message") class TestMessage(object): @pytest.mark.it("Instantiates from data type") @@ -105,3 +152,88 @@ def test_setting_message_as_security_message(self): assert msg.iothub_interface_id is None msg.set_as_security_message() assert msg.iothub_interface_id == constant.SECURITY_MESSAGE_INTERFACE_ID + + @pytest.mark.it("Measures payload using the MQTT transport encoding") + @pytest.mark.parametrize( + "data, expected_size", + [ + pytest.param("message", 7, id="String"), + pytest.param("\N{LATIN SMALL LETTER E WITH ACUTE}\N{SNAKE}", 6, id="Unicode string"), + pytest.param(b"\x00\xff", 2, id="Bytes"), + pytest.param(bytearray(b"\x00\xff"), 2, id="Bytearray"), + pytest.param(1234, 4, id="Integer"), + pytest.param(-1.25, 5, id="Float"), + pytest.param(None, 0, id="None"), + ], + ) + def test_get_size_payload_types(self, data, expected_size): + assert Message(data).get_size() == expected_size + + @pytest.mark.it("Raises TypeError for payload types not supported by the MQTT transport") + @pytest.mark.parametrize( + "data", + [ + pytest.param({"a": 1}, id="Dictionary"), + pytest.param([1, 2, 3], id="List"), + pytest.param(object(), id="Object"), + ], + ) + def test_get_size_invalid_payload_type(self, data): + with pytest.raises(TypeError): + Message(data).get_size() + + @pytest.mark.it("Counts custom property names and values after string conversion") + def test_get_size_custom_properties(self): + message = Message(b"body") + message.custom_properties = {1: 23, "custom": "property"} + + expected_property_size = len("1") + len("23") + len("custom") + len("property") + assert message.get_size() == len(b"body") + expected_property_size + + @pytest.mark.it("Counts system property values but not their names") + def test_get_size_system_properties(self): + message = Message(b"body") + values = _set_system_properties(message) + + expected_property_size = sum(len(value.encode("utf-8")) for value in values) + assert message.get_size() == len(b"body") + expected_property_size + + @pytest.mark.it("Does not count receive-only properties") + def test_get_size_receive_only_properties(self): + message = Message("body") + message.input_name = "input" + message.ack = "full" + + assert message.get_size() == len("body") + + @pytest.mark.it("Does not count MQTT topic encoding or system property names") + def test_get_size_excludes_protocol_overhead(self): + message = Message(b"") + message.message_id = "#" + message.custom_properties["#"] = "#" + + assert message.get_size() == 3 + + @pytest.mark.it("Is deterministic below, at, and above the 256 KB limit") + @pytest.mark.parametrize( + "message_factory", + [ + pytest.param(lambda target_size: Message(b"a" * target_size), id="Bytes payload"), + pytest.param(_make_unicode_message, id="Unicode payload"), + pytest.param(_make_custom_property_message, id="Custom properties"), + pytest.param(_make_system_property_message, id="System properties"), + ], + ) + @pytest.mark.parametrize( + "size_delta", + [ + pytest.param(-1, id="Below limit"), + pytest.param(0, id="At limit"), + pytest.param(1, id="Above limit"), + ], + ) + def test_get_size_boundary(self, message_factory, size_delta): + expected_size = constant.TELEMETRY_MESSAGE_SIZE_LIMIT + size_delta + message = message_factory(expected_size) + + assert message.get_size() == expected_size diff --git a/tests/unit/iothub/test_sync_clients.py b/tests/unit/iothub/test_sync_clients.py index 6f5385994..72da23b37 100644 --- a/tests/unit/iothub/test_sync_clients.py +++ b/tests/unit/iothub/test_sync_clients.py @@ -9,7 +9,6 @@ import threading import time import urllib -import sys from azure.iot.device.iothub import IoTHubDeviceClient, IoTHubModuleClient from azure.iot.device import exceptions as client_exceptions from azure.iot.device.common.auth import sastoken as st @@ -648,10 +647,10 @@ def test_raises_error_on_pipeline_op_error( [ pytest.param("message", id="String input"), pytest.param(222, id="Integer input"), - pytest.param(object(), id="Object input"), + pytest.param(1.5, id="Float input"), + pytest.param(b"message", id="Bytes input"), + pytest.param(bytearray(b"message"), id="Bytearray input"), pytest.param(None, id="None input"), - pytest.param([1, "str"], id="List input"), - pytest.param({"a": 2}, id="Dictionary input"), ], ) def test_wraps_data_in_message_and_calls_pipeline_send_message( @@ -663,44 +662,27 @@ def test_wraps_data_in_message_and_calls_pipeline_send_message( assert isinstance(sent_message, Message) assert sent_message.data == message_input - @pytest.mark.it("Raises error when message data size is greater than 256 KB") - def test_raises_error_when_message_data_greater_than_256(self, client, mqtt_pipeline): - data_input = "serpensortia" * 25600 - message = Message(data_input) - with pytest.raises(ValueError) as e_info: - client.send_message(message) - assert "256 KB" in e_info.value.args[0] - assert mqtt_pipeline.send_message.call_count == 0 - - @pytest.mark.it("Raises error when message size is greater than 256 KB") - def test_raises_error_when_message_size_greater_than_256(self, client, mqtt_pipeline): - data_input = "serpensortia" - message = Message(data_input) - message.custom_properties["spell"] = data_input * 25600 - with pytest.raises(ValueError) as e_info: - client.send_message(message) - assert "256 KB" in e_info.value.args[0] - assert mqtt_pipeline.send_message.call_count == 0 - - @pytest.mark.skipif( - sys.version_info >= (3, 12), - reason="Python 3.12 appears to have an issue. Investigate further.", - ) - @pytest.mark.it("Does not raises error when message data size is equal to 256 KB") - def test_raises_error_when_message_data_equal_to_256(self, client, mqtt_pipeline): - data_input = "a" * 262095 - message = Message(data_input) - # This check was put as message class may undergo the default content type encoding change - # and the above calculation will change. - if message.get_size() != device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT: - assert False - - client.send_message(message) + @pytest.mark.it("Validates message size at the 256 KB boundary") + @pytest.mark.parametrize( + "size_delta, raises_error", + [ + pytest.param(-1, False, id="Below limit"), + pytest.param(0, False, id="At limit"), + pytest.param(1, True, id="Above limit"), + ], + ) + def test_validates_message_size(self, client, mqtt_pipeline, size_delta, raises_error): + message_size = device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT + size_delta + message = Message(b"a" * message_size) - assert mqtt_pipeline.send_message.call_count == 1 - sent_message = mqtt_pipeline.send_message.call_args[0][0] - assert isinstance(sent_message, Message) - assert sent_message.data == data_input + if raises_error: + with pytest.raises(ValueError, match="256 KB"): + client.send_message(message) + assert mqtt_pipeline.send_message.call_count == 0 + else: + client.send_message(message) + assert mqtt_pipeline.send_message.call_count == 1 + assert mqtt_pipeline.send_message.call_args[0][0] is message class SharedClientReceiveMethodRequestTests(object): @@ -2303,10 +2285,10 @@ def test_raises_error_on_pipeline_op_error( [ pytest.param("message", id="String input"), pytest.param(222, id="Integer input"), - pytest.param(object(), id="Object input"), + pytest.param(1.5, id="Float input"), + pytest.param(b"message", id="Bytes input"), + pytest.param(bytearray(b"message"), id="Bytearray input"), pytest.param(None, id="None input"), - pytest.param([1, "str"], id="List input"), - pytest.param({"a": 2}, id="Dictionary input"), ], ) def test_send_message_to_output_calls_pipeline_wraps_data_in_message( @@ -2319,47 +2301,31 @@ def test_send_message_to_output_calls_pipeline_wraps_data_in_message( assert isinstance(sent_message, Message) assert sent_message.data == message_input - @pytest.mark.it("Raises error when message data size is greater than 256 KB") - def test_raises_error_when_message_to_output_data_greater_than_256(self, client, mqtt_pipeline): - output_name = "some_output" - data_input = "serpensortia" * 256000 - message = Message(data_input) - with pytest.raises(ValueError) as e_info: - client.send_message_to_output(message, output_name) - assert "256 KB" in e_info.value.args[0] - assert mqtt_pipeline.send_output_message.call_count == 0 - - @pytest.mark.it("Raises error when message size is greater than 256 KB") - def test_raises_error_when_message_to_output_size_greater_than_256(self, client, mqtt_pipeline): - output_name = "some_output" - data_input = "serpensortia" - message = Message(data_input) - message.custom_properties["spell"] = data_input * 256000 - with pytest.raises(ValueError) as e_info: - client.send_message_to_output(message, output_name) - assert "256 KB" in e_info.value.args[0] - assert mqtt_pipeline.send_output_message.call_count == 0 - - @pytest.mark.skipif( - sys.version_info >= (3, 12), - reason="Python 3.12 appears to have an issue. Investigate further.", + @pytest.mark.it("Validates message size at the 256 KB boundary") + @pytest.mark.parametrize( + "size_delta, raises_error", + [ + pytest.param(-1, False, id="Below limit"), + pytest.param(0, False, id="At limit"), + pytest.param(1, True, id="Above limit"), + ], ) - @pytest.mark.it("Does not raises error when message data size is equal to 256 KB") - def test_raises_error_when_message_to_output_data_equal_to_256(self, client, mqtt_pipeline): + def test_validates_message_size(self, client, mqtt_pipeline, size_delta, raises_error): output_name = "some_output" - data_input = "a" * 262095 - message = Message(data_input) - # This check was put as message class may undergo the default content type encoding change - # and the above calculation will change. - if message.get_size() != device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT: - assert False - - client.send_message_to_output(message, output_name) + message_size = device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT + size_delta + output_name_size = len(output_name.encode("utf-8")) + message = Message(b"a" * (message_size - output_name_size)) + + if raises_error: + with pytest.raises(ValueError, match="256 KB"): + client.send_message_to_output(message, output_name) + assert mqtt_pipeline.send_output_message.call_count == 0 + else: + client.send_message_to_output(message, output_name) + assert mqtt_pipeline.send_output_message.call_count == 1 + assert mqtt_pipeline.send_output_message.call_args[0][0] is message - assert mqtt_pipeline.send_output_message.call_count == 1 - sent_message = mqtt_pipeline.send_output_message.call_args[0][0] - assert isinstance(sent_message, Message) - assert sent_message.data == data_input + assert message.get_size() == message_size @pytest.mark.describe("IoTHubModuleClient (Synchronous) - .receive_message_on_input()")