From 82c8a298cdafd39a549f5b6e7d40250828551238 Mon Sep 17 00:00:00 2001 From: Stanislav Hejny Date: Tue, 27 May 2025 13:34:09 +0100 Subject: [PATCH] issue #23 - fixing issues from integration test --- amqp/adapter/cleverthis_service_adapter.py | 115 ++++++++++--- amqp/adapter/data_parser.py | 5 +- amqp/adapter/data_response_factory.py | 2 +- amqp/adapter/file_handler.py | 8 +- amqp/adapter/pydantic_serializer.py | 6 +- amqp/adapter/service_message_factory.py | 10 +- amqp/adapter/type_descriptor.py | 158 ++++++++++++++++++ amqp/model/model.py | 4 +- amqp/router/router_consumer.py | 2 +- amqp/router/router_producer.py | 2 +- amqp/service/amq_message_handler.py | 2 +- cleverbrag/clever_brag_adapter.py | 9 +- cleverbrag/main.py | 5 +- tests/adapter/test_backpressure_handler.py | 4 +- .../test_cleverthis_service_adapter.py | 2 +- tests/adapter/test_service_message_factory.py | 17 +- tests/adapter/test_type_descriptor.py | 72 ++++++++ tests/model/test_model.py | 2 +- 18 files changed, 364 insertions(+), 61 deletions(-) create mode 100644 amqp/adapter/type_descriptor.py create mode 100644 tests/adapter/test_type_descriptor.py diff --git a/amqp/adapter/cleverthis_service_adapter.py b/amqp/adapter/cleverthis_service_adapter.py index f09e843..9eab27c 100644 --- a/amqp/adapter/cleverthis_service_adapter.py +++ b/amqp/adapter/cleverthis_service_adapter.py @@ -4,6 +4,7 @@ Methods here will invoke the actual CleverThis Service code, which are passed as """ import asyncio +import enum import inspect import json import pathlib @@ -19,6 +20,7 @@ from aio_pika.abc import ( ) from aiohttp import ClientResponse, ClientSession from fastapi import HTTPException, UploadFile +from fastapi.params import File from fastapi.routing import APIRoute from fastapi.security import OAuth2PasswordRequestForm from pydantic import BaseModel @@ -29,12 +31,20 @@ from amqp.adapter import serializer from amqp.adapter.data_parser import AMQDataParser from amqp.adapter.data_response_factory import DataResponseFactory from amqp.adapter.file_handler import FileHandler, StreamingFileHandler -from amqp.adapter.logging_utils import logging_debug, logging_error, logging_info +from amqp.adapter.logging_utils import ( + logging_debug, + logging_error, + logging_info, + logging_warning, +) +from amqp.adapter.type_descriptor import TypeDescriptor from amqp.config.amq_configuration import AMQConfiguration from amqp.model.model import AMQErrorMessage, AMQMessage, DataMessage, DataResponse from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE from amqp.router.utils import await_result +g_extra_init_data: dict = dict() + async def _create_reply_exchange_and_queue( connection: AbstractRobustConnection, @@ -120,12 +130,17 @@ class CleverThisServiceAdapter: amq_configuration: AMQConfiguration, routes: List[APIRoute], session: ClientSession = None, + authenticated_user_arg_name: str = "authenticated_user", + extra_init_data=None, ): """ Initializes the CleverThisServiceAdapter with the given AMQ configuration and optional HTTP session. The HTTP session is for the case when the adapter uses loose coupling with the CleverThis service via HTTP REST API. """ + if extra_init_data is None: + extra_init_data = dict() + self.authenticated_user_arg_name = authenticated_user_arg_name self.amq_configuration = amq_configuration self.session = session self.service_host = self.amq_configuration.dispatch.service_host @@ -135,6 +150,8 @@ class CleverThisServiceAdapter: self.amq_configuration.amq_adapter.require_authenticated_user ) self.routes = self.group_routes_by_method(api_routes=routes) + global g_extra_init_data + g_extra_init_data = extra_init_data or {} def get_endpoint(self, endpoint: Any) -> Any: """ @@ -212,10 +229,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 of the message. + return: The content type_string of the message. """ for header, values in message_headers.items(): - if header.lower() == "content-type": + if header.lower() == "content-type_string": return values[0].split(";")[0] return "application/json" @@ -474,7 +491,9 @@ class CleverThisServiceAdapter: _match = _route.path_regex.match(_path) if _match: _args.update(_match.groupdict()) - _args["authenticated_user"] = auth_user + _signature_args = inspect.signature(_route.endpoint).parameters + if self.authenticated_user_arg_name in _signature_args.keys(): + _args[self.authenticated_user_arg_name] = auth_user # If the path matches, call the corresponding function return await CleverThisServiceAdapter.call_cleverthis_api( message, @@ -513,33 +532,39 @@ class CleverThisServiceAdapter: _form_data.update(_query_args) for _body_param in _route.dependant.body_params: if _body_param.name in _form_data: - _file_data = _form_data[_body_param.name] - actual_filepath = message.extra_data[_body_param.name] + _file_data = _form_data.get(_body_param.name, None) + actual_filepath = message.extra_data.get(_body_param.name, None) _form_data[_body_param.name] = ( - _body_param.type_.__name__ == "UploadFile" + ( + _body_param.type_.__name__ == "UploadFile" + or isinstance(_body_param.field_info, File) + ) + and _file_data is not None + and actual_filepath is not None and UploadFile( file=pathlib.Path(actual_filepath).open("rb"), - size=_file_data["size"], - filename=_file_data["filename"], - headers=Headers({"Content-Type": _file_data["mediaType"]}), + size=_file_data.get("size"), + filename=_file_data.get("filename"), + headers=Headers({"Content-Type": _file_data.get("mediaType")}), ) or _form_data[_body_param.name] ) _verified_args = {} _required_args = getattr(_route.endpoint, "__annotations__", {}) + _signature_args = inspect.signature(_route.endpoint).parameters for arg in _required_args: if arg in _form_data: _verified_args[arg] = _form_data[arg] else: - # needed for /login input - if arg == "authenticated_user": - _verified_args["authenticated_user"] = auth_user + # needed for CleverSwarm /login input if _required_args[arg].__name__ == "OAuth2PasswordRequestForm": try: _verified_args[arg] = OAuth2PasswordRequestForm(**_form_data) except Exception as e: print(f"Error: {e}") + if self.authenticated_user_arg_name in _signature_args.keys(): + _verified_args[self.authenticated_user_arg_name] = auth_user return await CleverThisServiceAdapter.call_cleverthis_api( message, @@ -584,18 +609,62 @@ class CleverThisServiceAdapter: # _required_args = getattr(api_method, '__annotations__', {}) for arg in _required_args: if arg.name in args: - # convert to another type if arg is not a str - if arg.annotation == int: - _verified_args[arg.name] = int(args[arg.name]) - elif arg.annotation == float: - _verified_args[arg.name] = float(args[arg.name]) - elif arg.annotation == bool: - _verified_args[arg.name] = serializer.str_to_bool(args[arg.name]) - # TODO: more? + _annotation = arg.annotation + _str_annotation = str(_annotation) + # convert to another type_string if arg is not a str + if _annotation == int: + _verified_args[arg.name] = int(args.get(arg.name)) + elif _annotation == float: + _verified_args[arg.name] = float(args.get(arg.name)) + elif _annotation == bool: + _verified_args[arg.name] = serializer.str_to_bool(args.get(arg.name)) + elif ( + "Optional" in _str_annotation + or _str_annotation.startswith("list") + or "class" in _str_annotation + ): + _type_descr = TypeDescriptor( + _str_annotation, extra_init_data=g_extra_init_data + ) + try: + _verified_args[arg.name] = _type_descr.instantiate( + args.get(arg.name, "") + ) + except Exception as err: + logging_error( + f"Could not instantiate argument {arg}, defaulting to its string value", + err, + ) + _verified_args[arg.name] = args.get(arg.name, None) else: - _verified_args[arg.name] = args[arg.name] + _temp = args.get(arg.name) + if isinstance(arg.annotation, enum.EnumType) or isinstance( + arg.annotation, enum.Enum + ): + try: + _temp = arg.annotation[_temp] + except KeyError as ke: + # some enums have string value different to key spelling, try different thing + try: + _temp = arg.annotation(_temp) + except ValueError as ve: + logging_warning( + f"cannot convert '{_temp}' to Enum type_string {arg.annotation}, {ke}, {ve}" + ) + print(f"ENUM -> {_temp}") + _verified_args[arg.name] = _temp else: - logging_info(f"{api_method.__name__}: Missing required argument: {arg.name}") + if hasattr(arg, "default"): + _temp = arg.default + if hasattr(_temp, "default_factory"): + _temp = _temp.default_factory() + elif hasattr(_temp, "default"): + _temp = _temp.default + _verified_args[arg.name] = _temp + else: + logging_warning( + f"{api_method.__name__}: Missing required argument: {arg.name}" + ) logging_info( "INVOKE ENDPOINT: {%s}, ARGS: {%s}", api_method.__name__, diff --git a/amqp/adapter/data_parser.py b/amqp/adapter/data_parser.py index 96d718f..8ae28fd 100644 --- a/amqp/adapter/data_parser.py +++ b/amqp/adapter/data_parser.py @@ -51,13 +51,14 @@ class MultipartFormDataParser: ) self.is_multipart = True except Exception as e: - logging_error(f"Error parsing multipart/form-data: {e}. Trying parsing as JSON.") + __error = f"Error parsing multipart/form-data: {e}. Trying parsing as JSON." try: self.form_data = json.loads(message.body()) self.is_json = True if self.form_data["files"] is not None and self.form_data["form"] is not None: self.form_data = AMQDataParser.process_form_data(self.form_data) except Exception as jsonerror: + logging_error(__error) logging_error(f"Parsing JSON: {jsonerror}") self.is_valid = False @@ -133,7 +134,7 @@ class AMQDataParser: @staticmethod def parse_request_body(message: AMQMessage): """ - Parses the HTTP POST request body based on the MIME type. + Parses the HTTP POST request body based on the MIME type_string. 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. diff --git a/amqp/adapter/data_response_factory.py b/amqp/adapter/data_response_factory.py index 46ef4b1..ea72ee2 100644 --- a/amqp/adapter/data_response_factory.py +++ b/amqp/adapter/data_response_factory.py @@ -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 + :param content_type: MIME content type_string :param body: payload :param trace_info: Jaeger trace info (id) :return: DataResponseMessage object diff --git a/amqp/adapter/file_handler.py b/amqp/adapter/file_handler.py index 02c209b..f07bbec 100644 --- a/amqp/adapter/file_handler.py +++ b/amqp/adapter/file_handler.py @@ -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 + content_type: str = "application/octet-stream", # Default content type_string 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 of the file data. + content_type: The content type_string 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 + content_type: str = "application/octet-stream", # Default content type_string 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 of the file data. + content_type: The content type_string of the file data. delivery_mode: The delivery mode (e.g., PERSISTENT or TRANSIENT). Returns: diff --git a/amqp/adapter/pydantic_serializer.py b/amqp/adapter/pydantic_serializer.py index 27a61ae..1a1afa4 100644 --- a/amqp/adapter/pydantic_serializer.py +++ b/amqp/adapter/pydantic_serializer.py @@ -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 - # for example, return `A` if the type is `A | None` + # handle union with none, this will give the true type_string + # for example, return `A` if the type_string 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 for processing + # return the none-None one as the type_string for processing if union_types[0] is type(None): key_type = union_types[1] else: diff --git a/amqp/adapter/service_message_factory.py b/amqp/adapter/service_message_factory.py index 2e03d74..061e38c 100644 --- a/amqp/adapter/service_message_factory.py +++ b/amqp/adapter/service_message_factory.py @@ -31,7 +31,7 @@ class ServiceMessageFactory: """ if message is not None: return ( - f"ServiceMessage{{id={message.id.as_string()}|type={self.format_message_type(message)}|" + f"ServiceMessage{{id={message.id.as_string()}|type_string={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 + :param message_type: enum Message type_string :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 + :param message_type: enum Message type_string :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 and decorates it with base64 flag at highest bit. + Formats the Enum message type_string 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 + :return: formatted message type_string """ num_type = ( message.message_type.value if message.message_type else ServiceMessageType.UNKNOWN.value diff --git a/amqp/adapter/type_descriptor.py b/amqp/adapter/type_descriptor.py new file mode 100644 index 0000000..82d246c --- /dev/null +++ b/amqp/adapter/type_descriptor.py @@ -0,0 +1,158 @@ +""" """ + +import importlib +import json +from typing import Any + + +class TypeDescriptor: + def __init__(self, type_string: str, extra_init_data: dict): + self.type_string = type_string + self.extra_init_data = extra_init_data or {} + self.is_json: bool = False + 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(" 1: + self.is_json = parts[1].lower() == "json" + return parts[0] if parts else "" + else: + # Handle simple Optional case + return annotated + + def instantiate(self, payload: Any) -> Any: + return _instantiate(self.type, payload, self.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 "" + + if _type.startswith("["): + _type = _type[1:] + else: + _type = "" + if _type.endswith("]"): + _type = _type[:-1] + else: + _type = "" + _type = _type.strip() + return _type + + +def merge_extra_init_data(cls, json_payload, extra_init_data: dict): + """ + Merges extra_init_data into json_payload only when: + 1. cls has an attribute named like the key + 2. json_payload doesn't already have the key + """ + return { + **json_payload, + **{ + k: v + for k, v in extra_init_data.items() + if (hasattr(cls, k) or (hasattr(cls, "model_fields") and k in cls.model_fields)) + and k not in json_payload + }, + } + + +def _instantiate( + type_string: str, payload: Any, is_json: bool = False, extra_init_data: dict = None +) -> Any: + """ + Dynamically imports and instantiates a class from a TypeDescriptor's type_string attribute. + + Args: + type_string: TypeDescriptor instance with the class path + payload: Argument to pass to the constructor + + Returns: + Instance of the target class initialized with payload + + Raises: + ImportError: If the module/class cannot be imported + TypeError: If instantiation fails + """ + # Split the full path into module and class names + 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(",") + if type_string == "int": + return int(payload) + + parts = type_string.split(".") + module_path = ".".join(parts[:-1]) + class_name = parts[-1] + + try: + # Import the module + module = importlib.import_module(module_path) + # Get the class + cls = getattr(module, class_name) + if isinstance(payload, cls): + return payload + # Instantiate with payload + if json_payload is not None: + _args = merge_extra_init_data(cls, json_payload, extra_init_data or {}) + if hasattr(cls, "from_dict"): + return cls.from_dict(_args) + elif hasattr(cls, "create"): + return cls.create(**_args) + return cls(**_args) + + if isinstance(payload, dict): + _args = merge_extra_init_data(cls, payload, extra_init_data or {}) + if hasattr(cls, "from_dict"): + return cls.from_dict(_args) + elif hasattr(cls, "create"): + return cls.create(**_args) + return cls(**_args) + + return cls(payload) + except ImportError as e: + raise ImportError(f"Could not import module '{module_path}'") from e + except AttributeError as e: + 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() diff --git a/amqp/model/model.py b/amqp/model/model.py index fe31fc4..9c1e57e 100644 --- a/amqp/model/model.py +++ b/amqp/model/model.py @@ -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 (REST, RPC, DataResponse) + magic - identifies the message type_string (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 to include the channel, which is required to create a dedicated queue + Expand the core DataMessage type_string to include the channel, which is required to create a dedicated queue in case the response should initiate file download. """ diff --git a/amqp/router/router_consumer.py b/amqp/router/router_consumer.py index 10f0a23..db42cdc 100644 --- a/amqp/router/router_consumer.py +++ b/amqp/router/router_consumer.py @@ -134,7 +134,7 @@ class RouterConsumer(RouterProducer): ) else: logging_error( - "******[%s] (ROUTING_DATA_REQ) [%s] produces invalid or no CM msg type.", + "******[%s] (ROUTING_DATA_REQ) [%s] produces invalid or no CM msg type_string.", delivery_tag, str(message.body), ) diff --git a/amqp/router/router_producer.py b/amqp/router/router_producer.py index 6224dc1..7f59bd3 100644 --- a/amqp/router/router_producer.py +++ b/amqp/router/router_producer.py @@ -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 of the message. + * routing data, invokes addRoute or removeRoute depending on the type_string of the message. """ logging_info( "[%s] Received ROUTING DATA: [%s], msgId=%s", diff --git a/amqp/service/amq_message_handler.py b/amqp/service/amq_message_handler.py index af67d0d..0692e0d 100644 --- a/amqp/service/amq_message_handler.py +++ b/amqp/service/amq_message_handler.py @@ -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 will be decided than + # internal, service-2-service communication, and the proper response type_string will be decided than # reply.response().set_result(Response.ok(result.body, content_type).build()) if future: response: DataResponse = DataResponseFactory.from_bytes(message.body) diff --git a/cleverbrag/clever_brag_adapter.py b/cleverbrag/clever_brag_adapter.py index 47112b7..e2d9f77 100644 --- a/cleverbrag/clever_brag_adapter.py +++ b/cleverbrag/clever_brag_adapter.py @@ -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) that are passed as input arguments to the BRAG endpoint. + (the argument type_string) 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: {type(arg)}") + print(f"Unsupported argument type_string: {type(arg)}") except Exception as e: print(f"Error converting argument {arg}: {e}") return None @@ -104,8 +104,11 @@ class CleverBragAmqpAdapter(CleverThisServiceAdapter): amq_configuration: AMQConfiguration, routes: List[APIRoute], auth_provider: core.base.providers.auth.AuthProvider, + extra_init_data: dict, ): - super().__init__(amq_configuration, routes, None) + super().__init__( + amq_configuration, routes, None, "auth_user", extra_init_data=extra_init_data + ) self.auth_provider = auth_provider _amq_service: AMQService = AMQService(amq_configuration, self) self.thread = Thread(target=_amq_service.run) diff --git a/cleverbrag/main.py b/cleverbrag/main.py index e8f97a6..5bf9fef 100644 --- a/cleverbrag/main.py +++ b/cleverbrag/main.py @@ -21,7 +21,10 @@ if __name__ == "__main__": r2rapp: R2RApp = create_r2r_app() adapter: CleverBragAmqpAdapter = CleverBragAmqpAdapter( - AMQConfiguration("./application.properties.local"), r2rapp.app.routes, r2rapp.providers.auth + AMQConfiguration("./application.properties.local"), + r2rapp.app.routes, + r2rapp.providers.auth, + {"app": r2rapp.config.app}, ) for method, routes in adapter.routes.items(): diff --git a/tests/adapter/test_backpressure_handler.py b/tests/adapter/test_backpressure_handler.py index 73482e0..9d6534e 100644 --- a/tests/adapter/test_backpressure_handler.py +++ b/tests/adapter/test_backpressure_handler.py @@ -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. + Test the to_json method with a different ScalingRequestAlert type_string. """ service_id = "test-service" task_id = "test-task" max_availability = 10 current_availability = 5 - request_type = ScalingRequestAlert.OVERLOAD # Change the alert type + request_type = ScalingRequestAlert.OVERLOAD # Change the alert type_string scale_request = ScaleRequestV1( service_id, diff --git a/tests/adapter/test_cleverthis_service_adapter.py b/tests/adapter/test_cleverthis_service_adapter.py index e3688c4..34ff75c 100644 --- a/tests/adapter/test_cleverthis_service_adapter.py +++ b/tests/adapter/test_cleverthis_service_adapter.py @@ -98,7 +98,7 @@ class TestCleverThisServiceAdapter(unittest.IsolatedAsyncioTestCase): def test_get_content_type(self): test_cases = [ ({"Content-Type": ["application/json"]}, "application/json"), - ({"content-type": ["text/xml"]}, "text/xml"), + ({"content-type_string": ["text/xml"]}, "text/xml"), ({"CONTENT-TYPE": ["text/plain; charset=utf-8"]}, "text/plain"), ({}, "application/json"), ({"Other-Header": ["value"]}, "application/json"), diff --git a/tests/adapter/test_service_message_factory.py b/tests/adapter/test_service_message_factory.py index 72ab467..97e40b1 100644 --- a/tests/adapter/test_service_message_factory.py +++ b/tests/adapter/test_service_message_factory.py @@ -1,9 +1,9 @@ from unittest import TestCase from amqp.adapter.service_message_factory import ( - ServiceMessageFactory, - RS, INVALID_MESSAGE, + RS, + ServiceMessageFactory, ) from amqp.model.model import ServiceMessage from amqp.model.service_message_type import ServiceMessageType @@ -13,17 +13,13 @@ class ServiceMessageTest(TestCase): def test_to_string(self): _f: ServiceMessageFactory = ServiceMessageFactory("bumble-bee") - message: ServiceMessage = _f.of( - ServiceMessageType.ROUTING_DATA, "Duck", "Donald" - ) + message: ServiceMessage = _f.of(ServiceMessageType.ROUTING_DATA, "Duck", "Donald") message_str = _f.to_string(message) - self.assertTrue("|type=01|" in message_str) + self.assertTrue("|type_string=01|" in message_str) self.assertTrue("|body=Duck" in message_str) - message: ServiceMessage = _f.of( - ServiceMessageType.ROUTING_DATA_REPLACE, "Duck", "Donald" - ) + message: ServiceMessage = _f.of(ServiceMessageType.ROUTING_DATA_REPLACE, "Duck", "Donald") message_str = _f.to_string(message) - self.assertTrue("|type=03|" in message_str) + self.assertTrue("|type_string=03|" in message_str) self.assertTrue("|body=Duck" in message_str) self.assertTrue("|replyTo=bumble-bee" in message_str) nullMsg = _f.to_string(None) @@ -57,3 +53,4 @@ class ServiceMessageTest(TestCase): message = _f.of(ServiceMessageType.ROUTING_DATA, "Duck", "Donald") serialized = _f.serialize(message) assert RS + "0" + str(ServiceMessageType.ROUTING_DATA.value) + assert len(serialized) > 16 diff --git a/tests/adapter/test_type_descriptor.py b/tests/adapter/test_type_descriptor.py new file mode 100644 index 0000000..ccff6fe --- /dev/null +++ b/tests/adapter/test_type_descriptor.py @@ -0,0 +1,72 @@ +import unittest + +from amqp.adapter.type_descriptor import TypeDescriptor + + +class TestTypeDescriptor(unittest.TestCase): + def test_annotated_with_json(self): + type_str = ( + "typing.Optional[typing.Annotated[core.base.providers.ingestion.IngestionConfig, Json]]" + ) + 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) + + 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) + + 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) + + self.assertEqual(td.type, "uuid.UUID") + self.assertFalse(td.is_json) + + def test_whitespace_variations(self): + test_cases = [ + ("typing.Optional[typing.Annotated[ str , Json ]]", "str", True), + ("typing.Optional[ typing.Annotated[dict,Json]]", "dict", True), + ("typing.Optional[ int ]", "int", False), + ] + + for type_str, expected_type, expected_json in test_cases: + with self.subTest(type_str=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]", + ] + + for type_str in test_cases: + with self.subTest(type_str=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) + + self.assertEqual(repr(td), "TypeDescriptor(type_string='str', is_json=True)") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/model/test_model.py b/tests/model/test_model.py index b636c6a..155f7d7 100644 --- a/tests/model/test_model.py +++ b/tests/model/test_model.py @@ -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 + # Test default content type_string message_no_content_type = CleverMicroMessage( magic="A", id=self.test_id,