adding a Python3 fix to JsonRPC serializer

aio_pika/patterns/rpc.py

@@ -5,6 +5,7 @@
 import time
 from enum import Enum
 from functools import partial
+from pydoc import locate
 from typing import Any, Callable, Dict, Hashable, Optional, TypeVar
 
 from aiormq.tools import awaitable
@@ -404,7 +405,15 @@ class JsonRPC(RPC):
     CONTENT_TYPE = "application/json"
 
     def serialize(self, data: Any) -> bytes:
-        return self.SERIALIZER.dumps(data, ensure_ascii=False, default=repr)
+        return self.SERIALIZER.dumps(
+            data, ensure_ascii=False, default=repr).encode('ascii')
+
+    def deserialize(self, data: Any) -> bytes:
+        res = super().deserialize(data)
+        if "error" in res:
+            cls = locate(res['error']['type'])
+            res = cls(res['error']['message'], res['error']['args'])
+        return res
 
     def serialize_exception(self, exception: Exception) -> bytes:
         return self.serialize(

tests/test_rpc.py

@@ -7,7 +7,7 @@
 from aio_pika import Message
 from aio_pika.exceptions import DeliveryError
 from aio_pika.message import IncomingMessage
-from aio_pika.patterns.rpc import RPC
+from aio_pika.patterns.rpc import RPC, JsonRPC
 from aio_pika.patterns.rpc import log as rpc_logger
 from tests import get_random_name
 
@@ -18,6 +18,8 @@ def rpc_func(*, foo, bar):
 
     return {"foo": "bar"}
 
+def rpc_raise_exception(*, foo, bar):
+    raise Exception('foo bar')
 
 class TestCase:
     async def test_simple(self, channel: aio_pika.Channel):
@@ -172,3 +174,39 @@ async def test_register_twice(self, channel: aio_pika.Channel):
         await rpc.unregister(rpc_func)
 
         await rpc.close()
+
+    async def test_jsonrpc_simple(self, channel: aio_pika.Channel):
+        rpc = await JsonRPC.create(channel, auto_delete=True)
+
+        await rpc.register("test.rpc", rpc_func, auto_delete=True)
+
+        result = await rpc.proxy.test.rpc(foo=None, bar=None)
+        assert result == {"foo": "bar"}
+
+        await rpc.unregister(rpc_func)
+        await rpc.close()
+
+        # Close already closed
+        await rpc.close()
+
+    async def test_jsonrpc_assert(self, channel: aio_pika.Channel):
+        rpc = await JsonRPC.create(channel, auto_delete=True)
+
+        await rpc.register("test.rpc", rpc_func, auto_delete=True)
+
+        with pytest.raises(AssertionError):
+            await rpc.proxy.test.rpc(foo=True, bar=None)
+
+        await rpc.unregister(rpc_func)
+        await rpc.close()
+
+    async def test_jsonrpc_error(self, channel: aio_pika.Channel):
+        rpc = await JsonRPC.create(channel, auto_delete=True)
+
+        await rpc.register("test.rpc_error", rpc_raise_exception, auto_delete=True)
+
+        with pytest.raises(AssertionError):
+            await rpc.proxy.test.rpc_error(foo=True, bar=None)
+
+        await rpc.unregister(rpc_raise_exception)
+        await rpc.close()