diff --git a/src/conductor/client/workflow/task/kafka_publish_input.py b/src/conductor/client/workflow/task/kafka_publish_input.py index 13f5ffde8..1158a4369 100644 --- a/src/conductor/client/workflow/task/kafka_publish_input.py +++ b/src/conductor/client/workflow/task/kafka_publish_input.py @@ -5,15 +5,39 @@ class KafkaPublishInput: - def __init__(self, - bootstrap_servers: str = None, - key: str = None, - key_serializer: str = None, - value: str = None, - request_timeout_ms: str = None, - max_block_ms: str = None, - headers: Dict[str, Any] = None, - topic: str = None) -> Self: + swagger_types = { + "_bootstrap_servers": "str", + "_key": "str", + "_key_serializer": "str", + "_value": "str", + "_request_timeout_ms": "str", + "_max_block_ms": "str", + "_headers": "dict[str, Any]", + "_topic": "str", + } + + attribute_map = { + "_bootstrap_servers": "bootStrapServers", + "_key": "key", + "_key_serializer": "keySerializer", + "_value": "value", + "_request_timeout_ms": "requestTimeoutMs", + "_max_block_ms": "maxBlockMs", + "_headers": "headers", + "_topic": "topic", + } + + def __init__( + self, + bootstrap_servers: str = None, + key: str = None, + key_serializer: str = None, + value: str = None, + request_timeout_ms: str = None, + max_block_ms: str = None, + headers: Dict[str, Any] = None, + topic: str = None, + ) -> Self: self._bootstrap_servers = deepcopy(bootstrap_servers) self._key = deepcopy(key) self._key_serializer = deepcopy(key_serializer) diff --git a/tests/unit/workflow/__init__.py b/tests/unit/workflow/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/unit/workflow/test_kafka_publish_input.py b/tests/unit/workflow/test_kafka_publish_input.py new file mode 100644 index 000000000..fac374137 --- /dev/null +++ b/tests/unit/workflow/test_kafka_publish_input.py @@ -0,0 +1,145 @@ +import unittest + +from conductor.client.http.api_client import ApiClient +from conductor.client.workflow.task.kafka_publish_input import \ + KafkaPublishInput + + +class TestKafkaPublishInput(unittest.TestCase): + """Integration tests for KafkaPublishInput with API client serialization.""" + + def setUp(self): + """Set up test fixtures.""" + self.api_client = ApiClient() + self.sample_kafka_input = KafkaPublishInput( + bootstrap_servers="kafka-broker:29092", + key="test-key", + key_serializer="org.apache.kafka.common.serialization.StringSerializer", + value='{"test": "data"}', + request_timeout_ms="30000", + max_block_ms="60000", + headers={"content-type": "application/json"}, + topic="test-topic", + ) + + def test_kafka_publish_input_serialization_structure(self): + """Test that serialized KafkaPublishInput has the correct structure.""" + serialized = self.api_client.sanitize_for_serialization(self.sample_kafka_input) + + expected_keys = [ + "bootStrapServers", + "key", + "keySerializer", + "value", + "requestTimeoutMs", + "maxBlockMs", + "headers", + "topic", + ] + + for key in expected_keys: + self.assertIn(key, serialized, f"Missing key '{key}' in serialized output") + + self.assertEqual(serialized["bootStrapServers"], "kafka-broker:29092") + self.assertEqual(serialized["key"], "test-key") + self.assertEqual( + serialized["keySerializer"], + "org.apache.kafka.common.serialization.StringSerializer", + ) + self.assertEqual(serialized["value"], '{"test": "data"}') + self.assertEqual(serialized["requestTimeoutMs"], "30000") + self.assertEqual(serialized["maxBlockMs"], "60000") + self.assertEqual(serialized["headers"], {"content-type": "application/json"}) + self.assertEqual(serialized["topic"], "test-topic") + + def test_kafka_publish_input_with_none_values_serialization(self): + """Test that KafkaPublishInput with None values serializes correctly.""" + kafka_input = KafkaPublishInput( + bootstrap_servers="kafka:9092", topic="test-topic" + ) + + serialized = self.api_client.sanitize_for_serialization(kafka_input) + + self.assertEqual(serialized["bootStrapServers"], "kafka:9092") + self.assertEqual(serialized["topic"], "test-topic") + + self.assertNotIn("key", serialized) + self.assertNotIn("keySerializer", serialized) + self.assertNotIn("value", serialized) + self.assertNotIn("requestTimeoutMs", serialized) + self.assertNotIn("maxBlockMs", serialized) + self.assertNotIn("headers", serialized) + + def test_kafka_publish_input_complex_headers_serialization(self): + """Test that KafkaPublishInput with complex headers serializes correctly.""" + complex_headers = { + "content-type": "application/json", + "correlation-id": "test-123", + "user-agent": "conductor-python-sdk", + "custom-header": "custom-value", + } + + kafka_input = KafkaPublishInput( + bootstrap_servers="kafka:9092", + headers=complex_headers, + topic="complex-topic", + value='{"complex": "data"}', + ) + + serialized = self.api_client.sanitize_for_serialization(kafka_input) + + self.assertEqual(serialized["headers"], complex_headers) + self.assertEqual(serialized["bootStrapServers"], "kafka:9092") + self.assertEqual(serialized["topic"], "complex-topic") + self.assertEqual(serialized["value"], '{"complex": "data"}') + + def test_kafka_publish_input_swagger_types_consistency(self): + """Test that swagger_types are consistent with actual serialization.""" + swagger_types = KafkaPublishInput.swagger_types + + kafka_input = KafkaPublishInput( + bootstrap_servers="test", + key="test", + key_serializer="test", + value="test", + request_timeout_ms="test", + max_block_ms="test", + headers={"test": "test"}, + topic="test", + ) + + serialized = self.api_client.sanitize_for_serialization(kafka_input) + + for internal_attr, expected_type in swagger_types.items(): + external_attr = KafkaPublishInput.attribute_map[internal_attr] + self.assertIn( + external_attr, + serialized, + f"Swagger type '{internal_attr}' not found in serialized output", + ) + + def test_kafka_publish_input_attribute_map_consistency(self): + """Test that attribute_map correctly maps all internal attributes.""" + kafka_input = self.sample_kafka_input + internal_attrs = [ + attr + for attr in dir(kafka_input) + if attr.startswith("_") and not attr.startswith("__") + ] + + for attr in internal_attrs: + self.assertIn( + attr, + KafkaPublishInput.attribute_map, + f"Internal attribute '{attr}' not found in attribute_map", + ) + + for internal_attr in KafkaPublishInput.attribute_map.keys(): + self.assertTrue( + hasattr(kafka_input, internal_attr), + f"Attribute_map key '{internal_attr}' not found in instance", + ) + + +if __name__ == "__main__": + unittest.main()