Stanislav Hejny
227 lines
7.8 KiB
Python
227 lines
7.8 KiB
Python
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
|