test: convert demo script to unittest framework in test_rpc.py

Co-authored-by: aider (openrouter/openai/o3-mini-high) <aider@aider.chat>

tests/adapter/test_rpc.py

@@ -1,18 +1,14 @@
 import importlib
 import asyncio
 import sys
+import unittest
 from amqp.service.amq_message_handler import DataMessageHandler
 
-# Create a dummy package and module for demonstration
+class TestRPCInvocation(unittest.IsolatedAsyncioTestCase):
-# You would typically have these in separate files/directories
+    """Test cases for the RPC invocation functionality in DataMessageHandler."""
-# e.g., my_package/
-#           __init__.py
-#           my_module.py
 
-# For this self-contained example, we'll create them dynamically in memory
+    async def asyncSetUp(self):
-# In a real application, 'my_package' would be a proper installed package or in sys.path
+        """Set up the test environment with a dummy module."""
-
-# --- Dummy Package/Module Setup ---
         # Define a simple module string
         dummy_module_content = """
 import asyncio
@@ -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")
 
-my_module = importlib.util.module_from_spec(spec)
+        self.my_module = importlib.util.module_from_spec(spec)
-exec(dummy_module_content, my_module.__dict__)
+        exec(dummy_module_content, self.my_module.__dict__)
-sys.modules["my_package"] = my_module # Add to sys.modules for importlib to find
+        sys.modules["my_package"] = self.my_module  # Add to sys.modules for importlib to find
 
-# --- Running the Demos ---
+    async def asyncTearDown(self):
-async def main():
+        """Clean up after tests."""
-    print("\n--- Demo: Invoking Global Synchronous Function ---")
+        # Remove the dummy module from sys.modules
-    try:
+        if "my_package" in sys.modules:
+            del sys.modules["my_package"]
+
+    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")
+        self.assertEqual(result, "Hello, Alice (sync global)!")
-    except Exception as e:
-        print(f"Demo Failed: {e}\n")
 
-    print("\n--- Demo: Invoking Global Asynchronous Function ---")
+    async def test_invoke_global_async_function(self):
-    try:
+        """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")
+        self.assertEqual(result, "Hello, Bob (async global)!")
-    except Exception as e:
-        print(f"Demo Failed: {e}\n")
 
-    print("\n--- Demo: Invoking Class Synchronous Method ---")
+    async def test_invoke_class_sync_method(self):
-    try:
+        """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")
+        self.assertEqual(
-    except Exception as e:
+            result,
-        print(f"Demo Failed: {e}\n")
+            "Service 'MyAwesomeService' processed 'report' 3 times (sync method)."
+        )
 
-    print("\n--- Demo: Invoking Class Asynchronous Method ---")
+    async def test_invoke_class_async_method(self):
-    try:
+        """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}
+        )
+        self.assertEqual(
+            result,
+            "Service 'MyAwesomeService' processed 'metrics' after 0.01s (async method)."
         )
-        print(f"Demo Result: {result}\n")
-    except Exception as e:
-        print(f"Demo Failed: {e}\n")
 
-    print("\n--- Demo: Error - Package Not Found ---")
+    async def test_package_not_found(self):
-    try:
+        """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={}
             )
-    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 ---")
+    async def test_class_not_found(self):
-    try:
+        """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={}
             )
-    except AttributeError:
-        print("Demo Handled: AttributeError (Class Not Found) as expected.\n")
-    except Exception as e:
-        print(f"Demo Failed Unexpectedly: {e}\n")
 
-    print("\n--- Demo: Error - Function/Method Not Found ---")
+    async def test_method_not_found(self):
-    try:
+        """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={}
             )
-    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) ---")
+    async def test_private_method_access(self):
-    try:
+        """Test that private methods can be accessed (Python doesn't enforce privacy)."""
-        await DataMessageHandler.invoke_command(
+        result = await DataMessageHandler.invoke_command(
             package_name="my_package.my_module",
             class_name="MyService",
-            function_name="_private_method", # Python convention, but still accessible via getattr
+            function_name="_private_method",
             kwargs={}
         )
-        print("Demo Result: Successfully called _private_method (Python allows this via getattr).\n")
+        self.assertEqual(result, "This is a private method.")
-    except Exception as e:
-        print(f"Demo Failed: {e}\n")
-
 
 if __name__ == "__main__":
-    import sys
+    unittest.main()
-    import importlib.util
-
-    # Run the main asynchronous demo
-    asyncio.run(main())