diff --git a/test/py/test_python.py b/test/py/test_python.py index c81ad053..9578f4ef 100644 --- a/test/py/test_python.py +++ b/test/py/test_python.py @@ -1191,6 +1191,110 @@ def test_kafkaproducer_time_uritemplate_is_normalized_before_cloudevent_creation assert "from_binary(message, data_unmarshaller=lambda data: data)" in generated_test_src +@pytest.mark.parametrize("style", ["kafkaproducer", "amqpproducer"]) +def test_python_producer_inherited_avro_content_type(style): + """Inherited media types select the serializer, not the Avro schema format.""" + import base64 + import glob + import inspect + import io + import py_compile + import avro.io + from unittest.mock import Mock + + avro_type = "application/vnd.apache.org.avro+avro" + base = { + "messageid": "Example.Post", + "envelope": "CloudEvents/1.0", + "envelopemetadata": { + "type": {"value": "Example.Post"}, + "source": {"value": "urn:test"}, + "subject": {"value": "post"}, + "datacontenttype": {"value": "application/json"}, + }, + "dataschemaformat": "Avro/1.11.1", + "dataschemauri": "#/schemagroups/Example/schemas/Example.Post", + } + document = {"messagegroups": { + "Base": {"messages": {"Example.Post": base}}, + "Avro": {"messages": {"Example.Post": { + "basemessage": "/messagegroups/Base/messages/Example.Post", + "envelopemetadata": {"datacontenttype": {"value": avro_type}}, + }}}, + "Inherited": {"messages": {"Example.Post": { + "basemessage": "/messagegroups/Avro/messages/Example.Post", + }}}, + }} + document["schemagroups"] = {"Example": {"schemas": {"Example.Post": { + "format": "Avro/1.11.1", + "versions": {"1": { + "format": "Avro/1.11.1", + "schema": { + "type": "record", + "name": "Post", + "namespace": "Example", + "fields": [{"name": "text", "type": "string"}], + }, + }}, + }}}} + directory, producer_file = _generate_python_entrypoint_from_document( + document, style, "producer.py") + for path in glob.glob(os.path.join(directory, "**", "*.py"), recursive=True): + py_compile.compile(path, doraise=True) + if style == "amqpproducer": + import ast + with open(producer_file, encoding="utf-8") as handle: + tree = ast.parse(handle.read()) + defaults = {} + for node in tree.body: + if isinstance(node, ast.ClassDef): + for method in node.body: + if isinstance(method, ast.FunctionDef) and method.name == "send_post": + arguments = method.args.args[-len(method.args.defaults):] + defaults[node.name] = dict(zip( + [arg.arg for arg in arguments], method.args.defaults))["content_type"].value + assert defaults["BaseProducer"] == "application/json" + assert defaults["AvroProducer"] == avro_type + assert defaults["InheritedProducer"] == avro_type + return + + paths = glob.glob(os.path.join(directory, "**", "src"), recursive=True) + module = _load_generated_python_module_from_path(producer_file, extra_paths=paths) + data = module.Post(text="hello") + for group, default_type in [("Base", "application/json"), ("Avro", avro_type), + ("Inherited", avro_type)]: + producer_class = getattr(module, f"{group}EventProducer") + assert inspect.signature(producer_class.send_example_post).parameters[ + "content_type"].default == default_type + for mode in ("binary", "structured"): + for override in ( + None, "application/json", avro_type, "avro/binary", + "application/vnd.apache.avro+avro", avro_type + "; charset=binary", + ): + recording = Mock() + producer = producer_class(recording, "posts", content_mode=mode) + kwargs = {"content_type": override} if override else {} + producer.send_example_post(data=data, flush_producer=False, **kwargs) + sent = recording.produce.call_args.kwargs + content_type = override or default_type + if mode == "binary": + assert sent["headers"]["content-type"] == content_type.encode("utf-8") + payload = sent["value"] + else: + assert sent["headers"]["content-type"] == b"application/cloudevents+json" + event = json.loads(sent["value"]) + assert event["datacontenttype"] == content_type + payload = base64.b64decode(event["data_base64"]) + assert "data" not in event + if content_type == "application/json": + assert json.loads(payload)["text"] == "hello" + else: + stream = io.BytesIO(payload) + assert avro.io.DatumReader(data.AvroType).read( + avro.io.BinaryDecoder(stream)) == {"text": "hello"} + assert stream.tell() == len(payload) + + def test_kafkaproducer_exposes_resilient_flush_helper(): """Regression for #430: the generated Kafka producer must expose a ``flush(timeout, retries, backoff_factor)`` helper that retries with @@ -1574,7 +1678,7 @@ def test_mqttclient_body_is_bytes_not_str(): have to double-decode JSON. """ src = _generate_mqtt_client_src() - assert "byte_data = data.to_byte_array(content_type)" in src, ( + assert "byte_data = data.to_byte_array(content_type.replace(" in src, ( "client.py must call to_byte_array on the dataclass" ) # The coercion guard must be immediately downstream. @@ -1792,7 +1896,7 @@ def test_ehproducer_body_is_bytes_not_str(): data section, not a string section containing escaped JSON. """ src = _generate_eh_producer_src("test/xreg/lightbulb.xreg.json") - assert "byte_data = data.to_byte_array(content_type)" in src + assert "byte_data = data.to_byte_array(content_type.replace(" in src assert "isinstance(byte_data, str)" in src and "byte_data.encode('utf-8')" in src, ( "ehproducer must coerce to_byte_array str result to bytes before " "handing to the CloudEvent constructor" diff --git a/xrcg/templates/py/_common/cloudevents.jinja.include b/xrcg/templates/py/_common/cloudevents.jinja.include index 09cf081c..4a32f147 100644 --- a/xrcg/templates/py/_common/cloudevents.jinja.include +++ b/xrcg/templates/py/_common/cloudevents.jinja.include @@ -14,7 +14,11 @@ {%- macro DefaultContentType(message) -%} {%- set schema_format = message.dataschemaformat | default("", true) | lower -%} {%- if schema_format.startswith("avro") -%} +{%- if message.envelopemetadata is defined and message.envelopemetadata and "datacontenttype" in message.envelopemetadata and message.envelopemetadata.datacontenttype.value -%} +'{{ message.envelopemetadata.datacontenttype.value }}' +{%- else -%} 'avro/binary' +{%- endif -%} {%- elif schema_format.startswith("proto") -%} 'application/protobuf' {%- elif schema_format.startswith("xsd") or schema_format.startswith("xmlschema") -%} @@ -32,6 +36,11 @@ {%- endif -%} {%- endmacro -%} +{# Normalize the Avro media-type alias for Avrotize, not for wire metadata. #} +{%- macro SerializerContentType(content_type) -%} +{{ content_type }}.replace("application/vnd.apache.org.avro+avro", "application/vnd.apache.avro+avro", 1) +{%- endmacro -%} + {%- macro DeclareRfc3339TimeNormalizer() -%} _RFC3339_TIMESTAMP_PATTERN = re.compile( r'^\d{4}-\d{2}-\d{2}[Tt]\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:[Zz]|[+-]\d{2}:\d{2})?$' diff --git a/xrcg/templates/py/amqpproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja b/xrcg/templates/py/amqpproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja index fdbc91f7..c948a3c0 100644 --- a/xrcg/templates/py/amqpproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja +++ b/xrcg/templates/py/amqpproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja @@ -664,6 +664,7 @@ class {{ class_name }}: if data is None: return b'' if hasattr(data, 'to_byte_array'): + content_type = {{ cloudEvents.SerializerContentType("content_type") }} payload = data.to_byte_array(content_type) elif hasattr(data, 'to_dict'): payload = json.dumps(data.to_dict()) diff --git a/xrcg/templates/py/ehproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja b/xrcg/templates/py/ehproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja index 345cad98..9c54005d 100644 --- a/xrcg/templates/py/ehproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja +++ b/xrcg/templates/py/ehproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja @@ -131,7 +131,7 @@ class {{ class_name }}: } attributes["datacontenttype"] = content_type attributes["time"] = _resolve_cloudevents_time(_time, attributes.get("time")) - byte_data = data.to_byte_array(content_type) if data is not None else b'' + byte_data = data.to_byte_array({{ cloudEvents.SerializerContentType("content_type") }}) if data is not None else b'' # to_byte_array returns str for text content types (e.g. JSON); # coerce to bytes so the cloudevents-sdk emits a binary AMQP data # section rather than a string section containing escaped JSON. diff --git a/xrcg/templates/py/kafkaproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja b/xrcg/templates/py/kafkaproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja index a2570367..ea4500ed 100644 --- a/xrcg/templates/py/kafkaproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja +++ b/xrcg/templates/py/kafkaproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja @@ -229,7 +229,7 @@ class {{ class_name }}: attributes["datacontenttype"] = content_type attributes["time"] = _resolve_cloudevents_time(_time, attributes.get("time")) {% if data_type != "object" -%} - byte_data = data.to_byte_array(content_type) if data is not None else b'' + byte_data = data.to_byte_array({{ cloudEvents.SerializerContentType("content_type") }}) if data is not None else b'' {%- else -%} byte_data = json.dumps(data).encode('utf-8') if data is not None else b'' {%- endif %} @@ -237,7 +237,12 @@ class {{ class_name }}: byte_data = byte_data.encode('utf-8') event = CloudEvent.create(attributes, byte_data) if self.content_mode == "structured": - message = to_structured(event, data_marshaller=lambda x: self.__binary_data_marshaller(x), key_mapper=lambda x: self.__key_mapper(x, data, key_mapper, kafka_key)) + message = to_structured( + event, + data_marshaller=lambda x: self.__binary_data_marshaller(x), + envelope_marshaller=lambda envelope: json.dumps({**envelope, "datacontenttype": content_type}), + key_mapper=lambda x: self.__key_mapper(x, data, key_mapper, kafka_key), + ) message.headers["content-type"] = b"application/cloudevents+json" else: # For binary mode, datacontenttype is already set in attributes above diff --git a/xrcg/templates/py/mqttclient/src/{mainprojectdir!dotunderscore!lower}client.py.jinja b/xrcg/templates/py/mqttclient/src/{mainprojectdir!dotunderscore!lower}client.py.jinja index 7c57e7c7..8bbbb7fb 100644 --- a/xrcg/templates/py/mqttclient/src/{mainprojectdir!dotunderscore!lower}client.py.jinja +++ b/xrcg/templates/py/mqttclient/src/{mainprojectdir!dotunderscore!lower}client.py.jinja @@ -823,7 +823,7 @@ class {{ class_name }}(_ClientBase): } attributes["datacontenttype"] = content_type attributes["time"] = _resolve_cloudevents_time(_time, attributes.get("time")) - byte_data = data.to_byte_array(content_type) if data is not None else b'' + byte_data = data.to_byte_array({{ cloudEvents.SerializerContentType("content_type") }}) if data is not None else b'' # to_byte_array returns str for text content types (e.g. JSON); # paho-mqtt will UTF-8 encode str payloads, but cloudevents-sdk's # to_binary/to_structured embed the str directly which then becomes @@ -953,7 +953,7 @@ class {{ class_name }}(_ClientBase): else: payload = json.dumps(data).encode('utf-8') {%- else %} - payload = data.to_byte_array(content_type) if data is not None else b'' + payload = data.to_byte_array({{ cloudEvents.SerializerContentType("content_type") }}) if data is not None else b'' if isinstance(payload, str): payload = payload.encode('utf-8') {%- endif %} diff --git a/xrcg/templates/py/sbproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja b/xrcg/templates/py/sbproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja index 3487f379..4769f8c9 100644 --- a/xrcg/templates/py/sbproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja +++ b/xrcg/templates/py/sbproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja @@ -150,7 +150,7 @@ class {{ class_name }}: } attributes["datacontenttype"] = content_type attributes["time"] = _resolve_cloudevents_time(_time, attributes.get("time")) - byte_data = data.to_byte_array(content_type) if data is not None else b'' + byte_data = data.to_byte_array({{ cloudEvents.SerializerContentType("content_type") }}) if data is not None else b'' # to_byte_array returns str for text content types (e.g. JSON); coerce # to bytes so the wire body is the exact serialized representation # (binary AMQP data section) and receivers do not double-decode JSON.