import json from datetime import date, datetime from unittest import TestCase from uuid import UUID from pydantic import BaseModel from amqp.adapter.pydantic_serializer import ( deserialize_object, parse_url_query, python_type_to_json, serialize_object, ) class NestedModel(BaseModel): """ Nested model for testing the pydantic serializer. """ value: str number: int class TestModel(BaseModel): """ Test model for testing the pydantic serializer. """ name: str age: int uuid_field: UUID datetime_field: datetime nested_model: NestedModel | None = None list_field: list = [] class PydanticSerializerTest(TestCase): def setUp(self): """ Prepare the test data. """ self.test_uuid = UUID("12345678-1234-5678-1234-567812345678") self.test_datetime = datetime(2024, 1, 1, 12, 0, 0) self.nested_model = NestedModel(value="test", number=42) self.test_model = TestModel( name="test", age=25, uuid_field=self.test_uuid, datetime_field=self.test_datetime, nested_model=self.nested_model, list_field=[1, 2, 3], ) def test_serialize_object(self): serialized = serialize_object(self.test_model) self.assertIsInstance(serialized, str) self.assertIn("TestModel", serialized) self.assertIn("test", serialized) self.assertIn("25", serialized) self.assertIn(str(self.test_uuid), serialized) self.assertIn(self.test_datetime.isoformat(), serialized) def test_deserialize_object(self): serialized = serialize_object(self.test_model) deserialized = deserialize_object(serialized) self.assertIsInstance(deserialized, TestModel) self.assertEqual(deserialized.name, "test") self.assertEqual(deserialized.age, 25) self.assertEqual(deserialized.uuid_field, self.test_uuid) self.assertEqual(deserialized.datetime_field, self.test_datetime) self.assertIsInstance(deserialized.nested_model, NestedModel) self.assertEqual(deserialized.nested_model.value, "test") self.assertEqual(deserialized.nested_model.number, 42) self.assertEqual(deserialized.list_field, [1, 2, 3]) def test_serialize_deserialize_with_none(self): model = TestModel( name="test", age=25, uuid_field=self.test_uuid, datetime_field=self.test_datetime, nested_model=None, list_field=[], ) serialized = serialize_object(model) deserialized = deserialize_object(serialized) self.assertIsNone(deserialized.nested_model) def test_serialize_deserialize_with_list_of_models(self): nested_models = [NestedModel(value=f"test{i}", number=i) for i in range(3)] model = TestModel( name="test", age=25, uuid_field=self.test_uuid, datetime_field=self.test_datetime, nested_model=None, list_field=nested_models, ) serialized = serialize_object(model) deserialized = deserialize_object(serialized) self.assertEqual(len(deserialized.list_field), 3) for i, item in enumerate(deserialized.list_field): self.assertIsInstance(item, NestedModel) self.assertEqual(item.value, f"test{i}") self.assertEqual(item.number, i) def test_parse_url_query_no_params(self): test_url = "http://example.com/path" path, params = parse_url_query(test_url) self.assertEqual(path, "/path") self.assertEqual(params, {}) def test_parse_url_query_single_value(self): test_url = "http://example.com/path?foo=1" path, params = parse_url_query(test_url) self.assertEqual(path, "/path") self.assertEqual(params["foo"], "1") def test_parse_url_query(self): path, args = parse_url_query( "http://localhost:8080/api/v0/unstructured/863e4d9f-4612-4173-8d0a-3de34cc7a741?response_type=JsonFile" ) self.assertEqual(path, "/api/v0/unstructured/863e4d9f-4612-4173-8d0a-3de34cc7a741") self.assertEqual(args, {"response_type": "JsonFile"}) path, args = parse_url_query( "http://localhost:8080/api/v0/unstructured/863e4d9f-4612-4173-8d0a-3de34cc7a741" ) self.assertEqual(path, "/api/v0/unstructured/863e4d9f-4612-4173-8d0a-3de34cc7a741") self.assertEqual(args, {}) class TestPythonTypeToJson: """ Test cases for the python_type_to_json function. """ def test_basic_types(self): """Test serialization of basic Python types.""" # Test dictionary data = {"key": "value", "number": 42} result = python_type_to_json(data) assert json.loads(result) == data # Test list data = [1, 2, 3, "four"] result = python_type_to_json(data) assert json.loads(result) == data # Test tuple data = (1, 2, 3) result = python_type_to_json(data) assert json.loads(result) == [1, 2, 3] # Tuples are converted to lists # Test set data = {1, 2, 3} result = python_type_to_json(data) assert set(json.loads(result)) == data # Sets are converted to lists def test_datetime_handling(self): """Test serialization of datetime objects.""" # Test datetime test_datetime = datetime(2024, 1, 1, 12, 0, 0) data = {"timestamp": test_datetime} result = python_type_to_json(data) assert json.loads(result) == {"timestamp": test_datetime.isoformat()} # Test date test_date = date(2024, 1, 1) data = {"date": test_date} result = python_type_to_json(data) assert json.loads(result) == {"date": test_date.isoformat()} def test_nested_structures(self): """Test serialization of nested data structures.""" data = { "list_of_datetimes": [datetime(2024, 1, 1), datetime(2024, 1, 2)], "dict_with_date": {"date": date(2024, 1, 1)}, "mixed": {"set": {1, 2, 3}, "tuple": (1, 2, 3), "datetime": datetime(2024, 1, 1)}, } result = python_type_to_json(data) parsed = json.loads(result) assert parsed["list_of_datetimes"] == [d.isoformat() for d in data["list_of_datetimes"]] assert parsed["dict_with_date"]["date"] == data["dict_with_date"]["date"].isoformat() assert set(parsed["mixed"]["set"]) == data["mixed"]["set"] assert parsed["mixed"]["tuple"] == list(data["mixed"]["tuple"]) assert parsed["mixed"]["datetime"] == data["mixed"]["datetime"].isoformat() def test_custom_object_serialization(self): """Test serialization of custom objects with __dict__ attribute.""" class TestObject: def __init__(self, value): self.value = value self.timestamp = datetime(2024, 1, 1) obj = TestObject("test") data = {"obj": obj} result = python_type_to_json(data) parsed = json.loads(result) assert parsed["obj"]["value"] == "test" assert parsed["obj"]["timestamp"] == obj.timestamp.isoformat() def test_empty_structures(self): """Test serialization of empty data structures.""" # Empty dict assert python_type_to_json({}) == "{}" # Empty list assert python_type_to_json([]) == "[]" # Empty set assert python_type_to_json(set()) == "[]" # Empty tuple assert python_type_to_json(()) == "[]" def test_unicode_handling(self): """Test handling of Unicode characters.""" data = {"text": "Hello, 世界!"} result = python_type_to_json(data) assert json.loads(result) == data # ensure_ascii=False should preserve Unicode