Increment version

setup.py

@@ -2,7 +2,7 @@ from setuptools import setup, find_packages
 
 setup(
     name="galaxy.plugin.api",
-    version="0.44",
+    version="0.45",
     description="GOG Galaxy Integrations Python API",
     author='Galaxy team',
     author_email='galaxy@gog.com',

src/galaxy/api/plugin.py

@@ -89,16 +89,9 @@ class Plugin:
         )
         self._detect_feature(Feature.ImportOwnedGames, ["get_owned_games"])
 
-        self._register_method(
-            "import_unlocked_achievements",
-            self.get_unlocked_achievements,
-            result_name="unlocked_achievements"
-        )
+        self._register_method("start_achievements_import", self._start_achievements_import)
         self._detect_feature(Feature.ImportAchievements, ["get_unlocked_achievements"])
 
-        self._register_method("start_achievements_import", self.start_achievements_import)
-        self._detect_feature(Feature.ImportAchievements, ["import_games_achievements"])
-
         self._register_method("import_local_games", self.get_local_games, result_name="local_games")
         self._detect_feature(Feature.ImportInstalledGames, ["get_local_games"])
 
@@ -117,11 +110,8 @@ class Plugin:
         self._register_method("import_friends", self.get_friends, result_name="friend_info_list")
         self._detect_feature(Feature.ImportFriends, ["get_friends"])
 
-        self._register_method("import_game_times", self.get_game_times, result_name="game_times")
-        self._detect_feature(Feature.ImportGameTime, ["get_game_times"])
-
-        self._register_method("start_game_times_import", self.start_game_times_import)
-        self._detect_feature(Feature.ImportGameTime, ["import_game_times"])
+        self._register_method("start_game_times_import", self._start_game_times_import)
+        self._detect_feature(Feature.ImportGameTime, ["get_game_time"])
 
     @property
     def features(self) -> List[Feature]:
@@ -218,7 +208,10 @@ class Plugin:
 
     def _initialize_cache(self, data: Dict):
         self._persistent_cache = data
-        self.handshake_complete()
+        try:
+            self.handshake_complete()
+        except Exception:
+            logging.exception("Unhandled exception during `handshake_complete` step")
         self._pass_control_task = asyncio.create_task(self._pass_control())
 
     @staticmethod
@@ -316,26 +309,14 @@ class Plugin:
         }
         self._notification_client.notify("achievement_unlocked", params)
 
-    def game_achievements_import_success(self, game_id: str, achievements: List[Achievement]) -> None:
-        """Notify the client that import of achievements for a given game has succeeded.
-        This method is called by import_games_achievements.
-
-        :param game_id: id of the game for which the achievements were imported
-        :param achievements: list of imported achievements
-        """
+    def _game_achievements_import_success(self, game_id: str, achievements: List[Achievement]) -> None:
         params = {
             "game_id": game_id,
             "unlocked_achievements": achievements
         }
         self._notification_client.notify("game_achievements_import_success", params)
 
-    def game_achievements_import_failure(self, game_id: str, error: ApplicationError) -> None:
-        """Notify the client that import of achievements for a given game has failed.
-        This method is called by import_games_achievements.
-
-        :param game_id: id of the game for which the achievements import failed
-        :param error: error which prevented the achievements import
-        """
+    def _game_achievements_import_failure(self, game_id: str, error: ApplicationError) -> None:
         params = {
             "game_id": game_id,
             "error": {
@@ -345,9 +326,7 @@ class Plugin:
         }
         self._notification_client.notify("game_achievements_import_failure", params)
 
-    def achievements_import_finished(self) -> None:
-        """Notify the client that importing achievements has finished.
-        This method is called by import_games_achievements_task"""
+    def _achievements_import_finished(self) -> None:
         self._notification_client.notify("achievements_import_finished", None)
 
     def update_local_game_status(self, local_game: LocalGame) -> None:
@@ -367,7 +346,7 @@ class Plugin:
                         continue
                     self.update_local_game_status(LocalGame(game.id, game.status))
                     self._cached_games_statuses[game.id] = game.status
-                asyncio.sleep(5)  # interval
+                await asyncio.sleep(5)  # interval
 
             def tick(self):
                 if self._check_statuses_task is None or self._check_statuses_task.done():
@@ -400,22 +379,11 @@ class Plugin:
         params = {"game_time": game_time}
         self._notification_client.notify("game_time_updated", params)
 
-    def game_time_import_success(self, game_time: GameTime) -> None:
-        """Notify the client that import of a given game_time has succeeded.
-        This method is called by import_game_times.
-
-        :param game_time: game_time which was imported
-        """
+    def _game_time_import_success(self, game_time: GameTime) -> None:
         params = {"game_time": game_time}
         self._notification_client.notify("game_time_import_success", params)
 
-    def game_time_import_failure(self, game_id: str, error: ApplicationError) -> None:
-        """Notify the client that import of a game time for a given game has failed.
-        This method is called by import_game_times.
-
-        :param game_id: id of the game for which the game time could not be imported
-        :param error:   error which prevented the game time import
-        """
+    def _game_time_import_failure(self, game_id: str, error: ApplicationError) -> None:
         params = {
             "game_id": game_id,
             "error": {
@@ -425,10 +393,7 @@ class Plugin:
         }
         self._notification_client.notify("game_time_import_failure", params)
 
-    def game_times_import_finished(self) -> None:
-        """Notify the client that importing game times has finished.
-        This method is called by :meth:`~.import_game_times_task`.
-        """
+    def _game_times_import_finished(self) -> None:
         self._notification_client.notify("game_times_import_finished", None)
 
     def lost_authentication(self) -> None:
@@ -557,51 +522,51 @@ class Plugin:
         """
         raise NotImplementedError()
 
-    async def get_unlocked_achievements(self, game_id: str) -> List[Achievement]:
-        """
-        .. deprecated:: 0.33
-            Use :meth:`~.import_games_achievements`.
-        """
-        raise NotImplementedError()
-
-    async def start_achievements_import(self, game_ids: List[str]) -> None:
-        """Starts the task of importing achievements.
-        This method is called by the GOG Galaxy Client.
-
-        :param game_ids: ids of the games for which the achievements are imported
-        """
+    async def _start_achievements_import(self, game_ids: List[str]) -> None:
         if self._achievements_import_in_progress:
             raise ImportInProgress()
 
-        async def import_games_achievements_task(game_ids):
+        context = await self.prepare_achievements_context(game_ids)
+
+        async def import_game_achievements(game_id, context_):
             try:
-                await self.import_games_achievements(game_ids)
+                achievements = await self.get_unlocked_achievements(game_id, context_)
+                self._game_achievements_import_success(game_id, achievements)
+            except ApplicationError as error:
+                self._game_achievements_import_failure(game_id, error)
+            except Exception:
+                logging.exception("Unexpected exception raised in import_game_achievements")
+                self._game_achievements_import_failure(game_id, UnknownError())
+
+        async def import_games_achievements(game_ids_, context_):
+            try:
+                imports = [import_game_achievements(game_id, context_) for game_id in game_ids_]
+                await asyncio.gather(*imports)
             finally:
-                self.achievements_import_finished()
+                self._achievements_import_finished()
                 self._achievements_import_in_progress = False
 
-        asyncio.create_task(import_games_achievements_task(game_ids))
+        self.create_task(import_games_achievements(game_ids, context), "Games unlocked achievements import")
         self._achievements_import_in_progress = True
 
-    async def import_games_achievements(self, game_ids: List[str]) -> None:
+    async def prepare_achievements_context(self, game_ids: List[str]) -> Any:
+        """Override this method to prepare context for get_unlocked_achievements.
+
+        This allows for optimizations like batch requests to platform API.
+        Default implementation returns None.
         """
-        Override this method to return the unlocked achievements
-        of the user that is currently logged in to the plugin.
-        Call game_achievements_import_success/game_achievements_import_failure for each game_id on the list.
-        This method is called by the GOG Galaxy Client.
+        return None
 
-        :param game_ids: ids of the games for which to import unlocked achievements
+    async def get_unlocked_achievements(self, game_id: str, context: Any) -> List[Achievement]:
+        """Override this method to return list of unlocked achievements
+        for the game identified by the provided game_id.
+        This method is called by import task initialized by GOG Galaxy Client.
+
+        :param game_id:
+        :param context: Value return from :meth:`prepare_achievements_context`
+        :return:
         """
-
-        async def import_game_achievements(game_id):
-            try:
-                achievements = await self.get_unlocked_achievements(game_id)
-                self.game_achievements_import_success(game_id, achievements)
-            except Exception as error:
-                self.game_achievements_import_failure(game_id, error)
-
-        imports = [import_game_achievements(game_id) for game_id in game_ids]
-        await asyncio.gather(*imports)
+        raise NotImplementedError()
 
     async def get_local_games(self) -> List[LocalGame]:
         """Override this method to return the list of
@@ -704,54 +669,50 @@ class Plugin:
         """
         raise NotImplementedError()
 
-    async def get_game_times(self) -> List[GameTime]:
-        """
-        .. deprecated:: 0.33
-            Use :meth:`~.import_game_times`.
-        """
-        raise NotImplementedError()
-
-    async def start_game_times_import(self, game_ids: List[str]) -> None:
-        """Starts the task of importing game times
-        This method is called by the GOG Galaxy Client.
-
-        :param game_ids: ids of the games for which the game time is imported
-        """
+    async def _start_game_times_import(self, game_ids: List[str]) -> None:
         if self._game_times_import_in_progress:
             raise ImportInProgress()
 
-        async def import_game_times_task(game_ids):
+        context = await self.prepare_game_times_context(game_ids)
+
+        async def import_game_time(game_id, context_):
             try:
-                await self.import_game_times(game_ids)
+                game_time = await self.get_game_time(game_id, context_)
+                self._game_time_import_success(game_time)
+            except ApplicationError as error:
+                self._game_time_import_failure(game_id, error)
+            except Exception:
+                logging.exception("Unexpected exception raised in import_game_time")
+                self._game_time_import_failure(game_id, UnknownError())
+
+        async def import_game_times(game_ids_, context_):
+            try:
+                imports = [import_game_time(game_id, context_) for game_id in game_ids_]
+                await asyncio.gather(*imports)
             finally:
-                self.game_times_import_finished()
+                self._game_times_import_finished()
                 self._game_times_import_in_progress = False
 
-        asyncio.create_task(import_game_times_task(game_ids))
+        self.create_task(import_game_times(game_ids, context), "Game times import")
         self._game_times_import_in_progress = True
 
-    async def import_game_times(self, game_ids: List[str]) -> None:
+    async def prepare_game_times_context(self, game_ids: List[str]) -> Any:
+        """Override this method to prepare context for get_game_time.
+        This allows for optimizations like batch requests to platform API.
+        Default implementation returns None.
         """
-        Override this method to return game times for
-        games owned by the currently authenticated user.
-        Call game_time_import_success/game_time_import_failure for each game_id on the list.
-        This method is called by GOG Galaxy Client.
+        return None
 
-        :param game_ids: ids of the games for which the game time is imported
+    async def get_game_time(self, game_id: str, context: Any) -> GameTime:
+        """Override this method to return the game time for the game
+        identified by the provided game_id.
+        This method is called by import task initialized by GOG Galaxy Client.
+
+        :param game_id:
+        :param context: Value return from :meth:`prepare_game_times_context`
+        :return:
         """
-        try:
-            game_times = await self.get_game_times()
-            game_ids_set = set(game_ids)
-            for game_time in game_times:
-                if game_time.game_id not in game_ids_set:
-                    continue
-                self.game_time_import_success(game_time)
-                game_ids_set.discard(game_time.game_id)
-            for game_id in game_ids_set:
-                self.game_time_import_failure(game_id, UnknownError())
-        except Exception as error:
-            for game_id in game_ids:
-                self.game_time_import_failure(game_id, error)
+        raise NotImplementedError()
 
 
 def create_and_run_plugin(plugin_class, argv):

src/galaxy/unittest/mock.py

@@ -1,12 +1,31 @@
-from asyncio import coroutine
+import asyncio
 from unittest.mock import MagicMock
 
+
 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=coroutine(coro))
+    corofunc = MagicMock(name="CoroutineFunction", side_effect=asyncio.coroutine(coro))
     corofunc.coro = coro
-    return corofunc
+    return corofunc
+
+async def skip_loop(iterations=1):
+    for _ in range(iterations):
+        await asyncio.sleep(0)
+
+
+async def async_return_value(return_value, loop_iterations_delay=0):
+    await skip_loop(loop_iterations_delay)
+    return return_value

tests/__init__.py

@@ -0,0 +1,19 @@
+import json
+
+
+def create_message(request):
+    return json.dumps(request).encode() + b"\n"
+
+
+def get_messages(write_mock):
+    messages = []
+    print("call_args_list", write_mock.call_args_list)
+    for call_args in write_mock.call_args_list:
+        print("call_args", call_args)
+        data = call_args[0][0]
+        print("data", data)
+        for line in data.splitlines():
+            message = json.loads(line)
+            messages.append(message)
+    return messages
+

tests/conftest.py

@@ -6,19 +6,18 @@ import pytest
 
 from galaxy.api.plugin import Plugin
 from galaxy.api.consts import Platform
-from galaxy.unittest.mock import AsyncMock, coroutine_mock
 
 @pytest.fixture()
 def reader():
     stream = MagicMock(name="stream_reader")
-    stream.read = AsyncMock()
+    stream.read = MagicMock()
     yield stream
 
 @pytest.fixture()
-def writer():
+async def writer():
     stream = MagicMock(name="stream_writer")
     stream.write = MagicMock()
-    stream.drain = AsyncMock()
+    stream.drain = MagicMock()
     yield stream
 
 @pytest.fixture()
@@ -32,28 +31,25 @@ 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",
+        "prepare_achievements_context",
         "get_unlocked_achievements",
         "get_local_games",
         "launch_game",
         "install_game",
         "uninstall_game",
         "get_friends",
-        "get_game_times",
-        "shutdown_platform_client"
-    )
-
-    methods = (
+        "get_game_time",
+        "prepare_game_times_context",
+        "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,94 +1,203 @@
-import asyncio
 import json
-from unittest.mock import call
 
 import pytest
 from pytest import raises
 
 from galaxy.api.types import Achievement
-from galaxy.api.errors import UnknownError, ImportInProgress, BackendError
+from galaxy.api.errors import BackendError
+from galaxy.unittest.mock import async_return_value
+
+from tests import create_message, get_messages
+
 
 def test_initialization_no_unlock_time():
     with raises(Exception):
         Achievement(achievement_id="lvl30", achievement_name="Got level 30")
 
+
 def test_initialization_no_id_nor_name():
     with raises(AssertionError):
         Achievement(unlock_time=1234567890)
 
-def test_success(plugin, read, write):
+
+@pytest.mark.asyncio
+async def test_get_unlocked_achievements_success(plugin, read, write):
+    plugin.prepare_achievements_context.return_value = async_return_value(5)
     request = {
         "jsonrpc": "2.0",
         "id": "3",
-        "method": "import_unlocked_achievements",
+        "method": "start_achievements_import",
         "params": {
-            "game_id": "14"
+            "game_ids": ["14"]
         }
     }
-    read.side_effect = [json.dumps(request).encode() + b"\n", b""]
-    plugin.get_unlocked_achievements.coro.return_value = [
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+    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)
-    ]
-    asyncio.run(plugin.run())
-    plugin.get_unlocked_achievements.assert_called_with(game_id="14")
-    response = json.loads(write.call_args[0][0])
+    ])
+    await plugin.run()
+    plugin.prepare_achievements_context.assert_called_with(["14"])
+    plugin.get_unlocked_achievements.assert_called_with("14", 5)
 
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "3",
-        "result": {
-            "unlocked_achievements": [
-                {
-                    "achievement_id": "lvl10",
-                    "unlock_time": 1548421241
-                },
-                {
-                    "achievement_name": "Got level 20",
-                    "unlock_time": 1548422395
-                },
-                {
-                    "achievement_id": "lvl30",
-                    "achievement_name": "Got level 30",
-                    "unlock_time": 1548495633
-                }
-            ]
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "result": None
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "game_achievements_import_success",
+            "params": {
+                "game_id": "14",
+                "unlocked_achievements": [
+                    {
+                        "achievement_id": "lvl10",
+                        "unlock_time": 1548421241
+                    },
+                    {
+                        "achievement_name": "Got level 20",
+                        "unlock_time": 1548422395
+                    },
+                    {
+                        "achievement_id": "lvl30",
+                        "achievement_name": "Got level 30",
+                        "unlock_time": 1548495633
+                    }
+                ]
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "achievements_import_finished",
+            "params": None
         }
-    }
+    ]
 
-def test_failure(plugin, read, write):
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("exception,code,message", [
+    (BackendError, 4, "Backend error"),
+    (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",
-        "method": "import_unlocked_achievements",
+        "method": "start_achievements_import",
         "params": {
-            "game_id": "14"
+            "game_ids": ["14"]
         }
     }
 
-    read.side_effect = [json.dumps(request).encode() + b"\n", b""]
-    plugin.get_unlocked_achievements.coro.side_effect = UnknownError()
-    asyncio.run(plugin.run())
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+    plugin.get_unlocked_achievements.side_effect = exception
+    await plugin.run()
     plugin.get_unlocked_achievements.assert_called()
-    response = json.loads(write.call_args[0][0])
 
-    assert response == {
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "result": None
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "game_achievements_import_failure",
+            "params": {
+                "game_id": "14",
+                "error": {
+                    "code": code,
+                    "message": message
+                }
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "achievements_import_finished",
+            "params": None
+        }
+    ]
+
+@pytest.mark.asyncio
+async def test_prepare_get_unlocked_achievements_context_error(plugin, read, write):
+    plugin.prepare_achievements_context.side_effect = BackendError()
+    request = {
         "jsonrpc": "2.0",
         "id": "3",
-        "error": {
-            "code": 0,
-            "message": "Unknown error"
+        "method": "start_achievements_import",
+        "params": {
+            "game_ids": ["14"]
         }
     }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    await plugin.run()
 
-def test_unlock_achievement(plugin, write):
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "error": {
+                "code": 4,
+                "message": "Backend error"
+            }
+        }
+    ]
+
+
+@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",
+            "id": "3",
+            "method": "start_achievements_import",
+            "params": {
+                "game_ids": ["14"]
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "id": "4",
+            "method": "start_achievements_import",
+            "params": {
+                "game_ids": ["15"]
+            }
+        }
+    ]
+    read.side_effect = [
+        async_return_value(create_message(requests[0])),
+        async_return_value(create_message(requests[1])),
+        async_return_value(b"")
+    ]
+
+    await plugin.run()
+
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "result": None
+        },
+        {
+            "jsonrpc": "2.0",
+            "id": "4",
+            "error": {
+                "code": 600,
+                "message": "Import already in progress"
+            }
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_unlock_achievement(plugin, write):
     achievement = Achievement(achievement_id="lvl20", unlock_time=1548422395)
-
-    async def couritine():
-        plugin.unlock_achievement("14", achievement)
-
-    asyncio.run(couritine())
+    plugin.unlock_achievement("14", achievement)
     response = json.loads(write.call_args[0][0])
 
     assert response == {
@@ -102,92 +211,3 @@ def test_unlock_achievement(plugin, write):
             }
         }
     }
-
-@pytest.mark.asyncio
-async def test_game_achievements_import_success(plugin, write):
-    achievements = [
-        Achievement(achievement_id="lvl10", unlock_time=1548421241),
-        Achievement(achievement_name="Got level 20", unlock_time=1548422395)
-    ]
-    plugin.game_achievements_import_success("134", achievements)
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "method": "game_achievements_import_success",
-        "params": {
-            "game_id": "134",
-            "unlocked_achievements": [
-                {
-                    "achievement_id": "lvl10",
-                    "unlock_time": 1548421241
-                },
-                {
-                    "achievement_name": "Got level 20",
-                    "unlock_time": 1548422395
-                }
-            ]
-        }
-    }
-
-@pytest.mark.asyncio
-async def test_game_achievements_import_failure(plugin, write):
-    plugin.game_achievements_import_failure("134", ImportInProgress())
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "method": "game_achievements_import_failure",
-        "params": {
-            "game_id": "134",
-            "error": {
-                "code": 600,
-                "message": "Import already in progress"
-            }
-        }
-    }
-
-@pytest.mark.asyncio
-async def test_achievements_import_finished(plugin, write):
-    plugin.achievements_import_finished()
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "method": "achievements_import_finished",
-        "params": None
-    }
-
-@pytest.mark.asyncio
-async def test_start_achievements_import(plugin, write, mocker):
-    game_achievements_import_success = mocker.patch.object(plugin, "game_achievements_import_success")
-    game_achievements_import_failure = mocker.patch.object(plugin, "game_achievements_import_failure")
-    achievements_import_finished = mocker.patch.object(plugin, "achievements_import_finished")
-
-    game_ids = ["1", "5", "9"]
-    error = BackendError()
-    achievements = [
-        Achievement(achievement_id="lvl10", unlock_time=1548421241),
-        Achievement(achievement_name="Got level 20", unlock_time=1548422395)
-    ]
-    plugin.get_unlocked_achievements.coro.side_effect = [
-        achievements,
-        [],
-        error
-    ]
-    await plugin.start_achievements_import(game_ids)
-
-    with pytest.raises(ImportInProgress):
-        await plugin.start_achievements_import(["4", "8"])
-
-    # wait until all tasks are finished
-    for _ in range(4):
-        await asyncio.sleep(0)
-
-    plugin.get_unlocked_achievements.coro.assert_has_calls([call("1"), call("5"), call("9")])
-    game_achievements_import_success.assert_has_calls([
-        call("1", achievements),
-        call("5", [])
-    ])
-    game_achievements_import_failure.assert_called_once_with("9", error)
-    achievements_import_finished.assert_called_once_with()

tests/test_authenticate.py

@@ -1,6 +1,3 @@
-import asyncio
-import json
-
 import pytest
 
 from galaxy.api.types import Authentication
@@ -8,29 +5,36 @@ from galaxy.api.errors import (
     UnknownError, InvalidCredentials, NetworkError, LoggedInElsewhere, ProtocolError,
     BackendNotAvailable, BackendTimeout, BackendError, TemporaryBlocked, Banned, AccessDenied
 )
+from galaxy.unittest.mock import async_return_value
 
-def test_success(plugin, read, write):
+from tests import create_message, get_messages
+
+
+@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)
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "method": "store_credentials",
+            "params": credentials
+        }
+    ]
 
-    asyncio.run(couritine())
-    response = json.loads(write.call_args[0][0])
 
-    assert response == {
-        "jsonrpc": "2.0",
-        "method": "store_credentials",
-        "params": credentials
-    }
+@pytest.mark.asyncio
+async def test_lost_authentication(plugin, write):
+    plugin.lost_authentication()
 
-def test_lost_authentication(plugin, write):
-
-    async def couritine():
-        plugin.lost_authentication()
-
-    asyncio.run(couritine())
-    response = json.loads(write.call_args[0][0])
-
-    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()

tests/test_features.py

@@ -39,11 +39,11 @@ def test_multi_features():
         async def get_owned_games(self):
             pass
 
-        async def import_games_achievements(self, game_ids) -> None:
+        async def get_unlocked_achievements(self, game_id, context):
             pass
 
-        async def start_game_times_import(self, game_ids) -> None:
+        async def get_game_time(self, game_id, context):
             pass
 
     plugin = PluginImpl(Platform.Generic, "0.1", None, None, None)
-    assert set(plugin.features) == {Feature.ImportAchievements, Feature.ImportOwnedGames}
+    assert set(plugin.features) == {Feature.ImportAchievements, Feature.ImportOwnedGames, Feature.ImportGameTime}

tests/test_friends.py

@@ -1,90 +1,94 @@
-import asyncio
-import json
-
 from galaxy.api.types import FriendInfo
 from galaxy.api.errors import UnknownError
+from galaxy.unittest.mock import async_return_value
+
+import pytest
+
+from tests import create_message, get_messages
 
 
-def test_get_friends_success(plugin, read, write):
+@pytest.mark.asyncio
+async def test_get_friends_success(plugin, read, write):
     request = {
         "jsonrpc": "2.0",
         "id": "3",
         "method": "import_friends"
     }
 
-    read.side_effect = [json.dumps(request).encode() + b"\n", b""]
-    plugin.get_friends.coro.return_value = [
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    plugin.get_friends.return_value = async_return_value([
         FriendInfo("3", "Jan"),
         FriendInfo("5", "Ola")
-    ]
-    asyncio.run(plugin.run())
+    ])
+    await plugin.run()
     plugin.get_friends.assert_called_with()
-    response = json.loads(write.call_args[0][0])
 
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "3",
-        "result": {
-            "friend_info_list": [
-                {"user_id": "3", "user_name": "Jan"},
-                {"user_id": "5", "user_name": "Ola"}
-            ]
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "result": {
+                "friend_info_list": [
+                    {"user_id": "3", "user_name": "Jan"},
+                    {"user_id": "5", "user_name": "Ola"}
+                ]
+            }
         }
-    }
+    ]
 
 
-def test_get_friends_failure(plugin, read, write):
+@pytest.mark.asyncio
+async def test_get_friends_failure(plugin, read, write):
     request = {
         "jsonrpc": "2.0",
         "id": "3",
         "method": "import_friends"
     }
 
-    read.side_effect = [json.dumps(request).encode() + b"\n", b""]
-    plugin.get_friends.coro.side_effect = UnknownError()
-    asyncio.run(plugin.run())
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    plugin.get_friends.side_effect = UnknownError()
+    await plugin.run()
     plugin.get_friends.assert_called_with()
-    response = json.loads(write.call_args[0][0])
 
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "3",
-        "error": {
-            "code": 0,
-            "message": "Unknown error",
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "error": {
+                "code": 0,
+                "message": "Unknown error",
+            }
         }
-    }
+    ]
 
 
-def test_add_friend(plugin, write):
+@pytest.mark.asyncio
+async def test_add_friend(plugin, write):
     friend = FriendInfo("7", "Kuba")
 
-    async def couritine():
-        plugin.add_friend(friend)
+    plugin.add_friend(friend)
 
-    asyncio.run(couritine())
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "method": "friend_added",
-        "params": {
-            "friend_info": {"user_id": "7", "user_name": "Kuba"}
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "method": "friend_added",
+            "params": {
+                "friend_info": {"user_id": "7", "user_name": "Kuba"}
+            }
         }
-    }
+    ]
 
 
-def test_remove_friend(plugin, write):
-    async def couritine():
-        plugin.remove_friend("5")
+@pytest.mark.asyncio
+async def test_remove_friend(plugin, write):
+    plugin.remove_friend("5")
 
-    asyncio.run(couritine())
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "method": "friend_removed",
-        "params": {
-            "user_id": "5"
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "method": "friend_removed",
+            "params": {
+                "user_id": "5"
+            }
         }
-    }
+    ]

tests/test_game_times.py

@@ -1,179 +1,214 @@
-import asyncio
-import json
 from unittest.mock import call
 
 import pytest
 from galaxy.api.types import GameTime
-from galaxy.api.errors import UnknownError, ImportInProgress, BackendError
+from galaxy.api.errors import BackendError
+from galaxy.unittest.mock import async_return_value
 
-def test_success(plugin, read, write):
+from tests import create_message, get_messages
+
+
+@pytest.mark.asyncio
+async def test_get_game_time_success(plugin, read, write):
+    plugin.prepare_game_times_context.return_value = async_return_value("abc")
     request = {
         "jsonrpc": "2.0",
         "id": "3",
-        "method": "import_game_times"
+        "method": "start_game_times_import",
+        "params": {
+            "game_ids": ["3", "5", "7"]
+        }
     }
-
-    read.side_effect = [json.dumps(request).encode() + b"\n", b""]
-    plugin.get_game_times.coro.return_value = [
-        GameTime("3", 60, 1549550504),
-        GameTime("5", 10, None),
-        GameTime("7", None, 1549550502),
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+    plugin.get_game_time.side_effect = [
+        async_return_value(GameTime("3", 60, 1549550504)),
+        async_return_value(GameTime("5", 10, None)),
+        async_return_value(GameTime("7", None, 1549550502)),
     ]
-    asyncio.run(plugin.run())
-    plugin.get_game_times.assert_called_with()
-    response = json.loads(write.call_args[0][0])
+    await plugin.run()
+    plugin.get_game_time.assert_has_calls([
+        call("3", "abc"),
+        call("5", "abc"),
+        call("7", "abc"),
+    ])
 
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "3",
-        "result": {
-            "game_times": [
-                {
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "result": None
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "game_time_import_success",
+            "params": {
+                "game_time": {
                     "game_id": "3",
-                    "time_played": 60,
-                    "last_played_time": 1549550504
-                },
-                {
-                    "game_id": "5",
-                    "time_played": 10,
-                },
-                {
-                    "game_id": "7",
-                     "last_played_time": 1549550502
+                    "last_played_time": 1549550504,
+                    "time_played": 60
                 }
-            ]
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "game_time_import_success",
+            "params": {
+                "game_time": {
+                    "game_id": "5",
+                    "time_played": 10
+                }
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "game_time_import_success",
+            "params": {
+                "game_time": {
+                    "game_id": "7",
+                    "last_played_time": 1549550502
+                }
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "game_times_import_finished",
+            "params": None
         }
-    }
+    ]
 
-def test_failure(plugin, read, write):
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("exception,code,message", [
+    (BackendError, 4, "Backend error"),
+    (KeyError, 0, "Unknown error")
+])
+async def test_get_game_time_error(exception, code, message, plugin, read, write):
+    plugin.prepare_game_times_context.return_value = async_return_value(None)
     request = {
         "jsonrpc": "2.0",
         "id": "3",
-        "method": "import_game_times"
+        "method": "start_game_times_import",
+        "params": {
+            "game_ids": ["6"]
+        }
     }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+    plugin.get_game_time.side_effect = exception
+    await plugin.run()
+    plugin.get_game_time.assert_called()
 
-    read.side_effect = [json.dumps(request).encode() + b"\n", b""]
-    plugin.get_game_times.coro.side_effect = UnknownError()
-    asyncio.run(plugin.run())
-    plugin.get_game_times.assert_called_with()
-    response = json.loads(write.call_args[0][0])
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "result": None
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "game_time_import_failure",
+            "params": {
+                "game_id": "6",
+                "error": {
+                    "code": code,
+                    "message": message
+                }
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "game_times_import_finished",
+            "params": None
+        }
+    ]
 
-    assert response == {
+
+@pytest.mark.asyncio
+async def test_prepare_get_game_time_context_error(plugin, read, write):
+    plugin.prepare_game_times_context.side_effect = BackendError()
+    request = {
         "jsonrpc": "2.0",
         "id": "3",
-        "error": {
-            "code": 0,
-            "message": "Unknown error",
+        "method": "start_game_times_import",
+        "params": {
+            "game_ids": ["6"]
         }
     }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    await plugin.run()
 
-def test_update_game(plugin, write):
-    game_time = GameTime("3", 60, 1549550504)
-
-    async def couritine():
-        plugin.update_game_time(game_time)
-
-    asyncio.run(couritine())
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "method": "game_time_updated",
-        "params": {
-            "game_time": {
-                "game_id": "3",
-                "time_played": 60,
-                "last_played_time": 1549550504
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "error": {
+                "code": 4,
+                "message": "Backend error"
             }
         }
-    }
+    ]
+
 
 @pytest.mark.asyncio
-async def test_game_time_import_success(plugin, write):
-    plugin.game_time_import_success(GameTime("3", 60, 1549550504))
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "method": "game_time_import_success",
-        "params": {
-            "game_time": {
-                "game_id": "3",
-                "time_played": 60,
-                "last_played_time": 1549550504
+async def test_import_in_progress(plugin, read, write):
+    plugin.prepare_game_times_context.return_value = async_return_value(None)
+    requests = [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "method": "start_game_times_import",
+            "params": {
+                "game_ids": ["6"]
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "id": "4",
+            "method": "start_game_times_import",
+            "params": {
+                "game_ids": ["7"]
             }
         }
-    }
+    ]
+    read.side_effect = [
+        async_return_value(create_message(requests[0])),
+        async_return_value(create_message(requests[1])),
+        async_return_value(b"")
+    ]
 
-@pytest.mark.asyncio
-async def test_game_time_import_failure(plugin, write):
-    plugin.game_time_import_failure("134", ImportInProgress())
-    response = json.loads(write.call_args[0][0])
+    await plugin.run()
 
-    assert response == {
-        "jsonrpc": "2.0",
-        "method": "game_time_import_failure",
-        "params": {
-            "game_id": "134",
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "result": None
+        },
+        {
+            "jsonrpc": "2.0",
+            "id": "4",
             "error": {
                 "code": 600,
                 "message": "Import already in progress"
             }
         }
-    }
-
-@pytest.mark.asyncio
-async def test_game_times_import_finished(plugin, write):
-    plugin.game_times_import_finished()
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "method": "game_times_import_finished",
-        "params": None
-    }
-
-@pytest.mark.asyncio
-async def test_start_game_times_import(plugin, write, mocker):
-    game_time_import_success = mocker.patch.object(plugin, "game_time_import_success")
-    game_time_import_failure = mocker.patch.object(plugin, "game_time_import_failure")
-    game_times_import_finished = mocker.patch.object(plugin, "game_times_import_finished")
-
-    game_ids = ["1", "5"]
-    game_time = GameTime("1", 10, 1549550502)
-    plugin.get_game_times.coro.return_value = [
-        game_time
     ]
-    await plugin.start_game_times_import(game_ids)
 
-    with pytest.raises(ImportInProgress):
-        await plugin.start_game_times_import(["4", "8"])
-
-    # wait until all tasks are finished
-    for _ in range(4):
-        await asyncio.sleep(0)
-
-    plugin.get_game_times.coro.assert_called_once_with()
-    game_time_import_success.assert_called_once_with(game_time)
-    game_time_import_failure.assert_called_once_with("5", UnknownError())
-    game_times_import_finished.assert_called_once_with()
 
 @pytest.mark.asyncio
-async def test_start_game_times_import_failure(plugin, write, mocker):
-    game_time_import_failure = mocker.patch.object(plugin, "game_time_import_failure")
-    game_times_import_finished = mocker.patch.object(plugin, "game_times_import_finished")
+async def test_update_game(plugin, write):
+    game_time = GameTime("3", 60, 1549550504)
+    plugin.update_game_time(game_time)
 
-    game_ids = ["1", "5"]
-    error = BackendError()
-    plugin.get_game_times.coro.side_effect = error
-
-    await plugin.start_game_times_import(game_ids)
-
-    # wait until all tasks are finished
-    for _ in range(4):
-        await asyncio.sleep(0)
-
-    plugin.get_game_times.coro.assert_called_once_with()
-
-    assert game_time_import_failure.mock_calls == [call("1", error), call("5", error)]
-    game_times_import_finished.assert_called_once_with()
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "method": "game_time_updated",
+            "params": {
+                "game_time": {
+                    "game_id": "3",
+                    "time_played": 60,
+                    "last_played_time": 1549550504
+                }
+            }
+        }
+    ]

tests/test_install_game.py

@@ -1,7 +1,12 @@
-import asyncio
-import json
+import pytest
 
-def test_success(plugin, read):
+from galaxy.unittest.mock import async_return_value
+
+from tests import create_message
+
+
+@pytest.mark.asyncio
+async def test_success(plugin, read):
     request = {
         "jsonrpc": "2.0",
         "method": "install_game",
@@ -10,7 +15,6 @@ def test_success(plugin, read):
         }
     }
 
-    read.side_effect = [json.dumps(request).encode() + b"\n", b""]
-    plugin.get_owned_games.return_value = None
-    asyncio.run(plugin.run())
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    await plugin.run()
     plugin.install_game.assert_called_with(game_id="3")

tests/test_internal.py

@@ -1,10 +1,14 @@
-import asyncio
-import json
+import pytest
 
 from galaxy.api.plugin import Plugin
 from galaxy.api.consts import Platform
+from galaxy.unittest.mock import async_return_value
 
-def test_get_capabilites(reader, writer, read, write):
+from tests import create_message, get_messages
+
+
+@pytest.mark.asyncio
+async def test_get_capabilities(reader, writer, read, write):
     class PluginImpl(Plugin): #pylint: disable=abstract-method
         async def get_owned_games(self):
             pass
@@ -16,64 +20,75 @@ def test_get_capabilites(reader, writer, read, write):
     }
     token = "token"
     plugin = PluginImpl(Platform.Generic, "0.1", reader, writer, token)
-    read.side_effect = [json.dumps(request).encode() + b"\n", b""]
-    asyncio.run(plugin.run())
-    response = json.loads(write.call_args[0][0])
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "3",
-        "result": {
-            "platform_name": "generic",
-            "features": [
-                "ImportOwnedGames"
-            ],
-            "token": token
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    await plugin.run()
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "result": {
+                "platform_name": "generic",
+                "features": [
+                    "ImportOwnedGames"
+                ],
+                "token": token
+            }
         }
-    }
+    ]
 
-def test_shutdown(plugin, read, write):
+
+@pytest.mark.asyncio
+async def test_shutdown(plugin, read, write):
     request = {
         "jsonrpc": "2.0",
         "id": "5",
         "method": "shutdown"
     }
-    read.side_effect = [json.dumps(request).encode() + b"\n", b""]
-    asyncio.run(plugin.run())
+    read.side_effect = [async_return_value(create_message(request))]
+    await plugin.run()
     plugin.shutdown.assert_called_with()
-    response = json.loads(write.call_args[0][0])
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "5",
-        "result": None
-    }
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "5",
+            "result": None
+        }
+    ]
 
-def test_ping(plugin, read, write):
+
+@pytest.mark.asyncio
+async def test_ping(plugin, read, write):
     request = {
         "jsonrpc": "2.0",
         "id": "7",
         "method": "ping"
     }
-    read.side_effect = [json.dumps(request).encode() + b"\n", b""]
-    asyncio.run(plugin.run())
-    response = json.loads(write.call_args[0][0])
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "7",
-        "result": None
-    }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    await plugin.run()
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "7",
+            "result": None
+        }
+    ]
 
-def test_tick_before_handshake(plugin, read):
-    read.side_effect = [b""]
-    asyncio.run(plugin.run())
+
+@pytest.mark.asyncio
+async def test_tick_before_handshake(plugin, read):
+    read.side_effect = [async_return_value(b"")]
+    await plugin.run()
     plugin.tick.assert_not_called()
 
-def test_tick_after_handshake(plugin, read):
+
+@pytest.mark.asyncio
+async def test_tick_after_handshake(plugin, read):
     request = {
         "jsonrpc": "2.0",
         "id": "6",
         "method": "initialize_cache",
         "params": {"data": {}}
     }
-    read.side_effect = [json.dumps(request).encode() + b"\n", b""]
-    asyncio.run(plugin.run())
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    await plugin.run()
     plugin.tick.assert_called_with()

tests/test_launch_game.py

@@ -1,7 +1,12 @@
-import asyncio
-import json
+import pytest
 
-def test_success(plugin, read):
+from galaxy.unittest.mock import async_return_value
+
+from tests import create_message
+
+
+@pytest.mark.asyncio
+async def test_success(plugin, read):
     request = {
         "jsonrpc": "2.0",
         "method": "launch_game",
@@ -10,7 +15,6 @@ def test_success(plugin, read):
         }
     }
 
-    read.side_effect = [json.dumps(request).encode() + b"\n", b""]
-    plugin.get_owned_games.return_value = None
-    asyncio.run(plugin.run())
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    await plugin.run()
     plugin.launch_game.assert_called_with(game_id="3")

tests/test_local_games.py

@@ -1,51 +1,55 @@
-import asyncio
-import json
-
 import pytest
 
 from galaxy.api.types import LocalGame
 from galaxy.api.consts import LocalGameState
 from galaxy.api.errors import UnknownError, FailedParsingManifest
+from galaxy.unittest.mock import async_return_value
 
-def test_success(plugin, read, write):
+from tests import create_message, get_messages
+
+
+@pytest.mark.asyncio
+async def test_success(plugin, read, write):
     request = {
         "jsonrpc": "2.0",
         "id": "3",
         "method": "import_local_games"
     }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
 
-    read.side_effect = [json.dumps(request).encode() + b"\n", b""]
-
-    plugin.get_local_games.coro.return_value = [
+    plugin.get_local_games.return_value = async_return_value([
         LocalGame("1", LocalGameState.Running),
         LocalGame("2", LocalGameState.Installed),
         LocalGame("3", LocalGameState.Installed | LocalGameState.Running)
-    ]
-    asyncio.run(plugin.run())
+    ])
+    await plugin.run()
     plugin.get_local_games.assert_called_with()
 
-    response = json.loads(write.call_args[0][0])
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "3",
-        "result": {
-            "local_games" : [
-                {
-                    "game_id": "1",
-                    "local_game_state": LocalGameState.Running.value
-                },
-                {
-                    "game_id": "2",
-                    "local_game_state": LocalGameState.Installed.value
-                },
-                {
-                    "game_id": "3",
-                    "local_game_state": (LocalGameState.Installed | LocalGameState.Running).value
-                }
-            ]
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "result": {
+                "local_games" : [
+                    {
+                        "game_id": "1",
+                        "local_game_state": LocalGameState.Running.value
+                    },
+                    {
+                        "game_id": "2",
+                        "local_game_state": LocalGameState.Installed.value
+                    },
+                    {
+                        "game_id": "3",
+                        "local_game_state": (LocalGameState.Installed | LocalGameState.Running).value
+                    }
+                ]
+            }
         }
-    }
+    ]
 
+
+@pytest.mark.asyncio
 @pytest.mark.parametrize(
     "error,code,message",
     [
@@ -53,44 +57,42 @@ def test_success(plugin, read, write):
         pytest.param(FailedParsingManifest, 200, "Failed parsing manifest", id="failed_parsing")
     ],
 )
-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": "import_local_games"
     }
-
-    read.side_effect = [json.dumps(request).encode() + b"\n", b""]
-    plugin.get_local_games.coro.side_effect = error()
-    asyncio.run(plugin.run())
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    plugin.get_local_games.side_effect = error()
+    await plugin.run()
     plugin.get_local_games.assert_called_with()
-    response = json.loads(write.call_args[0][0])
 
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "3",
-        "error": {
-            "code": code,
-            "message": message
-        }
-    }
-
-def test_local_game_state_update(plugin, write):
-    game = LocalGame("1", LocalGameState.Running)
-
-    async def couritine():
-        plugin.update_local_game_status(game)
-
-    asyncio.run(couritine())
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "method": "local_game_status_changed",
-        "params": {
-            "local_game": {
-                "game_id": "1",
-                "local_game_state": LocalGameState.Running.value
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "error": {
+                "code": code,
+                "message": message
             }
         }
-    }
+    ]
+
+@pytest.mark.asyncio
+async def test_local_game_state_update(plugin, write):
+    game = LocalGame("1", LocalGameState.Running)
+    plugin.update_local_game_status(game)
+
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "method": "local_game_status_changed",
+            "params": {
+                "local_game": {
+                    "game_id": "1",
+                    "local_game_state": LocalGameState.Running.value
+                }
+            }
+        }
+    ]

tests/test_owned_games.py

@@ -1,19 +1,23 @@
-import asyncio
-import json
+import pytest
 
 from galaxy.api.types import Game, Dlc, LicenseInfo
 from galaxy.api.consts import LicenseType
 from galaxy.api.errors import UnknownError
+from galaxy.unittest.mock import async_return_value
 
-def test_success(plugin, read, write):
+from tests import create_message, get_messages
+
+
+@pytest.mark.asyncio
+async def test_success(plugin, read, write):
     request = {
         "jsonrpc": "2.0",
         "id": "3",
         "method": "import_owned_games"
     }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
 
-    read.side_effect = [json.dumps(request).encode() + b"\n", b""]
-    plugin.get_owned_games.coro.return_value = [
+    plugin.get_owned_games.return_value = async_return_value([
         Game("3", "Doom", None, LicenseInfo(LicenseType.SinglePurchase, None)),
         Game(
             "5",
@@ -23,129 +27,126 @@ def test_success(plugin, read, write):
                 Dlc("8", "Temerian Armor Set", LicenseInfo(LicenseType.FreeToPlay, None)),
             ],
             LicenseInfo(LicenseType.SinglePurchase, None))
-    ]
-    asyncio.run(plugin.run())
+    ])
+    await plugin.run()
     plugin.get_owned_games.assert_called_with()
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "3",
-        "result": {
-            "owned_games": [
-                {
-                    "game_id": "3",
-                    "game_title": "Doom",
-                    "license_info": {
-                        "license_type": "SinglePurchase"
-                    }
-                },
-                {
-                    "game_id": "5",
-                    "game_title": "Witcher 3",
-                    "dlcs": [
-                        {
-                            "dlc_id": "7",
-                            "dlc_title": "Hearts of Stone",
-                            "license_info": {
-                                "license_type": "SinglePurchase"
-                            }
-                        },
-                        {
-                            "dlc_id": "8",
-                            "dlc_title": "Temerian Armor Set",
-                            "license_info": {
-                                "license_type": "FreeToPlay"
-                            }
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "result": {
+                "owned_games": [
+                    {
+                        "game_id": "3",
+                        "game_title": "Doom",
+                        "license_info": {
+                            "license_type": "SinglePurchase"
+                        }
+                    },
+                    {
+                        "game_id": "5",
+                        "game_title": "Witcher 3",
+                        "dlcs": [
+                            {
+                                "dlc_id": "7",
+                                "dlc_title": "Hearts of Stone",
+                                "license_info": {
+                                    "license_type": "SinglePurchase"
+                                }
+                            },
+                            {
+                                "dlc_id": "8",
+                                "dlc_title": "Temerian Armor Set",
+                                "license_info": {
+                                    "license_type": "FreeToPlay"
+                                }
+                            }
+                        ],
+                        "license_info": {
+                            "license_type": "SinglePurchase"
                         }
-                    ],
-                    "license_info": {
-                        "license_type": "SinglePurchase"
                     }
-                }
-            ]
+                ]
+            }
         }
-    }
+    ]
 
-def test_failure(plugin, read, write):
+
+@pytest.mark.asyncio
+async def test_failure(plugin, read, write):
     request = {
         "jsonrpc": "2.0",
         "id": "3",
         "method": "import_owned_games"
     }
 
-    read.side_effect = [json.dumps(request).encode() + b"\n", b""]
-    plugin.get_owned_games.coro.side_effect = UnknownError()
-    asyncio.run(plugin.run())
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    plugin.get_owned_games.side_effect = UnknownError()
+    await plugin.run()
     plugin.get_owned_games.assert_called_with()
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "3",
-        "error": {
-            "code": 0,
-            "message": "Unknown error"
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "error": {
+                "code": 0,
+                "message": "Unknown error"
+            }
         }
-    }
+    ]
 
-def test_add_game(plugin, write):
+
+@pytest.mark.asyncio
+async def test_add_game(plugin, write):
     game = Game("3", "Doom", None, LicenseInfo(LicenseType.SinglePurchase, None))
-
-    async def couritine():
-        plugin.add_game(game)
-
-    asyncio.run(couritine())
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "method": "owned_game_added",
-        "params": {
-            "owned_game": {
-                "game_id": "3",
-                "game_title": "Doom",
-                "license_info": {
-                    "license_type": "SinglePurchase"
+    plugin.add_game(game)
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "method": "owned_game_added",
+            "params": {
+                "owned_game": {
+                    "game_id": "3",
+                    "game_title": "Doom",
+                    "license_info": {
+                        "license_type": "SinglePurchase"
+                    }
                 }
             }
         }
-    }
+    ]
 
-def test_remove_game(plugin, write):
-    async def couritine():
-        plugin.remove_game("5")
 
-    asyncio.run(couritine())
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "method": "owned_game_removed",
-        "params": {
-            "game_id": "5"
+@pytest.mark.asyncio
+async def test_remove_game(plugin, write):
+    plugin.remove_game("5")
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "method": "owned_game_removed",
+            "params": {
+                "game_id": "5"
+            }
         }
-    }
+    ]
 
-def test_update_game(plugin, write):
+
+@pytest.mark.asyncio
+async def test_update_game(plugin, write):
     game = Game("3", "Doom", None, LicenseInfo(LicenseType.SinglePurchase, None))
-
-    async def couritine():
-        plugin.update_game(game)
-
-    asyncio.run(couritine())
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "method": "owned_game_updated",
-        "params": {
-            "owned_game": {
-                "game_id": "3",
-                "game_title": "Doom",
-                "license_info": {
-                    "license_type": "SinglePurchase"
+    plugin.update_game(game)
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "method": "owned_game_updated",
+            "params": {
+                "owned_game": {
+                    "game_id": "3",
+                    "game_title": "Doom",
+                    "license_info": {
+                        "license_type": "SinglePurchase"
+                    }
                 }
             }
         }
-    }
+    ]

tests/test_persistent_cache.py

@@ -1,23 +1,28 @@
-import asyncio
-import json
-
 import pytest
 
+from galaxy.unittest.mock import async_return_value
+
+from tests import create_message, get_messages
+
 
 def assert_rpc_response(write, response_id, result=None):
-    assert json.loads(write.call_args[0][0]) == {
-        "jsonrpc": "2.0",
-        "id": str(response_id),
-        "result": result
-    }
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": str(response_id),
+            "result": result
+        }
+    ]
 
 
 def assert_rpc_request(write, method, params=None):
-    assert json.loads(write.call_args[0][0]) == {
-        "jsonrpc": "2.0",
-        "method": method,
-        "params": {"data": params}
-    }
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "method": method,
+            "params": {"data": params}
+        }
+    ]
 
 
 @pytest.fixture
@@ -28,7 +33,8 @@ def cache_data():
     }
 
 
-def test_initialize_cache(plugin, read, write, cache_data):
+@pytest.mark.asyncio
+async def test_initialize_cache(plugin, read, write, cache_data):
     request_id = 3
     request = {
         "jsonrpc": "2.0",
@@ -36,36 +42,32 @@ def test_initialize_cache(plugin, read, write, cache_data):
         "method": "initialize_cache",
         "params": {"data": cache_data}
     }
-    read.side_effect = [json.dumps(request).encode() + b"\n"]
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
 
     assert {} == plugin.persistent_cache
-    asyncio.run(plugin.run())
+    await plugin.run()
     plugin.handshake_complete.assert_called_once_with()
     assert cache_data == plugin.persistent_cache
     assert_rpc_response(write, response_id=request_id)
 
 
-def test_set_cache(plugin, write, cache_data):
-    async def runner():
-        assert {} == plugin.persistent_cache
+@pytest.mark.asyncio
+async def test_set_cache(plugin, write, cache_data):
+    assert {} == plugin.persistent_cache
 
-        plugin.persistent_cache.update(cache_data)
-        plugin.push_cache()
+    plugin.persistent_cache.update(cache_data)
+    plugin.push_cache()
 
-        assert_rpc_request(write, "push_cache", cache_data)
-        assert cache_data == plugin.persistent_cache
-
-    asyncio.run(runner())
+    assert_rpc_request(write, "push_cache", cache_data)
+    assert cache_data == plugin.persistent_cache
 
 
-def test_clear_cache(plugin, write, cache_data):
-    async def runner():
-        plugin._persistent_cache = cache_data
+@pytest.mark.asyncio
+async def test_clear_cache(plugin, write, cache_data):
+    plugin._persistent_cache = cache_data
 
-        plugin.persistent_cache.clear()
-        plugin.push_cache()
+    plugin.persistent_cache.clear()
+    plugin.push_cache()
 
-        assert_rpc_request(write, "push_cache", {})
-        assert {} == plugin.persistent_cache
-
-    asyncio.run(runner())
+    assert_rpc_request(write, "push_cache", {})
+    assert {} == plugin.persistent_cache

tests/test_shutdown_platform_client.py

@@ -1,7 +1,9 @@
-import json
-
 import pytest
 
+from galaxy.unittest.mock import async_return_value
+
+from tests import create_message
+
 @pytest.mark.asyncio
 async def test_success(plugin, read):
     request = {
@@ -9,7 +11,7 @@ async def test_success(plugin, read):
         "method": "shutdown_platform_client"
     }
 
-    read.side_effect = [json.dumps(request).encode() + b"\n", b""]
-    plugin.shutdown_platform_client.return_value = None
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    plugin.shutdown_platform_client.return_value = async_return_value(None)
     await plugin.run()
     plugin.shutdown_platform_client.assert_called_with()

tests/test_stream_line_reader.py

@@ -1,52 +1,46 @@
-from unittest.mock import MagicMock
-
 import pytest
 
 from galaxy.reader import StreamLineReader
-from galaxy.unittest.mock import AsyncMock
+from galaxy.unittest.mock import async_return_value
+
 
 @pytest.fixture()
-def stream_reader():
-    reader = MagicMock()
-    reader.read = AsyncMock()
-    return reader
+def stream_line_reader(reader):
+    return StreamLineReader(reader)
 
-@pytest.fixture()
-def read(stream_reader):
-    return stream_reader.read
-
-@pytest.fixture()
-def reader(stream_reader):
-    return StreamLineReader(stream_reader)
 
 @pytest.mark.asyncio
-async def test_message(reader, read):
-    read.return_value = b"a\n"
-    assert await reader.readline() == b"a"
+async def test_message(stream_line_reader, read):
+    read.return_value = async_return_value(b"a\n")
+    assert await stream_line_reader.readline() == b"a"
     read.assert_called_once()
 
-@pytest.mark.asyncio
-async def test_separate_messages(reader, read):
-    read.side_effect = [b"a\n", b"b\n"]
-    assert await reader.readline() == b"a"
-    assert await reader.readline() == b"b"
-    assert read.call_count == 2
 
 @pytest.mark.asyncio
-async def test_connected_messages(reader, read):
-    read.return_value = b"a\nb\n"
-    assert await reader.readline() == b"a"
-    assert await reader.readline() == b"b"
+async def test_separate_messages(stream_line_reader, read):
+    read.side_effect = [async_return_value(b"a\n"), async_return_value(b"b\n")]
+    assert await stream_line_reader.readline() == b"a"
+    assert await stream_line_reader.readline() == b"b"
+    assert read.call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_connected_messages(stream_line_reader, read):
+    read.return_value = async_return_value(b"a\nb\n")
+    assert await stream_line_reader.readline() == b"a"
+    assert await stream_line_reader.readline() == b"b"
     read.assert_called_once()
 
-@pytest.mark.asyncio
-async def test_cut_message(reader, read):
-    read.side_effect = [b"a", b"b\n"]
-    assert await reader.readline() == b"ab"
-    assert read.call_count == 2
 
 @pytest.mark.asyncio
-async def test_half_message(reader, read):
-    read.side_effect = [b"a", b""]
-    assert await reader.readline() == b""
+async def test_cut_message(stream_line_reader, read):
+    read.side_effect = [async_return_value(b"a"), async_return_value(b"b\n")]
+    assert await stream_line_reader.readline() == b"ab"
+    assert read.call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_half_message(stream_line_reader, read):
+    read.side_effect = [async_return_value(b"a"), async_return_value(b"")]
+    assert await stream_line_reader.readline() == b""
     assert read.call_count == 2

tests/test_uninstall_game.py

@@ -1,7 +1,11 @@
-import asyncio
-import json
+import pytest
 
-def test_success(plugin, read):
+from galaxy.unittest.mock import async_return_value
+
+from tests import create_message
+
+@pytest.mark.asyncio
+async def test_success(plugin, read):
     request = {
         "jsonrpc": "2.0",
         "method": "uninstall_game",
@@ -9,8 +13,7 @@ def test_success(plugin, read):
             "game_id": "3"
         }
     }
-
-    read.side_effect = [json.dumps(request).encode() + b"\n", b""]
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
     plugin.get_owned_games.return_value = None
-    asyncio.run(plugin.run())
+    await plugin.run()
     plugin.uninstall_game.assert_called_with(game_id="3")