diff --git a/tests/adapter/test_rpc.py b/tests/adapter/test_rpc.py index baf27f9..6c43d9d 100644 --- a/tests/adapter/test_rpc.py +++ b/tests/adapter/test_rpc.py @@ -1,20 +1,16 @@ import importlib import asyncio import sys +import unittest from amqp.service.amq_message_handler import DataMessageHandler -# Create a dummy package and module for demonstration -# You would typically have these in separate files/directories -# e.g., my_package/ -# __init__.py -# my_module.py +class TestRPCInvocation(unittest.IsolatedAsyncioTestCase): + """Test cases for the RPC invocation functionality in DataMessageHandler.""" -# For this self-contained example, we'll create them dynamically in memory -# In a real application, 'my_package' would be a proper installed package or in sys.path - -# --- Dummy Package/Module Setup --- -# Define a simple module string -dummy_module_content = """ + async def asyncSetUp(self): + """Set up the test environment with a dummy module.""" + # Define a simple module string + dummy_module_content = """ import asyncio # Global function @@ -22,7 +18,7 @@ def greet_sync(name: str) -> str: return f"Hello, {name} (sync global)!" async def greet_async(name: str) -> str: - await asyncio.sleep(0.05) # Simulate async work + await asyncio.sleep(0.01) # Simulate async work return f"Hello, {name} (async global)!" # A simple class @@ -40,121 +36,106 @@ class MyService: def _private_method(self): return "This is a private method." """ + # Dynamically create a module object + spec = importlib.util.spec_from_loader("my_package.my_module", loader=None) + if spec is None: + raise ImportError("Could not create module spec for my_package.my_module") -# Dynamically create a module object -spec = importlib.util.spec_from_loader("my_package.my_module", loader=None) -if spec is None: - raise ImportError("Could not create module spec for my_package.my_module") + self.my_module = importlib.util.module_from_spec(spec) + exec(dummy_module_content, self.my_module.__dict__) + sys.modules["my_package"] = self.my_module # Add to sys.modules for importlib to find -my_module = importlib.util.module_from_spec(spec) -exec(dummy_module_content, my_module.__dict__) -sys.modules["my_package"] = my_module # Add to sys.modules for importlib to find + async def asyncTearDown(self): + """Clean up after tests.""" + # Remove the dummy module from sys.modules + if "my_package" in sys.modules: + del sys.modules["my_package"] -# --- Running the Demos --- -async def main(): - print("\n--- Demo: Invoking Global Synchronous Function ---") - try: + async def test_invoke_global_sync_function(self): + """Test invoking a global synchronous function.""" result = await DataMessageHandler.invoke_command( package_name="my_package.my_module", class_name="__global__", function_name="greet_sync", kwargs={"name": "Alice"} ) - print(f"Demo Result: {result}\n") - except Exception as e: - print(f"Demo Failed: {e}\n") + self.assertEqual(result, "Hello, Alice (sync global)!") - print("\n--- Demo: Invoking Global Asynchronous Function ---") - try: + async def test_invoke_global_async_function(self): + """Test invoking a global asynchronous function.""" result = await DataMessageHandler.invoke_command( package_name="my_package.my_module", class_name="__global__", function_name="greet_async", kwargs={"name": "Bob"} ) - print(f"Demo Result: {result}\n") - except Exception as e: - print(f"Demo Failed: {e}\n") + self.assertEqual(result, "Hello, Bob (async global)!") - print("\n--- Demo: Invoking Class Synchronous Method ---") - try: + async def test_invoke_class_sync_method(self): + """Test invoking a class synchronous method.""" result = await DataMessageHandler.invoke_command( package_name="my_package.my_module", class_name="MyService", function_name="process_data_sync", kwargs={"data": "report", "count": 3} ) - print(f"Demo Result: {result}\n") - except Exception as e: - print(f"Demo Failed: {e}\n") + self.assertEqual( + result, + "Service 'MyAwesomeService' processed 'report' 3 times (sync method)." + ) - print("\n--- Demo: Invoking Class Asynchronous Method ---") - try: + async def test_invoke_class_async_method(self): + """Test invoking a class asynchronous method.""" result = await DataMessageHandler.invoke_command( package_name="my_package.my_module", class_name="MyService", function_name="process_data_async", - kwargs={"data": "metrics", "delay": 0.1} + kwargs={"data": "metrics", "delay": 0.01} ) - print(f"Demo Result: {result}\n") - except Exception as e: - print(f"Demo Failed: {e}\n") - - print("\n--- Demo: Error - Package Not Found ---") - try: - await DataMessageHandler.invoke_command( - package_name="non_existent_package", - class_name="__global__", - function_name="some_func", - kwargs={} + self.assertEqual( + result, + "Service 'MyAwesomeService' processed 'metrics' after 0.01s (async method)." ) - except ModuleNotFoundError: - print("Demo Handled: ModuleNotFoundError as expected.\n") - except Exception as e: - print(f"Demo Failed Unexpectedly: {e}\n") - print("\n--- Demo: Error - Class Not Found ---") - try: - await DataMessageHandler.invoke_command( - package_name="my_package.my_module", - class_name="NonExistentClass", - function_name="some_method", - kwargs={} - ) - except AttributeError: - print("Demo Handled: AttributeError (Class Not Found) as expected.\n") - except Exception as e: - print(f"Demo Failed Unexpectedly: {e}\n") + async def test_package_not_found(self): + """Test error handling when package is not found.""" + with self.assertRaises(ModuleNotFoundError): + await DataMessageHandler.invoke_command( + package_name="non_existent_package", + class_name="__global__", + function_name="some_func", + kwargs={} + ) - print("\n--- Demo: Error - Function/Method Not Found ---") - try: - await DataMessageHandler.invoke_command( + async def test_class_not_found(self): + """Test error handling when class is not found.""" + with self.assertRaises(AttributeError): + await DataMessageHandler.invoke_command( + package_name="my_package.my_module", + class_name="NonExistentClass", + function_name="some_method", + kwargs={} + ) + + async def test_method_not_found(self): + """Test error handling when method is not found.""" + with self.assertRaises(AttributeError): + await DataMessageHandler.invoke_command( + package_name="my_package.my_module", + class_name="MyService", + function_name="non_existent_method", + kwargs={} + ) + + async def test_private_method_access(self): + """Test that private methods can be accessed (Python doesn't enforce privacy).""" + result = await DataMessageHandler.invoke_command( package_name="my_package.my_module", class_name="MyService", - function_name="non_existent_method", + function_name="_private_method", kwargs={} ) - except AttributeError: - print("Demo Handled: AttributeError (Function/Method Not Found) as expected.\n") - except Exception as e: - print(f"Demo Failed Unexpectedly: {e}\n") - - print("\n--- Demo: Error - Private Method (AttributeError) ---") - try: - await DataMessageHandler.invoke_command( - package_name="my_package.my_module", - class_name="MyService", - function_name="_private_method", # Python convention, but still accessible via getattr - kwargs={} - ) - print("Demo Result: Successfully called _private_method (Python allows this via getattr).\n") - except Exception as e: - print(f"Demo Failed: {e}\n") - + self.assertEqual(result, "This is a private method.") if __name__ == "__main__": - import sys - import importlib.util - - # Run the main asynchronous demo - asyncio.run(main()) + unittest.main()