clevermicro/amq-adapter-python
14
0
Fork 0
Code Issues 16 Pull Requests 4 Actions Packages Projects Releases Wiki Activity

issue #23 - fixing issues from integration test, part 3

Browse Source
This commit is contained in:
Stanislav Hejny
2025-05-29 01:05:05 +01:00
parent 59e04300b0
commit 5261b1f959
17 changed files with 290 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
amqp/adapter/cleverthis_service_adapter.py
+14 -8
View File
@@ -110,6 +110,8 @@ async def _convert_file_response(
def is_likely_json(text: str) -> bool:
"""Quick test to see if should convert the response string to JSON format."""
if not isinstance(text, str):
return False
_text = text.strip()
return (_text.startswith("{") and _text.endswith("}")) or (
_text.startswith("[") and _text.endswith("]")
@@ -229,10 +231,10 @@ class CleverThisServiceAdapter:
"""
Pull the value of the Content-Type header from the message headers.
:param message_headers: The headers of the message.
return: The content type_string of the message.
return: The content type of the message.
"""
for header, values in message_headers.items():
if header.lower() == "content-type_string":
if header.lower() == "content-type":
return values[0].split(";")[0]
return "application/json"
@@ -602,7 +604,7 @@ class CleverThisServiceAdapter:
if arg.name in args:
_annotation = arg.annotation
_str_annotation = str(_annotation)
# convert to another type_string if arg is not a str
# convert to another type if arg is not a str
if arg.name == self.authenticated_user_arg_name:
_verified_args[arg.name] = args.get(arg.name)
elif _annotation == int:
@@ -613,15 +615,17 @@ class CleverThisServiceAdapter:
_verified_args[arg.name] = serializer.str_to_bool(args.get(arg.name))
elif (
"Optional" in _str_annotation
or _str_annotation.startswith("list")
or _str_annotation.lower().startswith("list")
or _str_annotation.startswith("typing.List")
or "class" in _str_annotation
):
_type_descr = TypeDescriptor(
_str_annotation, extra_init_data=g_extra_init_data
)
try:
_payload = args.get(arg.name, "")
_verified_args[arg.name] = _type_descr.instantiate(
args.get(arg.name, "")
_payload, is_json=is_likely_json(_payload)
)
except Exception as err:
logging_error(
@@ -642,7 +646,7 @@ class CleverThisServiceAdapter:
_temp = arg.annotation(_temp)
except ValueError as ve:
logging_warning(
f"cannot convert '{_temp}' to Enum type_string {arg.annotation}, {ke}, {ve}"
f"cannot convert '{_temp}' to Enum type {arg.annotation}, {ke}, {ve}"
)
print(f"ENUM -> {_temp}")
_verified_args[arg.name] = _temp
@@ -653,13 +657,15 @@ class CleverThisServiceAdapter:
_temp = _temp.default_factory()
elif hasattr(_temp, "default"):
_temp = _temp.default
if not str(_temp) == "<class 'inspect._empty'>":
_verified_args[arg.name] = _temp
else:
logging_warning(
f"{api_method.__name__}: Missing required argument: {arg.name}"
)
except Exception as unchecked:
logging_error("Failed to validate and instantiate input args", unchecked)
except Exception as e:
logging_error("Failed to validate and instantiate input args", e)
return _verified_args
async def call_cleverthis_api(
amqp/adapter/data_parser.py
+1 -1
View File
@@ -134,7 +134,7 @@ class AMQDataParser:
@staticmethod
def parse_request_body(message: AMQMessage):
"""
Parses the HTTP POST request body based on the MIME type_string.
Parses the HTTP POST request body based on the MIME type.
This is typically called fom the concrete CleverThis Service to extract parameters from the message body.
Args:
message (DataMessage): Contains raw request and MIME headers.
amqp/adapter/data_response_factory.py
+1 -1
View File
@@ -44,7 +44,7 @@ class DataResponseFactory:
Create the DataResponse message supplying all key values
:param id: SnowflakeID
:param response_code: HTTP response code
:param content_type: MIME content type_string
:param content_type: MIME content type
:param body: payload
:param trace_info: Jaeger trace info (id)
:return: DataResponseMessage object
amqp/adapter/file_handler.py
+4 -4
View File
@@ -120,7 +120,7 @@ class FileHandler:
routing_key: str,
loop: AbstractEventLoop,
max_chunk_size: int = 2**20, # 1MB default chunk size
content_type: str = "application/octet-stream", # Default content type_string
content_type: str = "application/octet-stream", # Default content type
delivery_mode: DeliveryMode = DeliveryMode.PERSISTENT,
) -> tuple[str, int]:
"""
@@ -135,7 +135,7 @@ class FileHandler:
routing_key: The name of the RabbitMQ exchange routing key to publish to correct queue.
loop: correct event loop
max_chunk_size: chunk size to send
content_type: The content type_string of the file data.
content_type: The content type of the file data.
delivery_mode: The delivery mode (e.g., PERSISTENT or TRANSIENT).
Returns:
@@ -347,7 +347,7 @@ class StreamingFileHandler(FileHandler):
routing_key: str,
loop: AbstractEventLoop,
max_chunk_size: int = 2**20, # 1MB default chunk size
content_type: str = "application/octet-stream", # Default content type_string
content_type: str = "application/octet-stream", # Default content type
delivery_mode: DeliveryMode = DeliveryMode.PERSISTENT,
) -> tuple[str, int]:
"""
@@ -362,7 +362,7 @@ class StreamingFileHandler(FileHandler):
routing_key: The name of the RabbitMQ exchange routing key to publish to correct queue.
loop: correct event loop
max_chunk_size: chunk size to send
content_type: The content type_string of the file data.
content_type: The content type of the file data.
delivery_mode: The delivery mode (e.g., PERSISTENT or TRANSIENT).
Returns:
amqp/adapter/pydantic_serializer.py
+3 -3
View File
@@ -99,15 +99,15 @@ def deserialize_object(data: str) -> BaseModel:
# process nested model, dict and list
for key, value in obj_dict.items():
key_type: type[Any] = cls.model_fields[key].annotation
# handle union with none, this will give the true type_string
# for example, return `A` if the type_string is `A | None`
# handle union with none, this will give the true type
# for example, return `A` if the type is `A | None`
if get_origin(key_type) is UnionType or get_origin(key_type) is Union:
union_types = get_args(key_type)
if len(union_types) > 2:
# if there are multiple unions, like `A | B | None`
# then we have no way to figure out which one to use
raise ValueError(f"Unsupported Union type: {key_type}")
# return the none-None one as the type_string for processing
# return the none-None one as the type for processing
if union_types[0] is type(None):
key_type = union_types[1]
else:
amqp/adapter/service_message_factory.py
+5 -5
View File
@@ -31,7 +31,7 @@ class ServiceMessageFactory:
"""
if message is not None:
return (
f"ServiceMessage{{id={message.id.as_string()}|type_string={self.format_message_type(message)}|"
f"ServiceMessage{{id={message.id.as_string()}|type={self.format_message_type(message)}|"
f"recipient={message.recipient_name}|replyTo={message.reply_to}|"
f"traceInfo={message.trace_info}|body={base64.b64encode(message.message.encode()).decode() if message.payload_type & 0x80 else message.message}}}"
)
@@ -42,7 +42,7 @@ class ServiceMessageFactory:
) -> ServiceMessage:
"""
Builds ServiceMessage object from relevant values.
:param message_type: enum Message type_string
:param message_type: enum Message type
:param payload: Message content, stored as-is
:param recipient_name: Recipient name (from the routing expression)
:return: ServiceMessage object
@@ -63,7 +63,7 @@ class ServiceMessageFactory:
) -> ServiceMessage:
"""
Builds ServiceMessage object from relevant values, stores content Base64 encoded.
:param message_type: enum Message type_string
:param message_type: enum Message type
:param payload: Message content, stored as Base64 string
:param recipient_name: Recipient name (from the routing expression)
:return: ServiceMessage object
@@ -135,10 +135,10 @@ class ServiceMessageFactory:
def format_message_type(self, message: ServiceMessage) -> str:
"""
Formats the Enum message type_string and decorates it with base64 flag at highest bit.
Formats the Enum message type and decorates it with base64 flag at highest bit.
For compatibility with 0-based enum ordinals in Java subtract 1 from ordinal value.
:param message: Service message to send
:return: formatted message type_string
:return: formatted message type
"""
num_type = (
message.message_type.value if message.message_type else ServiceMessageType.UNKNOWN.value
amqp/adapter/type_descriptor.py
+84 -48
View File
@@ -13,15 +13,14 @@ class TypeDescriptor:
self.type: str = self._extract_type()
def _extract_type(self) -> str:
# Handle Annotated case
if self.type_string.startswith("list") or self.type_string == "str":
return self.type_string
if self.type_string.startswith("<class '"):
return self.type_string[8:-2]
annotated = _extract_type_of(self.type_string, "typing.Optional")
if "Annotated" in annotated:
# Handle Annotated case of FastAPI annotation
if "typing.Optional" in self.type_string:
annotated = _extract_type_of(self.type_string)
if "typing.Annotated" in annotated:
# Extract content between the first square brackets after Annotated
match = _extract_type_of(annotated, "typing.Annotated")
match = _extract_type_of(annotated)
if match:
# Split by comma and get first part, then strip whitespace
parts = [p.strip() for p in match.split(",")]
@@ -31,27 +30,46 @@ class TypeDescriptor:
else:
# Handle simple Optional case
return annotated
return _extract_type_of(self.type_string)
def instantiate(self, payload: Any) -> Any:
return _instantiate(self.type, payload, self.is_json, extra_init_data=self.extra_init_data)
def instantiate(self, payload: Any, is_json: bool = False) -> Any:
return _instantiate(
(
self.type
if "typing.Optional" in self.type_string or self.type_string.startswith("<class ")
else self.type_string
),
payload,
self.is_json or is_json,
extra_init_data=self.extra_init_data,
)
def __repr__(self) -> str:
return f"TypeDescriptor(type='{self.type}', is_json={self.is_json})"
def _extract_type_of(type_string: str, prefix: str) -> str:
_type = type_string.strip().replace(prefix, "") if prefix in type_string else ""
def _extract_type_of(input_string: str) -> str:
stack = []
start_index = -1
end_index = -1
if _type.startswith("["):
_type = _type[1:]
for i, char in enumerate(input_string):
if i > 0: # this assumes paramerized type, that must be at lest 1 char long
if char == "[":
if start_index == -1:
start_index = i
stack.append(char)
elif char == "]":
if len(stack) == 1:
end_index = i
break
if len(stack) > 0:
stack.pop()
if start_index != -1 and end_index != -1:
return input_string[start_index + 1 : end_index].strip()
else:
_type = ""
if _type.endswith("]"):
_type = _type[:-1]
else:
_type = ""
_type = _type.strip()
return _type
return input_string.strip() if start_index == 1 and end_index == -1 else ""
def merge_extra_init_data(cls, json_payload, extra_init_data: dict):
@@ -88,23 +106,57 @@ def _instantiate(
ImportError: If the module/class cannot be imported
TypeError: If instantiation fails
"""
# Split the full path into module and class names
# Handle JSON payload first
try:
json_payload = json.loads(payload) if is_json and isinstance(payload, str) else None
if type_string.startswith("dict") or type_string == "str":
return payload
if type_string.startswith("list"):
if json_payload is not None:
return json_payload
_list_type = _extract_type_of(type_string, "list")
if _list_type:
return [
_instantiate(_list_type, p, extra_init_data=extra_init_data)
for p in payload.split(",")
]
return payload.split(",")
except Exception:
json_payload = None
# Handle basic types
if type_string == "str":
return str(payload)
if type_string == "int":
return int(payload)
if type_string == "float":
return float(payload)
if type_string == "bool":
return bool(payload)
# Handle typing module types
if type_string.startswith("typing."):
if type_string.startswith("typing.List") or type_string.startswith("typing.Sequence"):
if json_payload is not None:
return json_payload
inner_type = _extract_type_of(type_string)
if inner_type:
return [
_instantiate(inner_type, p, extra_init_data=extra_init_data)
for p in (payload if isinstance(payload, list) else payload.split(","))
]
return payload if isinstance(payload, list) else payload.split(",")
if type_string.startswith("typing.Dict") or type_string.startswith("typing.Mapping"):
if json_payload is not None:
return json_payload
return dict(payload)
if type_string.startswith("typing.Set"):
if json_payload is not None:
return set(json_payload)
return set(payload if isinstance(payload, list) else payload.split(","))
if type_string.startswith("typing.Tuple"):
if json_payload is not None:
return tuple(json_payload)
return tuple(payload if isinstance(payload, list) else payload.split(","))
if type_string.startswith("typing.Optional"):
inner_type = _extract_type_of(type_string)
if inner_type:
return _instantiate(inner_type, payload, is_json, extra_init_data)
return payload
# Handle regular class instantiation
parts = type_string.split(".")
module_path = ".".join(parts[:-1])
class_name = parts[-1]
@@ -140,19 +192,3 @@ def _instantiate(
raise ImportError(f"Module '{module_path}' has no class '{class_name}'") from e
except TypeError as e:
raise TypeError(f"Failed to instantiate {type_string} with payload {payload}") from e
# Test cases
if __name__ == "__main__":
test_cases = [
"typing.Optional[typing.Annotated[core.base.providers.ingestion.IngestionConfig, Json]]",
"typing.Optional[uuid.UUID]",
"typing.Optional[typing.Annotated[str, SomethingElse]]",
"typing.Optional[typing.Annotated[list[int], Json, more_args]]",
]
for test in test_cases:
td = TypeDescriptor(test)
print(f"Input: {test}")
print(f"Result: {td}")
print()
amqp/model/model.py
+2 -2
View File
@@ -31,7 +31,7 @@ class CleverMicroMessage:
The common structure of a message that transfers the user request to the Service and back. All data related messages
conform to this base structure:
magic - identifies the message type_string (REST, RPC, DataResponse)
magic - identifies the message type (REST, RPC, DataResponse)
id - unique message ID
m_field - method name (for REST style calls) or service name (for RPC style calls) or response code for DataResponse
d_field - domain name (for REST style calls) or module name (for RPC style calls) or error code for DataResponse
@@ -145,7 +145,7 @@ class MultipartDataMessage(DataMessage):
@dataclass
class AMQMessage(DataMessage):
"""
Expand the core DataMessage type_string to include the channel, which is required to create a dedicated queue
Expand the core DataMessage type to include the channel, which is required to create a dedicated queue
in case the response should initiate file download.
"""
amqp/router/router_consumer.py
+1 -1
View File
@@ -134,7 +134,7 @@ class RouterConsumer(RouterProducer):
)
else:
logging_error(
"******[%s] (ROUTING_DATA_REQ) [%s] produces invalid or no CM msg type_string.",
"******[%s] (ROUTING_DATA_REQ) [%s] produces invalid or no CM msg type.",
delivery_tag,
str(message.body),
)
amqp/router/router_producer.py
+1 -1
View File
@@ -115,7 +115,7 @@ class RouterProducer(RouterBase):
async def handle_routing_data_message(self, message: AbstractIncomingMessage):
"""
* Handler for CLEVER_MICRO_RESPONSE queue messages. Used in Master mode to handle receipt of
* routing data, invokes addRoute or removeRoute depending on the type_string of the message.
* routing data, invokes addRoute or removeRoute depending on the type of the message.
"""
logging_info(
"[%s] Received ROUTING DATA: [%s], msgId=%s",
amqp/service/amq_message_handler.py
+1 -1
View File
@@ -268,7 +268,7 @@ class DataMessageHandler:
# reply.tracing_span().end()
# TODO - figure out the right response format (not needed for Clever Swarm v0.1)
# that currently runs Java version, this method here is for the future compatibility to enable
# internal, service-2-service communication, and the proper response type_string will be decided than
# internal, service-2-service communication, and the proper response type will be decided than
# reply.response().set_result(Response.ok(result.body, content_type).build())
if future:
response: DataResponse = DataResponseFactory.from_bytes(message.body)
cleverbrag/clever_brag_adapter.py
+2 -2
View File
@@ -24,7 +24,7 @@ from amqp.service.amq_service import AMQService
class BRAGArgument:
"""
This class is used to represent a BRAG API input argument. It is uses FastAPI annotation to define the arguments
(the argument type_string) that are passed as input arguments to the BRAG endpoint.
(the argument type) that are passed as input arguments to the BRAG endpoint.
"""
def __init__(self, arg: Any):
@@ -88,7 +88,7 @@ def convert_to_argument(arg: BRAGArgument, value: str) -> Any:
elif arg.is_base_model:
return arg.arg.parse_raw(value)
else:
print(f"Unsupported argument type_string: {type(arg)}")
print(f"Unsupported argument type: {type(arg)}")
except Exception as e:
print(f"Error converting argument {arg}: {e}")
return None
tests/adapter/test_backpressure_handler.py
+2 -2
View File
@@ -86,13 +86,13 @@ class TestScaleRequestV1:
def test_scale_request_v1_to_json_different_alert_type(self):
"""
Test the to_json method with a different ScalingRequestAlert type_string.
Test the to_json method with a different ScalingRequestAlert type.
"""
service_id = "test-service"
task_id = "test-task"
max_availability = 10
current_availability = 5
request_type = ScalingRequestAlert.OVERLOAD # Change the alert type_string
request_type = ScalingRequestAlert.OVERLOAD # Change the alert type
scale_request = ScaleRequestV1(
service_id,
tests/adapter/test_cleverthis_service_adapter.py
+145 -1
View File
@@ -1,10 +1,13 @@
import asyncio
import unittest
from enum import Enum
from typing import Any, List, Optional
from unittest.mock import AsyncMock, MagicMock, patch
from aio_pika.abc import AbstractChannel, AbstractRobustConnection
from aiohttp import ClientResponse
from aiohttp.web_fileresponse import FileResponse
from pydantic import BaseModel
from amqp.adapter.cleverthis_service_adapter import (
AMQMessage,
@@ -19,6 +22,26 @@ from amqp.model.model import DataResponse
from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE
class NestedModel(BaseModel):
id: int
name: str
class ComplexModel(BaseModel):
nested: NestedModel
items: List[str]
class TestEnum(Enum):
VALUE1 = "value1"
VALUE2 = "value2"
class TestModel(BaseModel):
name: str
value: int
class TestCleverThisServiceAdapter(unittest.IsolatedAsyncioTestCase):
async def test_create_reply_exchange_and_queue_success(self):
@@ -89,6 +112,7 @@ class TestCleverThisServiceAdapter(unittest.IsolatedAsyncioTestCase):
amq_configuration=self.mock_amq_config, routes=[], session=self.mock_session
)
self.adapter.service_host = "testhost"
self.adapter.authenticated_user_arg_name = "user"
def test_init(self):
self.assertEqual(self.adapter.service_host, "testhost")
@@ -98,7 +122,7 @@ class TestCleverThisServiceAdapter(unittest.IsolatedAsyncioTestCase):
def test_get_content_type(self):
test_cases = [
({"Content-Type": ["application/json"]}, "application/json"),
({"content-type_string": ["text/xml"]}, "text/xml"),
({"content-type": ["text/xml"]}, "text/xml"),
({"CONTENT-TYPE": ["text/plain; charset=utf-8"]}, "text/plain"),
({}, "application/json"),
({"Other-Header": ["value"]}, "application/json"),
@@ -282,6 +306,126 @@ class TestCleverThisServiceAdapter(unittest.IsolatedAsyncioTestCase):
with self.assertRaises(NotImplementedError):
await self.adapter.head(mock_message, None)
def test_basic_type_conversion(self):
async def test_method(x: int, y: float, z: bool, text: str):
pass
args = {"x": "123", "y": "45.67", "z": "true", "text": "hello"}
result = self.adapter._verify_and_instantiate_arguments(args, test_method)
self.assertEqual(result["x"], 123)
self.assertEqual(result["y"], 45.67)
self.assertEqual(result["z"], True)
self.assertEqual(result["text"], "hello")
def test_enum_conversion(self):
async def test_method(status: TestEnum):
pass
args = {"status": "value1"}
result = self.adapter._verify_and_instantiate_arguments(args, test_method)
self.assertEqual(result["status"], TestEnum.VALUE1)
def test_optional_parameter(self):
async def test_method(required: str, optional: Optional[str] = None):
pass
args = {"required": "test"}
result = self.adapter._verify_and_instantiate_arguments(args, test_method)
self.assertEqual(result["required"], "test")
self.assertIsNone(result["optional"])
def test_list_parameter(self):
async def test_method(items: List[str]):
pass
args = {"items": "item1,item2,item3"}
result = self.adapter._verify_and_instantiate_arguments(args, test_method)
self.assertEqual(result["items"], ["item1", "item2", "item3"])
def test_pydantic_model_parameter(self):
async def test_method(model: TestModel):
pass
args = {"model": '{"name": "test", "value": 42}'}
result = self.adapter._verify_and_instantiate_arguments(args, test_method)
self.assertIsInstance(result["model"], TestModel)
self.assertEqual(result["model"].name, "test")
self.assertEqual(result["model"].value, 42)
def test_authenticated_user_parameter(self):
async def test_method(user: Any):
pass
user_obj = MagicMock()
args = {"user": user_obj}
result = self.adapter._verify_and_instantiate_arguments(args, test_method)
self.assertEqual(result["user"], user_obj)
def test_missing_required_parameter(self):
async def test_method(required: str):
pass
args = {}
result = self.adapter._verify_and_instantiate_arguments(args, test_method)
self.assertNotIn("required", result)
def test_invalid_enum_value(self):
async def test_method(status: TestEnum):
pass
args = {"status": "invalid_value"}
result = self.adapter._verify_and_instantiate_arguments(args, test_method)
self.assertEqual(result["status"], "invalid_value")
def test_invalid_type_conversion(self):
async def test_method(x: int):
pass
args = {"x": "not_a_number"}
result = {}
try:
result = self.adapter._verify_and_instantiate_arguments(args, test_method)
except Exception as e:
self.assertTrue(isinstance(e, TypeError))
self.assertEqual(len(result), 0)
def test_complex_nested_model(self):
async def test_method(model: ComplexModel):
pass
args = {"model": {"nested": {"id": 1, "name": "test"}, "items": ["item1", "item2"]}}
result = self.adapter._verify_and_instantiate_arguments(args, test_method)
self.assertIsInstance(result["model"], ComplexModel)
self.assertEqual(result["model"].nested.id, 1)
self.assertEqual(result["model"].nested.name, "test")
self.assertEqual(result["model"].items, ["item1", "item2"])
def test_default_factory_parameter(self):
class DefaultFactory:
@staticmethod
def default_factory():
return "default_value"
async def test_method(param: str = DefaultFactory()):
pass
args = {}
result = self.adapter._verify_and_instantiate_arguments(args, test_method)
self.assertEqual(result["param"], "default_value")
if __name__ == "__main__":
unittest.main()
tests/adapter/test_service_message_factory.py
+2 -2
View File
@@ -15,11 +15,11 @@ class ServiceMessageTest(TestCase):
_f: ServiceMessageFactory = ServiceMessageFactory("bumble-bee")
message: ServiceMessage = _f.of(ServiceMessageType.ROUTING_DATA, "Duck", "Donald")
message_str = _f.to_string(message)
self.assertTrue("|type_string=01|" in message_str)
self.assertTrue("|type=01|" in message_str)
self.assertTrue("|body=Duck" in message_str)
message: ServiceMessage = _f.of(ServiceMessageType.ROUTING_DATA_REPLACE, "Duck", "Donald")
message_str = _f.to_string(message)
self.assertTrue("|type_string=03|" in message_str)
self.assertTrue("|type=03|" in message_str)
self.assertTrue("|body=Duck" in message_str)
self.assertTrue("|replyTo=bumble-bee" in message_str)
nullMsg = _f.to_string(None)
tests/adapter/test_type_descriptor.py
+9 -10
View File
@@ -8,28 +8,28 @@ class TestTypeDescriptor(unittest.TestCase):
type_str = (
"typing.Optional[typing.Annotated[core.base.providers.ingestion.IngestionConfig, Json]]"
)
td = TypeDescriptor(type_str)
td = TypeDescriptor(type_str, {})
self.assertEqual(td.type, "core.base.providers.ingestion.IngestionConfig")
self.assertTrue(td.is_json)
def test_annotated_without_json(self):
type_str = "typing.Optional[typing.Annotated[str, SomethingElse]]"
td = TypeDescriptor(type_str)
td = TypeDescriptor(type_str, {})
self.assertEqual(td.type, "str")
self.assertFalse(td.is_json)
def test_annotated_multiple_args(self):
type_str = "typing.Optional[typing.Annotated[list[int], Json, more_args]]"
td = TypeDescriptor(type_str)
td = TypeDescriptor(type_str, {})
self.assertEqual(td.type, "list[int]")
self.assertTrue(td.is_json)
def test_simple_optional(self):
type_str = "typing.Optional[uuid.UUID]"
td = TypeDescriptor(type_str)
td = TypeDescriptor(type_str, {})
self.assertEqual(td.type, "uuid.UUID")
self.assertFalse(td.is_json)
@@ -43,29 +43,28 @@ class TestTypeDescriptor(unittest.TestCase):
for type_str, expected_type, expected_json in test_cases:
with self.subTest(type_str=type_str):
td = TypeDescriptor(type_str)
td = TypeDescriptor(type_str, {})
self.assertEqual(td.type, expected_type)
self.assertEqual(td.is_json, expected_json)
def test_invalid_input(self):
test_cases = [
"not.a.valid.type_string.string",
"typing.Optional[missing_bracket",
"typing.Optional[]",
"typing.Annotated[without.optional]",
"[without.optional]",
]
for type_str in test_cases:
with self.subTest(type_str=type_str):
td = TypeDescriptor(type_str)
td = TypeDescriptor(type_str, {})
self.assertEqual(td.type, "")
self.assertFalse(td.is_json)
def test_repr(self):
type_str = "typing.Optional[typing.Annotated[str, Json]]"
td = TypeDescriptor(type_str)
td = TypeDescriptor(type_str, {})
self.assertEqual(repr(td), "TypeDescriptor(type_string='str', is_json=True)")
self.assertEqual(repr(td), "TypeDescriptor(type='str', is_json=True)")
if __name__ == "__main__":
tests/model/test_model.py
+1 -1
View File
@@ -49,7 +49,7 @@ class TestCleverMicroMessage(unittest.TestCase):
def test_content_type(self):
self.assertEqual(self.message.content_type(), "application/json")
# Test default content type_string
# Test default content type
message_no_content_type = CleverMicroMessage(
magic="A",
id=self.test_id,