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 @vivjd, thx for opening the PR.

Unfortunately your change is not really what the desired behavior should be. I've mentioned the desired one here: https://github.com/cloudevents/sdk-python/issues/211#issuecomment-1533212334

So if you're up for changing this to comply with the desired behavior I'd be glad to review the PR.
vivjd: Hi @xSAVIKx! Thank you for you feedback. I've reread the original issue and the desired behavior, and changed it accordingly in my PR. If you could review the PR that would be great! Thank you for your time!
vivjd: Hi @xSAVIKx, I changed the test cases to match the change. Could you please kindly review this again? Thank you in advance!
Cali0707: Hey @xSAVIKx @duglin would one of you be able to approve the workflows on this PR?
xSAVIKx: Sorry, my bad. I was.pretty sure I did that when I was reviewing the PR

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()
         )