amq-adapter-python/tests/adapter/test_rpc.py

import importlib
import sys
import unittest

from amqp.service.amq_message_handler import invoke_command


class TestRPCInvocation(unittest.IsolatedAsyncioTestCase):
    """Test cases for the RPC invocation functionality in DataMessageHandler."""

    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
def greet_sync(name: str) -> str:
    return f"Hello, {name} (sync global)!"

async def greet_async(name: str) -> str:
    await asyncio.sleep(0.01) # Simulate async work
    return f"Hello, {name} (async global)!"

# A simple class
class MyService:
    def __init__(self):
        self.service_name = "MyAwesomeService"

    def process_data_sync(self, data: str, count: int) -> str:
        return f"Service '{self.service_name}' processed '{data}' {count} times (sync method)."

    async def process_data_async(self, data: str, delay: float) -> str:
        await asyncio.sleep(delay) # Simulate async work
        return f"Service '{self.service_name}' processed '{data}' after {delay}s (async method)."

    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")

        self.my_module = importlib.util.module_from_spec(spec)
        exec(dummy_module_content, self.my_module.__dict__)
        sys.modules["my_package.my_module"] = (
            self.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"]

    async def test_invoke_global_sync_function(self):
        """Test invoking a global synchronous function."""
        result = await invoke_command(
            package_name="my_package.my_module",
            class_name="__global__",
            function_name="greet_sync",
            kwargs={"name": "Alice"},
        )
        self.assertEqual(result.body().decode(), "Hello, Alice (sync global)!")

    async def test_invoke_global_async_function(self):
        """Test invoking a global asynchronous function."""
        result = await invoke_command(
            package_name="my_package.my_module",
            class_name="__global__",
            function_name="greet_async",
            kwargs={"name": "Bob"},
        )
        self.assertEqual(result.body().decode(), "Hello, Bob (async global)!")

    async def test_invoke_class_sync_method(self):
        """Test invoking a class synchronous method."""
        result = await invoke_command(
            package_name="my_package.my_module",
            class_name="MyService",
            function_name="process_data_sync",
            kwargs={"data": "report", "count": 3},
        )
        self.assertEqual(
            result.body().decode(),
            "Service 'MyAwesomeService' processed 'report' 3 times (sync method).",
        )

    async def test_invoke_class_async_method(self):
        """Test invoking a class asynchronous method."""
        result = await invoke_command(
            package_name="my_package.my_module",
            class_name="MyService",
            function_name="process_data_async",
            kwargs={"data": "metrics", "delay": 0.01},
        )
        self.assertEqual(
            result.body().decode(),
            "Service 'MyAwesomeService' processed 'metrics' after 0.01s (async method).",
        )

    async def test_package_not_found(self):
        """Test error handling when package is not found."""
        with self.assertRaises(ModuleNotFoundError):
            await invoke_command(
                package_name="non_existent_package",
                class_name="__global__",
                function_name="some_func",
                kwargs={},
            )

    async def test_class_not_found(self):
        """Test error handling when class is not found."""
        with self.assertRaises(AttributeError):
            await 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 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 invoke_command(
            package_name="my_package.my_module",
            class_name="MyService",
            function_name="_private_method",
            kwargs={},
        )
        self.assertEqual(result.body().decode(), "This is a private method.")


if __name__ == "__main__":
    unittest.main()