mirror of
https://github.com/microsoft/autogen.git
synced 2025-07-07 00:51:38 +00:00
e1a823fb6d
* initial impl of topics and subscriptions * Update python/src/agnext/core/_agent_runtime.py Co-authored-by: Eric Zhu <ekzhu@users.noreply.github.com> * add topic in context * migrate * migrate code for topics * migrate team one * edit notebooks * formatting * fix imports * Build proto * Fix circular import --------- Co-authored-by: Eric Zhu <ekzhu@users.noreply.github.com>
106 lines
3.2 KiB
Python
106 lines
3.2 KiB
Python
#custom type
|
|
|
|
from pydantic import BaseModel
|
|
from dataclasses import dataclass
|
|
|
|
import pytest
|
|
|
|
from agnext.core import Serialization
|
|
|
|
class PydanticMessage(BaseModel):
|
|
message: str
|
|
|
|
class NestingPydanticMessage(BaseModel):
|
|
message: str
|
|
nested: PydanticMessage
|
|
|
|
@dataclass
|
|
class DataclassMessage:
|
|
message: str
|
|
|
|
@dataclass
|
|
class NestingDataclassMessage:
|
|
message: str
|
|
nested: DataclassMessage
|
|
|
|
@dataclass
|
|
class NestingPydanticDataclassMessage:
|
|
message: str
|
|
nested: PydanticMessage
|
|
|
|
def test_pydantic() -> None:
|
|
serde = Serialization()
|
|
serde.add_type(PydanticMessage)
|
|
|
|
message = PydanticMessage(message="hello")
|
|
name = serde.type_name(message)
|
|
json = serde.serialize(message, type_name=name)
|
|
assert name == "PydanticMessage"
|
|
assert json == '{"message":"hello"}'
|
|
deserialized = serde.deserialize(json, type_name=name)
|
|
assert deserialized == message
|
|
|
|
def test_nested_pydantic() -> None:
|
|
serde = Serialization()
|
|
serde.add_type(NestingPydanticMessage)
|
|
|
|
message = NestingPydanticMessage(message="hello", nested=PydanticMessage(message="world"))
|
|
name = serde.type_name(message)
|
|
json = serde.serialize(message, type_name=name)
|
|
assert json == '{"message":"hello","nested":{"message":"world"}}'
|
|
deserialized = serde.deserialize(json, type_name=name)
|
|
assert deserialized == message
|
|
|
|
def test_dataclass() -> None:
|
|
serde = Serialization()
|
|
serde.add_type(DataclassMessage)
|
|
|
|
message = DataclassMessage(message="hello")
|
|
name = serde.type_name(message)
|
|
json = serde.serialize(message, type_name=name)
|
|
assert json == '{"message": "hello"}'
|
|
deserialized = serde.deserialize(json, type_name=name)
|
|
assert deserialized == message
|
|
|
|
def test_nesting_dataclass_dataclass() -> None:
|
|
serde = Serialization()
|
|
serde.add_type(NestingDataclassMessage)
|
|
|
|
message = NestingDataclassMessage(message="hello", nested=DataclassMessage(message="world"))
|
|
name = serde.type_name(message)
|
|
with pytest.raises(ValueError):
|
|
_json = serde.serialize(message, type_name=name)
|
|
|
|
def test_nesting_dataclass_pydantic() -> None:
|
|
serde = Serialization()
|
|
serde.add_type(NestingPydanticDataclassMessage)
|
|
|
|
message = NestingPydanticDataclassMessage(message="hello", nested=PydanticMessage(message="world"))
|
|
name = serde.type_name(message)
|
|
with pytest.raises(ValueError):
|
|
_json = serde.serialize(message, type_name=name)
|
|
|
|
def test_invalid_type() -> None:
|
|
serde = Serialization()
|
|
try:
|
|
serde.add_type(str) # type: ignore
|
|
except ValueError as e:
|
|
assert str(e) == "Unsupported type <class 'str'>"
|
|
|
|
def test_custom_type() -> None:
|
|
serde = Serialization()
|
|
|
|
class CustomStringTypeDeserializer:
|
|
def deserialize(self, message: str) -> str:
|
|
return message[1:-1]
|
|
|
|
class CustomStringTypeSerializer:
|
|
def serialize(self, message: str) -> str:
|
|
return f'"{message}"'
|
|
|
|
serde.add_type_custom("custom_str", CustomStringTypeDeserializer(), CustomStringTypeSerializer())
|
|
message = "hello"
|
|
json = serde.serialize(message, type_name="custom_str")
|
|
assert json == '"hello"'
|
|
deserialized = serde.deserialize(json, type_name="custom_str")
|
|
assert deserialized == message |