hurui200320
45357c42a8
Unit test coverage / pytest (push) Failing after 3h11m8s
This commit write unit tests for pydantic serializer using cursor ai, and found an issue related to serialization. Originally, the code uses comma to concat string, but map and list also has comma inside. During the deserialization, split using comma will break map and list, leaving `{` with first kv pair for map, and `[` with first element with list. Instead of using concat with comma as separator, I use json to encode the object dict.
133 lines
4.4 KiB
Python
133 lines
4.4 KiB
Python
from datetime import datetime
|
|
from unittest import TestCase
|
|
from uuid import UUID
|
|
|
|
from pydantic import BaseModel
|
|
|
|
from amqp.adapter.pydantic_serializer import (
|
|
deserialize_object,
|
|
parse_url_query,
|
|
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, {})
|