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

test: update tests to use invoke_command directly

Browse Source
This commit is contained in:
Stanislav Hejny
2025-07-24 08:16:41 +01:00
committed by Stanislav Hejny (aider)
parent ad49cb2836
commit f3be2fbfea
1 changed files with 10 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/adapter/test_rpc.py
+10 -18
View File
@@ -1,8 +1,8 @@
import importlib import importlib
import asyncio
import sys import sys
import unittest import unittest
from amqp.service.amq_message_handler import DataMessageHandler from amqp.service.amq_message_handler import invoke_command
class TestRPCInvocation(unittest.IsolatedAsyncioTestCase): class TestRPCInvocation(unittest.IsolatedAsyncioTestCase):
"""Test cases for the RPC invocation functionality in DataMessageHandler.""" """Test cases for the RPC invocation functionality in DataMessageHandler."""
@@ -53,8 +53,7 @@ class MyService:
async def test_invoke_global_sync_function(self): async def test_invoke_global_sync_function(self):
"""Test invoking a global synchronous function.""" """Test invoking a global synchronous function."""
result = await DataMessageHandler.invoke_command( result = await invoke_command(
package_name="my_package.my_module",
class_name="__global__", class_name="__global__",
function_name="greet_sync", function_name="greet_sync",
kwargs={"name": "Alice"} kwargs={"name": "Alice"}
@@ -63,8 +62,7 @@ class MyService:
async def test_invoke_global_async_function(self): async def test_invoke_global_async_function(self):
"""Test invoking a global asynchronous function.""" """Test invoking a global asynchronous function."""
result = await DataMessageHandler.invoke_command( result = await invoke_command(
package_name="my_package.my_module",
class_name="__global__", class_name="__global__",
function_name="greet_async", function_name="greet_async",
kwargs={"name": "Bob"} kwargs={"name": "Bob"}
@@ -73,8 +71,7 @@ class MyService:
async def test_invoke_class_sync_method(self): async def test_invoke_class_sync_method(self):
"""Test invoking a class synchronous method.""" """Test invoking a class synchronous method."""
result = await DataMessageHandler.invoke_command( result = await invoke_command(
package_name="my_package.my_module",
class_name="MyService", class_name="MyService",
function_name="process_data_sync", function_name="process_data_sync",
kwargs={"data": "report", "count": 3} kwargs={"data": "report", "count": 3}
@@ -86,8 +83,7 @@ class MyService:
async def test_invoke_class_async_method(self): async def test_invoke_class_async_method(self):
"""Test invoking a class asynchronous method.""" """Test invoking a class asynchronous method."""
result = await DataMessageHandler.invoke_command( result = await invoke_command(
package_name="my_package.my_module",
class_name="MyService", class_name="MyService",
function_name="process_data_async", function_name="process_data_async",
kwargs={"data": "metrics", "delay": 0.01} kwargs={"data": "metrics", "delay": 0.01}
@@ -100,8 +96,7 @@ class MyService:
async def test_package_not_found(self): async def test_package_not_found(self):
"""Test error handling when package is not found.""" """Test error handling when package is not found."""
with self.assertRaises(ModuleNotFoundError): with self.assertRaises(ModuleNotFoundError):
await DataMessageHandler.invoke_command( await invoke_command(
package_name="non_existent_package",
class_name="__global__", class_name="__global__",
function_name="some_func", function_name="some_func",
kwargs={} kwargs={}
@@ -110,8 +105,7 @@ class MyService:
async def test_class_not_found(self): async def test_class_not_found(self):
"""Test error handling when class is not found.""" """Test error handling when class is not found."""
with self.assertRaises(AttributeError): with self.assertRaises(AttributeError):
await DataMessageHandler.invoke_command( await invoke_command(
package_name="my_package.my_module",
class_name="NonExistentClass", class_name="NonExistentClass",
function_name="some_method", function_name="some_method",
kwargs={} kwargs={}
@@ -120,8 +114,7 @@ class MyService:
async def test_method_not_found(self): async def test_method_not_found(self):
"""Test error handling when method is not found.""" """Test error handling when method is not found."""
with self.assertRaises(AttributeError): with self.assertRaises(AttributeError):
await DataMessageHandler.invoke_command( await invoke_command(
package_name="my_package.my_module",
class_name="MyService", class_name="MyService",
function_name="non_existent_method", function_name="non_existent_method",
kwargs={} kwargs={}
@@ -129,8 +122,7 @@ class MyService:
async def test_private_method_access(self): async def test_private_method_access(self):
"""Test that private methods can be accessed (Python doesn't enforce privacy).""" """Test that private methods can be accessed (Python doesn't enforce privacy)."""
result = await DataMessageHandler.invoke_command( result = await invoke_command(
package_name="my_package.my_module",
class_name="MyService", class_name="MyService",
function_name="_private_method", function_name="_private_method",
kwargs={} kwargs={}