Refactoring tests

Author: leandrodamascena

aws_lambda_powertools/utilities/data_classes/kafka_event.py

@@ -10,6 +10,18 @@
     from collections.abc import Iterator
 
 
+class ConsumerRecordSchemaMetadata(DictWrapper):
+    @property
+    def data_format(self) -> str | None:
+        """The data format of the Kafka record."""
+        return self.get("dataFormat", None)
+
+    @property
+    def schema_id(self) -> str | None:
+        """The schema id of the Kafka record."""
+        return self.get("schemaId", None)
+
+
 class KafkaEventRecordBase(DictWrapper):
     @property
     def topic(self) -> str:
@@ -36,6 +48,22 @@ def timestamp_type(self) -> str:
         """The Kafka record timestamp type."""
         return self["timestampType"]
 
+    @property
+    def key_schema_metadata(self) -> ConsumerRecordSchemaMetadata | None:
+        """The metadata of the Key Kafka record."""
+        return (
+            None if self.get("keySchemaMetadata") is None else ConsumerRecordSchemaMetadata(self["keySchemaMetadata"])
+        )
+
+    @property
+    def value_schema_metadata(self) -> ConsumerRecordSchemaMetadata | None:
+        """The metadata of the Value Kafka record."""
+        return (
+            None
+            if self.get("valueSchemaMetadata") is None
+            else ConsumerRecordSchemaMetadata(self["valueSchemaMetadata"])
+        )
+
 
 class KafkaEventRecord(KafkaEventRecordBase):
     @property

aws_lambda_powertools/utilities/kafka/consumer_records.py

@@ -3,7 +3,7 @@
 from functools import cached_property
 from typing import TYPE_CHECKING, Any
 
-from aws_lambda_powertools.utilities.data_classes.common import CaseInsensitiveDict, DictWrapper
+from aws_lambda_powertools.utilities.data_classes.common import CaseInsensitiveDict
 from aws_lambda_powertools.utilities.data_classes.kafka_event import KafkaEventBase, KafkaEventRecordBase
 from aws_lambda_powertools.utilities.kafka.deserializer.deserializer import get_deserializer
 from aws_lambda_powertools.utilities.kafka.serialization.serialization import serialize_to_output_type
@@ -14,18 +14,6 @@
     from aws_lambda_powertools.utilities.kafka.schema_config import SchemaConfig
 
 
-class ConsumerRecordSchemaMetadata(DictWrapper):
-    @property
-    def data_format(self) -> str | None:
-        """The data format of the Kafka record."""
-        return self.get("dataFormat", None)
-
-    @property
-    def schema_id(self) -> str | None:
-        """The schema id of the Kafka record."""
-        return self.get("schemaId", None)
-
-
 class ConsumerRecordRecords(KafkaEventRecordBase):
     """
     A Kafka Consumer Record
@@ -114,22 +102,6 @@ def headers(self) -> dict[str, bytes]:
         """Decodes the headers as a single dictionary."""
         return CaseInsensitiveDict((k, bytes(v)) for chunk in self.original_headers for k, v in chunk.items())
 
-    @property
-    def key_schema_metadata(self) -> ConsumerRecordSchemaMetadata | None:
-        """The metadata of the Key Kafka record."""
-        return (
-            None if self.get("keySchemaMetadata") is None else ConsumerRecordSchemaMetadata(self["keySchemaMetadata"])
-        )
-
-    @property
-    def value_schema_metadata(self) -> ConsumerRecordSchemaMetadata | None:
-        """The metadata of the Value Kafka record."""
-        return (
-            None
-            if self.get("valueSchemaMetadata") is None
-            else ConsumerRecordSchemaMetadata(self["valueSchemaMetadata"])
-        )
-
 
 class ConsumerRecords(KafkaEventBase):
     """Self-managed or MSK Apache Kafka event trigger

aws_lambda_powertools/utilities/kafka/serialization/dataclass.py

@@ -12,9 +12,6 @@
 
 class DataclassOutputSerializer(OutputSerializerBase):
     def serialize(self, data: dict[str, Any], output: type[T] | Callable | None = None) -> T | dict[str, Any]:
-        if output is None:
-            return data
-
         if not is_dataclass(output):
             raise ValueError("Output class must be a dataclass")
 

aws_lambda_powertools/utilities/kafka/serialization/pydantic.py

@@ -14,9 +14,6 @@
 
 class PydanticOutputSerializer(OutputSerializerBase):
     def serialize(self, data: dict[str, Any], output: type[T] | Callable | None = None) -> T | dict[str, Any]:
-        if output is None:
-            return data
-
         # Use TypeAdapter for better support of Union types and other complex types
         adapter: TypeAdapter = TypeAdapter(output)
         return adapter.validate_python(data)

tests/functional/kafka_consumer/required_dependencies/test_kafka_consumer.py

@@ -272,3 +272,22 @@ def handler(event: ConsumerRecords, context):
 
     # Verify the results
     assert result is None
+
+
+def test_kafka_consumer_metadata_fields(kafka_event_with_json_data, lambda_context):
+    """Test Kafka consumer when no schema config is provided."""
+
+    kafka_event_with_json_data["records"]["my-topic-1"][0]["key"] = None
+
+    @kafka_consumer()
+    def handler(event: ConsumerRecords, context):
+        return event.record
+
+    # Call the handler
+    result = handler(kafka_event_with_json_data, lambda_context)
+
+    # Verify the results
+    assert result.original_value == kafka_event_with_json_data["records"]["my-topic-1"][0]["value"]
+    assert result.original_key == kafka_event_with_json_data["records"]["my-topic-1"][0]["key"]
+    assert result.original_headers == kafka_event_with_json_data["records"]["my-topic-1"][0]["headers"]
+    assert result.headers