Use MagicMocks with async_return_value and pytest.mark.asyncio

Author: rbierbasz-gog

src/galaxy/unittest/mock.py

@@ -3,17 +3,24 @@
 
 
 class AsyncMock(MagicMock):
+    """
+    ..deprecated:: 0.45
+    Use: :class:`MagicMock` with meth:`~.async_return_value`.
+    """
     async def __call__(self, *args, **kwargs):
         return super(AsyncMock, self).__call__(*args, **kwargs)
 
 
 def coroutine_mock():
+    """
+    ..deprecated:: 0.45
+    Use: :class:`MagicMock` with meth:`~.async_return_value`.
+    """
     coro = MagicMock(name="CoroutineResult")
     corofunc = MagicMock(name="CoroutineFunction", side_effect=asyncio.coroutine(coro))
     corofunc.coro = coro
     return corofunc
 
-
 async def skip_loop(iterations=1):
     for _ in range(iterations):
         await asyncio.sleep(0)

tests/__init__.py

@@ -16,3 +16,4 @@ def get_messages(write_mock):
             message = json.loads(line)
             messages.append(message)
     return messages
+

tests/conftest.py

@@ -6,21 +6,19 @@
 
 from galaxy.api.plugin import Plugin
 from galaxy.api.consts import Platform
-from galaxy.unittest.mock import AsyncMock, coroutine_mock, skip_loop
 
 @pytest.fixture()
 def reader():
     stream = MagicMock(name="stream_reader")
-    stream.read = AsyncMock()
+    stream.read = MagicMock()
     yield stream
 
 @pytest.fixture()
 async def writer():
     stream = MagicMock(name="stream_writer")
     stream.write = MagicMock()
-    stream.drain = AsyncMock(return_value=None)
+    stream.drain = MagicMock()
     yield stream
-    await skip_loop(1) # drain
 
 @pytest.fixture()
 def read(reader):
@@ -33,7 +31,7 @@ def write(writer):
 @pytest.fixture()
 def plugin(reader, writer):
     """Return plugin instance with all feature methods mocked"""
-    async_methods = (
+    methods = (
         "handshake_complete",
         "authenticate",
         "get_owned_games",
@@ -46,17 +44,12 @@ def plugin(reader, writer):
         "get_friends",
         "get_game_time",
         "prepare_game_times_context",
-        "shutdown_platform_client"
-    )
-
-    methods = (
+        "shutdown_platform_client",
         "shutdown",
         "tick"
     )
 
     with ExitStack() as stack:
-        for method in async_methods:
-            stack.enter_context(patch.object(Plugin, method, new_callable=coroutine_mock))
         for method in methods:
             stack.enter_context(patch.object(Plugin, method))
         yield Plugin(Platform.Generic, "0.1", reader, writer, "token")

tests/test_achievements.py

@@ -1,5 +1,4 @@
 import json
-from unittest.mock import MagicMock
 
 import pytest
 from pytest import raises
@@ -21,17 +20,9 @@ def test_initialization_no_id_nor_name():
         Achievement(unlock_time=1234567890)
 
 
-# TODO replace AsyncMocks with MagicMocks in conftest and use async_return_value
-@pytest.fixture()
-def reader():
-    stream = MagicMock(name="stream_reader")
-    stream.read = MagicMock()
-    yield stream
-
-
 @pytest.mark.asyncio
 async def test_get_unlocked_achievements_success(plugin, read, write):
-    plugin.prepare_achievements_context.coro.return_value = 5
+    plugin.prepare_achievements_context.return_value = async_return_value(5)
     request = {
         "jsonrpc": "2.0",
         "id": "3",
@@ -41,11 +32,11 @@ async def test_get_unlocked_achievements_success(plugin, read, write):
         }
     }
     read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
-    plugin.get_unlocked_achievements.coro.return_value = [
+    plugin.get_unlocked_achievements.return_value = async_return_value([
         Achievement(achievement_id="lvl10", unlock_time=1548421241),
         Achievement(achievement_name="Got level 20", unlock_time=1548422395),
         Achievement(achievement_id="lvl30", achievement_name="Got level 30", unlock_time=1548495633)
-    ]
+    ])
     await plugin.run()
     plugin.prepare_achievements_context.assert_called_with(["14"])
     plugin.get_unlocked_achievements.assert_called_with("14", 5)
@@ -92,6 +83,7 @@ async def test_get_unlocked_achievements_success(plugin, read, write):
     (KeyError, 0, "Unknown error")
 ])
 async def test_get_unlocked_achievements_error(exception, code, message, plugin, read, write):
+    plugin.prepare_achievements_context.return_value = async_return_value(None)
     request = {
         "jsonrpc": "2.0",
         "id": "3",
@@ -102,7 +94,7 @@ async def test_get_unlocked_achievements_error(exception, code, message, plugin,
     }
 
     read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
-    plugin.get_unlocked_achievements.coro.side_effect = exception
+    plugin.get_unlocked_achievements.side_effect = exception
     await plugin.run()
     plugin.get_unlocked_achievements.assert_called()
 
@@ -132,7 +124,7 @@ async def test_get_unlocked_achievements_error(exception, code, message, plugin,
 
 @pytest.mark.asyncio
 async def test_prepare_get_unlocked_achievements_context_error(plugin, read, write):
-    plugin.prepare_achievements_context.coro.side_effect = BackendError()
+    plugin.prepare_achievements_context.side_effect = BackendError()
     request = {
         "jsonrpc": "2.0",
         "id": "3",
@@ -158,6 +150,7 @@ async def test_prepare_get_unlocked_achievements_context_error(plugin, read, wri
 
 @pytest.mark.asyncio
 async def test_import_in_progress(plugin, read, write):
+    plugin.prepare_achievements_context.return_value = async_return_value(None)
     requests = [
         {
             "jsonrpc": "2.0",

tests/test_authenticate.py

@@ -1,36 +1,40 @@
-import asyncio
-import json
-
 import pytest
 
 from galaxy.api.types import Authentication
 from galaxy.api.errors import (
     UnknownError, InvalidCredentials, NetworkError, LoggedInElsewhere, ProtocolError,
     BackendNotAvailable, BackendTimeout, BackendError, TemporaryBlocked, Banned, AccessDenied
 )
+from galaxy.unittest.mock import async_return_value
+
+from tests import create_message, get_messages
+
 
-def test_success(plugin, read, write):
+@pytest.mark.asyncio
+async def test_success(plugin, read, write):
     request = {
         "jsonrpc": "2.0",
         "id": "3",
         "method": "init_authentication"
     }
-
-    read.side_effect = [json.dumps(request).encode() + b"\n", b""]
-    plugin.authenticate.coro.return_value = Authentication("132", "Zenek")
-    asyncio.run(plugin.run())
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    plugin.authenticate.return_value = async_return_value(Authentication("132", "Zenek"))
+    await plugin.run()
     plugin.authenticate.assert_called_with()
-    response = json.loads(write.call_args[0][0])
 
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "3",
-        "result": {
-            "user_id": "132",
-            "user_name": "Zenek"
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "result": {
+                "user_id": "132",
+                "user_name": "Zenek"
+            }
         }
-    }
+    ]
+
 
+@pytest.mark.asyncio
 @pytest.mark.parametrize("error,code,message", [
     pytest.param(UnknownError, 0, "Unknown error", id="unknown_error"),
     pytest.param(BackendNotAvailable, 2, "Backend not available", id="backend_not_available"),
@@ -44,29 +48,32 @@ def test_success(plugin, read, write):
     pytest.param(Banned, 105, "Banned", id="banned"),
     pytest.param(AccessDenied, 106, "Access denied", id="access_denied"),
 ])
-def test_failure(plugin, read, write, error, code, message):
+async def test_failure(plugin, read, write, error, code, message):
     request = {
         "jsonrpc": "2.0",
         "id": "3",
         "method": "init_authentication"
     }
 
-    read.side_effect = [json.dumps(request).encode() + b"\n", b""]
-    plugin.authenticate.coro.side_effect = error()
-    asyncio.run(plugin.run())
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    plugin.authenticate.side_effect = error()
+    await plugin.run()
     plugin.authenticate.assert_called_with()
-    response = json.loads(write.call_args[0][0])
 
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "3",
-        "error": {
-            "code": code,
-            "message": message
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "error": {
+                "code": code,
+                "message": message
+            }
         }
-    }
+    ]
+
 
-def test_stored_credentials(plugin, read, write):
+@pytest.mark.asyncio
+async def test_stored_credentials(plugin, read, write):
     request = {
         "jsonrpc": "2.0",
         "id": "3",
@@ -77,39 +84,37 @@ def test_stored_credentials(plugin, read, write):
             }
         }
     }
-    read.side_effect = [json.dumps(request).encode() + b"\n", b""]
-    plugin.authenticate.coro.return_value = Authentication("132", "Zenek")
-    asyncio.run(plugin.run())
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    plugin.authenticate.return_value = async_return_value(Authentication("132", "Zenek"))
+    await plugin.run()
     plugin.authenticate.assert_called_with(stored_credentials={"token": "ABC"})
     write.assert_called()
 
-def test_store_credentials(plugin, write):
+
+@pytest.mark.asyncio
+async def test_store_credentials(plugin, write):
     credentials = {
         "token": "ABC"
     }
+    plugin.store_credentials(credentials)
 
-    async def couritine():
-        plugin.store_credentials(credentials)
-
-    asyncio.run(couritine())
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "method": "store_credentials",
-        "params": credentials
-    }
-
-def test_lost_authentication(plugin, write):
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "method": "store_credentials",
+            "params": credentials
+        }
+    ]
 
-    async def couritine():
-        plugin.lost_authentication()
 
-    asyncio.run(couritine())
-    response = json.loads(write.call_args[0][0])
+@pytest.mark.asyncio
+async def test_lost_authentication(plugin, write):
+    plugin.lost_authentication()
 
-    assert response == {
-        "jsonrpc": "2.0",
-        "method": "authentication_lost",
-        "params": None
-    }
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "method": "authentication_lost",
+            "params": None
+        }
+    ]

tests/test_chunk_messages.py

@@ -1,7 +1,12 @@
-import asyncio
 import json
 
-def test_chunked_messages(plugin, read):
+import pytest
+
+from galaxy.unittest.mock import async_return_value
+
+
+@pytest.mark.asyncio
+async def test_chunked_messages(plugin, read):
     request = {
         "jsonrpc": "2.0",
         "method": "install_game",
@@ -11,11 +16,13 @@ def test_chunked_messages(plugin, read):
     }
 
     message = json.dumps(request).encode() + b"\n"
-    read.side_effect = [message[:5], message[5:], b""]
-    asyncio.run(plugin.run())
+    read.side_effect = [async_return_value(message[:5]), async_return_value(message[5:]), async_return_value(b"")]
+    await plugin.run()
     plugin.install_game.assert_called_with(game_id="3")
 
-def test_joined_messages(plugin, read):
+
+@pytest.mark.asyncio
+async def test_joined_messages(plugin, read):
     requests = [
         {
             "jsonrpc": "2.0",
@@ -34,12 +41,14 @@ def test_joined_messages(plugin, read):
     ]
     data = b"".join([json.dumps(request).encode() + b"\n" for request in requests])
 
-    read.side_effect = [data, b""]
-    asyncio.run(plugin.run())
+    read.side_effect = [async_return_value(data), async_return_value(b"")]
+    await plugin.run()
     plugin.install_game.assert_called_with(game_id="3")
     plugin.launch_game.assert_called_with(game_id="3")
 
-def test_not_finished(plugin, read):
+
+@pytest.mark.asyncio
+async def test_not_finished(plugin, read):
     request = {
         "jsonrpc": "2.0",
         "method": "install_game",
@@ -49,6 +58,6 @@ def test_not_finished(plugin, read):
     }
 
     message = json.dumps(request).encode() # no new line
-    read.side_effect = [message, b""]
-    asyncio.run(plugin.run())
+    read.side_effect = [async_return_value(message), async_return_value(b"")]
+    await plugin.run()
     plugin.install_game.assert_not_called()