Repository: cloudevents/sdk-python

Issue:
`cloudevents.kafka.to_binary` checks for a `content-type` attribute, which would not be a valid attribute name
https://github.com/cloudevents/sdk-python/blob/ef982743b68866abbe0049dbffac76f5a2e3efb4/cloudevents/kafka/conversion.py#L90

This will call `CloudEvent._get_attributes()["content-type"]` but a `content-type` attribute will violate the attribute naming rules:

https://github.com/cloudevents/spec/blob/30f1a081743bfe06b45b85519dd145f14a1ad22c/cloudevents/spec.md?plain=1#L173-L175

> CloudEvents attribute names MUST consist of lower-case letters ('a' to 'z') or digits ('0' to '9') from the ASCII character set. Attribute names SHOULD be descriptive and terse and SHOULD NOT exceed 20 characters in length.

It is not really that clear to me why this is being done, but it also causes some problems when `event` is a Pydantic cloud event, but regardless of that it seems pretty clear that it is wrong.

Discussion:
@xSAVIKx: Hey @aucampia, thx for raising this. I believe you're correct and the event itself must not contain the `content-type` attribute. The conversion here should convert the `datacontenttype` of the event to the `content-type` header of the Kafka message.

Ans similarly the backwards conversion must take care of the respective mapping.

Here's the supportive docs section: https://github.com/cloudevents/spec/blob/30f1a081743bfe06b45b85519dd145f14a1ad22c/cloudevents/bindings/kafka-protocol-binding.md#321-content-type@duglin: someone want to PR this?@aucampia: I want to, but realistically I probably won't get to it before the end of next week, though I may look at it then. Will notify before I start working on it though.@xSAVIKx: I'm out of capacity till the end of next week as well. Even though it's a small change in the codebase, I'd appreciate a PR. If none will start working on it, I'll find some time next week.@MaryamTaj: Hi! I would be happy to work on this!


Solution patch:
diff --git a/cloudevents/kafka/conversion.py b/cloudevents/kafka/conversion.py
index 832594d..97c355f 100644
--- a/cloudevents/kafka/conversion.py
+++ b/cloudevents/kafka/conversion.py
@@ -87,10 +87,10 @@ def to_binary(
         )
 
     headers = {}
-    if event["content-type"]:
-        headers["content-type"] = event["content-type"].encode("utf-8")
+    if event["datacontenttype"]:
+        headers["content-type"] = event["datacontenttype"].encode("utf-8")
     for attr, value in event.get_attributes().items():
-        if attr not in ["data", "partitionkey", "content-type"]:
+        if attr not in ["data", "partitionkey", "datacontenttype"]:
             if value is not None:
                 headers["ce_{0}".format(attr)] = value.encode("utf-8")
 
@@ -126,7 +126,7 @@ def from_binary(
     for header, value in message.headers.items():
         header = header.lower()
         if header == "content-type":
-            attributes["content-type"] = value.decode()
+            attributes["datacontenttype"] = value.decode()
         elif header.startswith("ce_"):
             attributes[header[3:]] = value.decode()
 
@@ -189,8 +189,8 @@ def to_structured(
         attrs["data"] = data
 
     headers = {}
-    if "content-type" in attrs:
-        headers["content-type"] = attrs.pop("content-type").encode("utf-8")
+    if "datacontenttype" in attrs:
+        headers["content-type"] = attrs.pop("datacontenttype").encode("utf-8")
 
     try:
         value = envelope_marshaller(attrs)
@@ -255,7 +255,10 @@ def from_structured(
             attributes[name] = decoded_value
 
     for header, val in message.headers.items():
-        attributes[header.lower()] = val.decode()
+        if header.lower() == "content-type":
+            attributes["datacontenttype"] = val.decode()
+        else:
+            attributes[header.lower()] = val.decode()
     if event_type:
         result = event_type.create(attributes, data)
     else:


Test patch:
diff --git a/cloudevents/tests/test_kafka_conversions.py b/cloudevents/tests/test_kafka_conversions.py
index 696e75c..5580773 100644
--- a/cloudevents/tests/test_kafka_conversions.py
+++ b/cloudevents/tests/test_kafka_conversions.py
@@ -59,7 +59,7 @@ class KafkaConversionTestBase:
                 "source": "pytest",
                 "type": "com.pytest.test",
                 "time": datetime.datetime(2000, 1, 1, 6, 42, 33).isoformat(),
-                "content-type": "foo",
+                "datacontenttype": "foo",
                 "partitionkey": "test_key_123",
             },
             data=self.expected_data,
@@ -123,7 +123,7 @@ class TestToBinary(KafkaConversionTestBase):
         assert result.headers["ce_source"] == source_event["source"].encode("utf-8")
         assert result.headers["ce_type"] == source_event["type"].encode("utf-8")
         assert result.headers["ce_time"] == source_event["time"].encode("utf-8")
-        assert result.headers["content-type"] == source_event["content-type"].encode(
+        assert result.headers["content-type"] == source_event["datacontenttype"].encode(
             "utf-8"
         )
         assert "data" not in result.headers
@@ -163,7 +163,7 @@ class TestFromBinary(KafkaConversionTestBase):
                 "ce_time": datetime.datetime(2000, 1, 1, 6, 42, 33)
                 .isoformat()
                 .encode("utf-8"),
-                "content-type": "foo".encode("utf-8"),
+                "datacontenttype": "foo".encode("utf-8"),
             },
             value=simple_serialize(self.expected_data),
             key="test_key_123",
@@ -205,7 +205,7 @@ class TestFromBinary(KafkaConversionTestBase):
         assert result["type"] == source_binary_json_message.headers["ce_type"].decode()
         assert result["time"] == source_binary_json_message.headers["ce_time"].decode()
         assert (
-            result["content-type"]
+            result["datacontenttype"]
             == source_binary_json_message.headers["content-type"].decode()
         )
 
@@ -328,7 +328,7 @@ class TestToStructured(KafkaConversionTestBase):
     def test_sets_headers(self, source_event):
         result = to_structured(source_event)
         assert len(result.headers) == 1
-        assert result.headers["content-type"] == source_event["content-type"].encode(
+        assert result.headers["content-type"] == source_event["datacontenttype"].encode(
             "utf-8"
         )
 
@@ -474,7 +474,7 @@ class TestFromStructured(KafkaConversionTestBase):
     ):
         result = from_structured(source_structured_json_message)
         assert (
-            result["content-type"]
+            result["datacontenttype"]
             == source_structured_json_message.headers["content-type"].decode()
         )
 
@@ -487,7 +487,7 @@ class TestFromStructured(KafkaConversionTestBase):
             envelope_unmarshaller=custom_unmarshaller,
         )
         assert (
-            result["content-type"]
+            result["datacontenttype"]
             == source_structured_bytes_bytes_message.headers["content-type"].decode()
         )