Wait for drain before writing

This reverts commit 21ab8bf33d3c2714b8c7af2da8036fe8baae20ea.

.gitignore

@@ -1,2 +1,10 @@
 # pytest
 __pycache__/
+.vscode/
+.venv/
+src/galaxy.plugin.api.egg-info/
+docs/build/
+Pipfile
+.idea
+docs/source/_build
+.mypy_cache

.gitlab-ci.yml

@@ -0,0 +1,32 @@
+image: registry-gitlab.gog.com/galaxy-client/gitlab-ci-tools:latest
+
+stages:
+  - test
+  - deploy
+
+test_package:
+  stage: test
+  script:
+    - pip install -r requirements.txt
+    - pytest
+  except:
+    - tags
+
+deploy_package:
+  stage: deploy
+  variables:
+    TWINE_USERNAME: $PYPI_USERNAME
+    TWINE_PASSWORD: $PYPI_PASSWORD
+  script:
+    - pip install twine wheel
+    - rm -rf dist
+    - export VERSION=$(python setup.py --version)
+    - python setup.py sdist --formats=gztar bdist_wheel
+    - twine upload dist/*
+    - curl -X POST --silent --show-error --fail
+        "https://gitlab.gog.com/api/v4/projects/${CI_PROJECT_ID}/repository/tags?tag_name=${VERSION}&ref=${CI_COMMIT_REF_NAME}&private_token=${PACKAGE_DEPLOYER_API_TOKEN}"
+  when: manual
+  only:
+    - master
+  except:
+    - tags

.travis.yml

@@ -0,0 +1,8 @@
+dist: xenial   # required for Python >= 3.7
+language: python
+python:
+  - "3.7"
+install:
+  - pip install -r requirements.txt
+script:
+  - pytest

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2019 GOG sp. z o.o.
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

PLATFORM_IDs.md

@@ -0,0 +1,82 @@
+### PLATFORM ID LIST
+
+Platform ID list for GOG Galaxy 2.0 Integrations
+
+| ID | Name |
+| --- | --- |
+| steam	| Steam |
+| psn	| PlayStation Network |
+| xboxone	| Xbox Live |
+| generic	| Manually added games |
+| origin	| Origin |
+| uplay	| Uplay |
+| battlenet	| Battle.net |
+| epic	| Epic Games Store |
+| bethesda	| Bethesda.net |
+| paradox	| Paradox Plaza |
+| humble	| Humble Bundle |
+| kartridge	| Kartridge |
+| itch	| Itch.io |
+| nswitch	| Nintendo Switch |
+| nwiiu	| Nintendo Wii U |
+| nwii	| Nintendo Wii |
+| ncube	| Nintendo GameCube |
+| riot	| Riot |
+| wargaming	| Wargaming |
+| ngameboy	| Nintendo Game Boy |
+| atari	| Atari |
+| amiga	| Amiga |
+| snes	| SNES |
+| beamdog	| Beamdog |
+| d2d	| Direct2Drive |
+| discord	| Discord |
+| dotemu	| DotEmu |
+| gamehouse	| GameHouse |
+| gmg	| Green Man Gaming |
+| weplay	| WePlay |
+| zx	| Zx Spectrum PC |
+| vision	| ColecoVision |
+| nes	| NES |
+| sms	| Sega Master System |
+| c64	| Commodore 64 |
+| pce 	| PC Engine |
+| segag	| Sega Genesis |
+| neo	| NeoGeo |
+| sega32	| Sega 32X |
+| segacd	| Sega CD |
+| 3do	| 3DO Interactive  |
+| saturn	| SegaSaturn |
+| psx	| Sony PlayStation |
+| ps2	| Sony PlayStation 2 |
+| n64	| Nintendo64 |
+| jaguar	| Atari Jaguar |
+| dc	| Sega Dreamcast |
+| xboxog	| Original Xbox games |
+| amazon	| Amazon |
+| gg	| GamersGate |
+| egg	| Newegg |
+| bb	| BestBuy |
+| gameuk	| Game UK |
+| fanatical	| Fanatical store |
+| playasia 	| Play-Asia |
+| stadia	| Google Stadia |
+| arc	| ARC |
+| eso	| ESO |
+| glyph	| Trion World |
+| aionl	| Aion: Legions of War |
+| aion	| Aion |
+| blade	| Blade & Soul |
+| gw	| Guild Wars |
+| gw2	| Guild Wars 2 |
+| lin2	| Lineage 2 |
+| ffxi	| Final Fantasy XI |
+| ffxiv	| Final Fantasy XIV  |
+| totalwar	| Total War |
+| winstore	| Windows Store |
+| elites	| Elite Dangerous |
+| star	| Star Citizen |
+| psp	| PlayStation Portable |
+| psvita	| PlayStation Vita |
+| nds	| Nintendo DS |
+| 3ds	| Nintendo 3DS |
+| pathofexile | Path of Exile |

README.md

@@ -1,29 +1,35 @@
 # GOG Galaxy Integrations Python API
 
-This Python library allows to easily build community integrations for various gaming platforms with GOG Galaxy 2.0.
+This Python library allows developers to easily build community integrations for various gaming platforms with GOG Galaxy 2.0.
 
 - refer to our <a href='https://galaxy-integrations-python-api.readthedocs.io'>documentation</a>
 
 ## Features
 
-Each integration in GOG Galaxy 2.0 comes as a separate Python script, and is launched as a separate process, that which needs to communicate with main instance of GOG Galaxy 2.0.
+Each integration in GOG Galaxy 2.0 comes as a separate Python script and is launched as a separate process that needs to communicate with the main instance of GOG Galaxy 2.0.
 
 The provided features are:
 
-- multistep authorisation using a browser built into GOG Galaxy 2.0
+- multistep authorization using a browser built into GOG Galaxy 2.0
 - support for GOG Galaxy 2.0 features:
-    - importing owned and detecting installed games
-    - installing and launching games
-    - importing achievements and game time
-    - importing friends lists and statuses
-    - importing friends recomendations list
-    - receiving and sending chat messages
+  - importing owned and detecting installed games
+  - installing and launching games
+  - importing achievements and game time
+  - importing friends lists and statuses
+  - importing friends recommendations list
+  - receiving and sending chat messages
 - cache storage
 
+## Platform Id's
+
+Each integration can implement only one platform. Each integration must declare which platform it's integrating.
+
+[List of possible Platform IDs](PLATFORM_IDs.md)
+
 ## Basic usage
 
-Eeach integration should inherit from the :class:`~galaxy.api.plugin.Plugin` class. Supported methods like :meth:`~galaxy.api.plugin.Plugin.get_owned_games` should be overwritten - they are called from the GOG Galaxy client in the appropriate times.
-Each of those method can raise exceptions inherited from the :exc:`~galaxy.api.jsonrpc.ApplicationError`.
+Each integration should inherit from the :class:`~galaxy.api.plugin.Plugin` class. Supported methods like :meth:`~galaxy.api.plugin.Plugin.get_owned_games` should be overwritten - they are called from the GOG Galaxy client at the appropriate times.
+Each of those methods can raise exceptions inherited from the :exc:`~galaxy.api.jsonrpc.ApplicationError`.
 Communication between an integration and the client is also possible with the use of notifications, for example: :meth:`~galaxy.api.plugin.Plugin.update_local_game_status`.
 
 ```python
@@ -55,10 +61,13 @@ if __name__ == "__main__":
 
 ## Deployment
 
-The client has a built-in Python 3.7 interpreter, so the integrations are delivered as python modules.
-In order to be found by GOG Galaxy 2.0 an integration folder should be placed in [lookup directory](#deploy-location). Beside all the python files, the integration folder has to contain [manifest.json](#deploy-manifest) and all third-party dependencies. See an [examplary structure](#deploy-structure-example).
+The client has a built-in Python 3.7 interpreter, so integrations are delivered as Python modules.
+In order to be found by GOG Galaxy 2.0 an integration folder should be placed in [lookup directory](#deploy-location). Beside all the Python files, the integration folder must contain [manifest.json](#deploy-manifest) and all third-party dependencies. See an [exemplary structure](#deploy-structure-example).
+
+### Lookup directory
+
+<a name="deploy-location"></a>
 
-### <a name="deploy-location"></a> Lookup directory:
 - Windows:
 
     `%localappdata%\GOG.com\Galaxy\plugins\installed`
@@ -67,8 +76,10 @@ In order to be found by GOG Galaxy 2.0 an integration folder should be placed in
 
     `~/Library/Application Support/GOG.com/Galaxy/plugins/installed`
 
-### <a name="deploy-manifest"></a> Manifest
-Obligatory JSON file to be placed in a integration folder.
+### Manifest
+
+<a name="deploy-manifest"></a>
+Obligatory JSON file to be placed in an integration folder.
 
 ```json
 {
@@ -83,6 +94,7 @@ Obligatory JSON file to be placed in a integration folder.
     "script": "plugin.py"
 }
 ```
+
 | property      | description |
 |---------------|---|
 | `guid`        |   |
@@ -91,13 +103,15 @@ Obligatory JSON file to be placed in a integration folder.
 | `script`      | path of the entry point module, relative to the integration folder |
 
 ### Dependencies
-All third-party packages (packages not included in Python 3.7 standard library) should be deployed along with plugin files. Use the folowing command structure:
+
+All third-party packages (packages not included in the Python 3.7 standard library) should be deployed along with plugin files. Use the following command structure:
 
 ```pip install DEP --target DIR --implementation cp --python-version 37```
 
-For example plugin that uses *requests* has structure as follows:
+For example, a plugin that uses *requests* could have the following structure:
 
 <a name="deploy-structure-example"></a>
+
 ```bash
 installed
 └── my_integration

docs/requirements.txt

@@ -1,4 +1,5 @@
 Sphinx==2.0.1
 sphinx-rtd-theme==0.4.3
 sphinx-autodoc-typehints==1.6.0
+sphinxcontrib-asyncio==0.2.0
 m2r==0.2.1

docs/source/conf.py

@@ -32,12 +32,13 @@ release = _version
 # ones.
 extensions = [
     'sphinx.ext.autodoc',
+    'sphinxcontrib.asyncio',
     'sphinx_autodoc_typehints',
     'm2r'  # mdinclude directive for makrdown files
 ]
 autodoc_member_order = 'bysource'
 autodoc_inherit_docstrings = False
-autodoc_mock_imports = ["galaxy.http"]
+autodoc_mock_imports = ["aiohttp"]
 
 set_type_checking_flag = True
 
@@ -47,7 +48,7 @@ templates_path = ['_templates']
 # List of patterns, relative to source directory, that match files and
 # directories to ignore when looking for source files.
 # This pattern also affects html_static_path and html_extra_path.
-exclude_patterns = []
+exclude_patterns = [] # type: ignore
 
 
 # -- Options for HTML output -------------------------------------------------

docs/source/galaxy.http.rst

@@ -0,0 +1,8 @@
+galaxy.http
+=================
+
+.. automodule:: galaxy.http
+    :members:
+    :special-members: __init__
+    :undoc-members:
+    :show-inheritance:

docs/source/index.rst

@@ -6,7 +6,9 @@ GOG Galaxy Integrations Python API
    :includehidden:
 
    Overview <overview>
-   API <galaxy.api>
+   galaxy.api
+   galaxy.http
+   Platform ID's <platforms>
 
 Index
 -------------------

docs/source/overview.rst

@@ -5,3 +5,11 @@
 
 .. mdinclude:: ../../README.md
    :start-line: 6
+   :end-line: 26
+   
+.. excluding Platforms Id's link
+
+:ref:`platforms-link`
+
+.. mdinclude:: ../../README.md
+   :start-line: 28

docs/source/platforms.rst

@@ -0,0 +1,2 @@
+.. _platforms-link:
+.. mdinclude:: ../../PLATFORM_IDs.md

mypy.ini

@@ -0,0 +1,2 @@
+[mypy]
+ignore_missing_imports = True

pytest.ini

@@ -1,2 +1,2 @@
 [pytest]
-addopts = --flakes
+addopts = --flakes --mypy

requirements.txt

@@ -2,7 +2,9 @@
 pytest==4.2.0
 pytest-asyncio==0.10.0
 pytest-mock==1.10.3
+pytest-mypy==0.4.1
 pytest-flakes==4.0.0
 # because of pip bug https://github.com/pypa/pip/issues/4780
 aiohttp==3.5.4
-certifi==2019.3.9
+certifi==2019.3.9
+psutil==5.6.3; sys_platform == 'darwin'

setup.py

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

src/galaxy/__init__.py

@@ -1 +1 @@
-__path__ = __import__('pkgutil').extend_path(__path__, __name__)
+__path__: str = __import__('pkgutil').extend_path(__path__, __name__) # type: ignore

src/galaxy/api/consts.py

@@ -13,6 +13,84 @@ class Platform(Enum):
     Uplay = "uplay"
     Battlenet = "battlenet"
     Epic = "epic"
+    Bethesda = "bethesda"
+    ParadoxPlaza = "paradox"
+    HumbleBundle = "humble"
+    Kartridge = "kartridge"
+    ItchIo = "itch"
+    NintendoSwitch = "nswitch"
+    NintendoWiiU = "nwiiu"
+    NintendoWii = "nwii"
+    NintendoGameCube = "ncube"
+    RiotGames = "riot"
+    Wargaming = "wargaming"
+    NintendoGameBoy = "ngameboy"
+    Atari = "atari"
+    Amiga = "amiga"
+    SuperNintendoEntertainmentSystem = "snes"
+    Beamdog = "beamdog"
+    Direct2Drive = "d2d"
+    Discord = "discord"
+    DotEmu = "dotemu"
+    GameHouse = "gamehouse"
+    GreenManGaming = "gmg"
+    WePlay = "weplay"
+    ZxSpectrum = "zx"
+    ColecoVision = "vision"
+    NintendoEntertainmentSystem = "nes"
+    SegaMasterSystem = "sms"
+    Commodore64 = "c64"
+    PcEngine = "pce"
+    SegaGenesis = "segag"
+    NeoGeo = "neo"
+    Sega32X = "sega32"
+    SegaCd = "segacd"
+    _3Do = "3do"
+    SegaSaturn = "saturn"
+    PlayStation = "psx"
+    PlayStation2 = "ps2"
+    Nintendo64 = "n64"
+    AtariJaguar = "jaguar"
+    SegaDreamcast = "dc"
+    Xbox = "xboxog"
+    Amazon = "amazon"
+    GamersGate = "gg"
+    Newegg = "egg"
+    BestBuy = "bb"
+    GameUk = "gameuk"
+    Fanatical = "fanatical"
+    PlayAsia = "playasia"
+    Stadia = "stadia"
+    Arc = "arc"
+    ElderScrollsOnline = "eso"
+    Glyph = "glyph"
+    AionLegionsOfWar = "aionl"
+    Aion = "aion"
+    BladeAndSoul = "blade"
+    GuildWars = "gw"
+    GuildWars2 = "gw2"
+    Lineage2 = "lin2"
+    FinalFantasy11 = "ffxi"
+    FinalFantasy14 = "ffxiv"
+    TotalWar = "totalwar"
+    WindowsStore = "winstore"
+    EliteDangerous = "elites"
+    StarCitizen = "star"
+    PlayStationPortable = "psp"
+    PlayStationVita = "psvita"
+    NintendoDs = "nds"
+    Nintendo3Ds = "3ds"
+    PathOfExile = "pathofexile"
+    Twitch = "twitch"
+    Minecraft = "minecraft"
+    GameSessions = "gamesessions"
+    Nuuvem = "nuuvem"
+    FXStore = "fxstore"
+    IndieGala = "indiegala"
+    Playfire = "playfire"
+    Oculus = "oculus"
+    Test = "test"
+    Rockstar = "rockstar"
 
 
 class Feature(Enum):
@@ -31,6 +109,11 @@ class Feature(Enum):
     ImportUsers = "ImportUsers"
     VerifyGame = "VerifyGame"
     ImportFriends = "ImportFriends"
+    ShutdownPlatformClient = "ShutdownPlatformClient"
+    LaunchPlatformClient = "LaunchPlatformClient"
+    ImportGameLibrarySettings = "ImportGameLibrarySettings"
+    ImportOSCompatibility = "ImportOSCompatibility"
+    ImportUserPresence = "ImportUserPresence"
 
 
 class LicenseType(Enum):
@@ -51,9 +134,18 @@ class LocalGameState(Flag):
     Running = 2
 
 
+class OSCompatibility(Flag):
+    """Possible game OS compatibility.
+    Use "bitwise or" to express multiple OSs compatibility, e.g. ``os=OSCompatibility.Windows|OSCompatibility.MacOS``
+    """
+    Windows = 0b001
+    MacOS   = 0b010
+    Linux   = 0b100
+
+
 class PresenceState(Enum):
-    """"Possible states that a user can be in."""
-    Unknown = "Unknown"
+    """"Possible states of a user."""
+    Unknown = "unknown"
     Online = "online"
     Offline = "offline"
     Away = "away"

src/galaxy/api/errors.py

@@ -1,6 +1,6 @@
 from galaxy.api.jsonrpc import ApplicationError, UnknownError
 
-UnknownError = UnknownError
+assert UnknownError
 
 class AuthenticationRequired(ApplicationError):
     def __init__(self, data=None):

src/galaxy/api/jsonrpc.py

@@ -5,6 +5,9 @@ import logging
 import inspect
 import json
 
+from galaxy.reader import StreamLineReader
+from galaxy.task_manager import TaskManager
+
 class JsonRpcError(Exception):
     def __init__(self, code, message, data=None):
         self.code = code
@@ -15,6 +18,17 @@ class JsonRpcError(Exception):
     def __eq__(self, other):
         return self.code == other.code and self.message == other.message and self.data == other.data
 
+    def json(self):
+        obj = {
+            "code": self.code,
+            "message": self.message
+        }
+
+        if self.data is not None:
+            obj["error"]["data"] = self.data
+
+        return obj
+
 class ParseError(JsonRpcError):
     def __init__(self):
         super().__init__(-32700, "Parse error")
@@ -50,7 +64,8 @@ class UnknownError(ApplicationError):
         super().__init__(0, "Unknown error", data)
 
 Request = namedtuple("Request", ["method", "params", "id"], defaults=[{}, None])
-Method = namedtuple("Method", ["callback", "signature", "internal", "sensitive_params"])
+Method = namedtuple("Method", ["callback", "signature", "immediate", "sensitive_params"])
+
 
 def anonymise_sensitive_params(params, sensitive_params):
     anomized_data = "****"
@@ -67,14 +82,15 @@ def anonymise_sensitive_params(params, sensitive_params):
 class Server():
     def __init__(self, reader, writer, encoder=json.JSONEncoder()):
         self._active = True
-        self._reader = reader
+        self._reader = StreamLineReader(reader)
         self._writer = writer
         self._encoder = encoder
         self._methods = {}
         self._notifications = {}
-        self._eof_listeners = []
+        self._task_manager = TaskManager("jsonrpc server")
+        self._write_lock = asyncio.Lock()
 
-    def register_method(self, name, callback, internal, sensitive_params=False):
+    def register_method(self, name, callback, immediate, sensitive_params=False):
         """
         Register method
 
@@ -84,9 +100,9 @@ class Server():
         :param sensitive_params: list of parameters that are anonymized before logging; \
             if False - no params are considered sensitive, if True - all params are considered sensitive
         """
-        self._methods[name] = Method(callback, inspect.signature(callback), internal, sensitive_params)
+        self._methods[name] = Method(callback, inspect.signature(callback), immediate, sensitive_params)
 
-    def register_notification(self, name, callback, internal, sensitive_params=False):
+    def register_notification(self, name, callback, immediate, sensitive_params=False):
         """
         Register notification
 
@@ -96,10 +112,7 @@ class Server():
         :param sensitive_params: list of parameters that are anonymized before logging; \
             if False - no params are considered sensitive, if True - all params are considered sensitive
         """
-        self._notifications[name] = Method(callback, inspect.signature(callback), internal, sensitive_params)
-
-    def register_eof(self, callback):
-        self._eof_listeners.append(callback)
+        self._notifications[name] = Method(callback, inspect.signature(callback), immediate, sensitive_params)
 
     async def run(self):
         while self._active:
@@ -114,15 +127,19 @@ class Server():
             data = data.strip()
             logging.debug("Received %d bytes of data", len(data))
             self._handle_input(data)
+            await asyncio.sleep(0) # To not starve task queue
 
-    def stop(self):
-        self._active = False
+    def close(self):
+        if self._active:
+            logging.info("Closing JSON-RPC server - not more messages will be read")
+            self._active = False
+
+    async def wait_closed(self):
+        await self._task_manager.wait()
 
     def _eof(self):
         logging.info("Received EOF")
-        self.stop()
-        for listener in self._eof_listeners:
-            listener()
+        self.close()
 
     def _handle_input(self, data):
         try:
@@ -142,7 +159,7 @@ class Server():
             logging.error("Received unknown notification: %s", request.method)
             return
 
-        callback, signature, internal, sensitive_params = method
+        callback, signature, immediate, sensitive_params = method
         self._log_request(request, sensitive_params)
 
         try:
@@ -150,12 +167,11 @@ class Server():
         except TypeError:
             self._send_error(request.id, InvalidParams())
 
-        if internal:
-            # internal requests are handled immediately
+        if immediate:
             callback(*bound_args.args, **bound_args.kwargs)
         else:
             try:
-                asyncio.create_task(callback(*bound_args.args, **bound_args.kwargs))
+                self._task_manager.create_task(callback(*bound_args.args, **bound_args.kwargs), request.method)
             except Exception:
                 logging.exception("Unexpected exception raised in notification handler")
 
@@ -166,7 +182,7 @@ class Server():
             self._send_error(request.id, MethodNotFound())
             return
 
-        callback, signature, internal, sensitive_params = method
+        callback, signature, immediate, sensitive_params = method
         self._log_request(request, sensitive_params)
 
         try:
@@ -174,8 +190,7 @@ class Server():
         except TypeError:
             self._send_error(request.id, InvalidParams())
 
-        if internal:
-            # internal requests are handled immediately
+        if immediate:
             response = callback(*bound_args.args, **bound_args.kwargs)
             self._send_response(request.id, response)
         else:
@@ -187,11 +202,13 @@ class Server():
                     self._send_error(request.id, MethodNotFound())
                 except JsonRpcError as error:
                     self._send_error(request.id, error)
+                except asyncio.CancelledError:
+                    self._send_error(request.id, Aborted())
                 except Exception as e:  #pylint: disable=broad-except
                     logging.exception("Unexpected exception raised in plugin handler")
                     self._send_error(request.id, UnknownError(str(e)))
 
-            asyncio.create_task(handle())
+            self._task_manager.create_task(handle(), request.method)
 
     @staticmethod
     def _parse_request(data):
@@ -207,12 +224,16 @@ class Server():
             raise InvalidRequest()
 
     def _send(self, data):
+        async def send_task(data_):
+            async with self._write_lock:
+                self._writer.write(data_)
+                await self._writer.drain()
+
         try:
             line = self._encoder.encode(data)
             logging.debug("Sending data: %s", line)
             data = (line + "\n").encode("utf-8")
-            self._writer.write(data)
-            asyncio.create_task(self._writer.drain())
+            self._task_manager.create_task(send_task(data), "send")
         except TypeError as error:
             logging.error(str(error))
 
@@ -228,15 +249,9 @@ class Server():
         response = {
             "jsonrpc": "2.0",
             "id": request_id,
-            "error": {
-                "code": error.code,
-                "message": error.message
-            }
+            "error": error.json()
         }
 
-        if error.data is not None:
-            response["error"]["data"] = error.data
-
         self._send(response)
 
     @staticmethod
@@ -252,6 +267,8 @@ class NotificationClient():
         self._writer = writer
         self._encoder = encoder
         self._methods = {}
+        self._task_manager = TaskManager("notification client")
+        self._write_lock = asyncio.Lock()
 
     def notify(self, method, params, sensitive_params=False):
         """
@@ -270,13 +287,21 @@ class NotificationClient():
         self._log(method, params, sensitive_params)
         self._send(notification)
 
+    async def close(self):
+        self._task_manager.cancel()
+        await self._task_manager.wait()
+
     def _send(self, data):
+        async def send_task(data_):
+            async with self._write_lock:
+                self._writer.write(data_)
+                await self._writer.drain()
+
         try:
             line = self._encoder.encode(data)
             data = (line + "\n").encode("utf-8")
             logging.debug("Sending %d byte of data", len(data))
-            self._writer.write(data)
-            asyncio.create_task(self._writer.drain())
+            self._task_manager.create_task(send_task(data), "send")
         except TypeError as error:
             logging.error("Failed to parse outgoing message: %s", str(error))
 

src/galaxy/api/types.py

@@ -1,10 +1,11 @@
 from dataclasses import dataclass
-from typing import List, Dict, Optional
+from typing import Dict, List, Optional
 
 from galaxy.api.consts import LicenseType, LocalGameState, PresenceState
 
+
 @dataclass
-class Authentication():
+class Authentication:
     """Return this from :meth:`.authenticate` or :meth:`.pass_login_credentials`
     to inform the client that authentication has successfully finished.
 
@@ -14,8 +15,9 @@ class Authentication():
     user_id: str
     user_name: str
 
+
 @dataclass
-class Cookie():
+class Cookie:
     """Cookie
 
     :param name: name of the cookie
@@ -28,8 +30,9 @@ class Cookie():
     domain: Optional[str] = None
     path: Optional[str] = None
 
+
 @dataclass
-class NextStep():
+class NextStep:
     """Return this from :meth:`.authenticate` or :meth:`.pass_login_credentials` to open client built-in browser with given url.
     For example:
 
@@ -61,15 +64,15 @@ class NextStep():
     :param auth_params: configuration options: {"window_title": :class:`str`, "window_width": :class:`str`, "window_height": :class:`int`, "start_uri": :class:`int`, "end_uri_regex": :class:`str`}
     :param cookies: browser initial set of cookies
     :param js: a map of the url regex patterns into the list of *js* scripts that should be executed on every document at given step of internal browser authentication.
-
     """
     next_step: str
     auth_params: Dict[str, str]
     cookies: Optional[List[Cookie]] = None
     js: Optional[Dict[str, List[str]]] = None
 
+
 @dataclass
-class LicenseInfo():
+class LicenseInfo:
     """Information about the license of related product.
 
     :param license_type: type of license
@@ -78,8 +81,9 @@ class LicenseInfo():
     license_type: LicenseType
     owner: Optional[str] = None
 
+
 @dataclass
-class Dlc():
+class Dlc:
     """Downloadable content object.
 
     :param dlc_id: id of the dlc
@@ -90,8 +94,9 @@ class Dlc():
     dlc_title: str
     license_info: LicenseInfo
 
+
 @dataclass
-class Game():
+class Game:
     """Game object.
 
     :param game_id: unique identifier of the game, this will be passed as parameter for methods such as launch_game
@@ -104,8 +109,9 @@ class Game():
     dlcs: Optional[List[Dlc]]
     license_info: LicenseInfo
 
+
 @dataclass
-class Achievement():
+class Achievement:
     """Achievement, has to be initialized with either id or name.
 
     :param unlock_time: unlock time of the achievement
@@ -120,8 +126,9 @@ class Achievement():
         assert self.achievement_id or self.achievement_name, \
             "One of achievement_id or achievement_name is required"
 
+
 @dataclass
-class LocalGame():
+class LocalGame:
     """Game locally present on the authenticated user's computer.
 
     :param game_id: id of the game
@@ -130,36 +137,9 @@ class LocalGame():
     game_id: str
     local_game_state: LocalGameState
 
-@dataclass
-class Presence():
-    """Information about a presence of a user.
-
-    :param presence_state: the state in which the user's presence is
-    :param game_id: id of the game which the user is currently playing
-    :param presence_status: optional attached string with the detailed description of the user's presence
-    """
-    presence_state: PresenceState
-    game_id: Optional[str] = None
-    presence_status: Optional[str] = None
 
 @dataclass
-class UserInfo():
-    """Detailed information about a user.
-
-    :param user_id: of the user
-    :param is_friend: whether the user is a friend of the currently authenticated user
-    :param user_name: of the user
-    :param avatar_url: to the avatar of the user
-    :param presence: about the users presence
-    """
-    user_id: str
-    is_friend: bool
-    user_name: str
-    avatar_url: str
-    presence: Presence
-
-@dataclass
-class FriendInfo():
+class FriendInfo:
     """Information about a friend of the currently authenticated user.
 
     :param user_id: id of the user
@@ -168,34 +148,9 @@ class FriendInfo():
     user_id: str
     user_name: str
 
-@dataclass
-class Room():
-    """WIP, Chatroom.
-
-    :param room_id: id of the room
-    :param unread_message_count: number of unread messages in the room
-    :param last_message_id: id of the last message in the room
-    """
-    room_id: str
-    unread_message_count: int
-    last_message_id: str
 
 @dataclass
-class Message():
-    """WIP, A chatroom message.
-
-    :param message_id: id of the message
-    :param sender_id: id of the sender of the message
-    :param sent_time: time at which the message was sent
-    :param message_text: text attached to the message
-    """
-    message_id: str
-    sender_id: str
-    sent_time: int
-    message_text: str
-
-@dataclass
-class GameTime():
+class GameTime:
     """Game time of a game, defines the total time spent in the game
     and the last time the game was played.
 
@@ -204,5 +159,33 @@ class GameTime():
     :param last_time_played: last time the game was played (**unix timestamp**)
     """
     game_id: str
-    time_played: int
-    last_played_time: int
+    time_played: Optional[int]
+    last_played_time: Optional[int]
+
+
+@dataclass
+class GameLibrarySettings:
+    """Library settings of a game, defines assigned tags and visibility flag.
+
+    :param game_id: id of the related game
+    :param tags: collection of tags assigned to the game
+    :param hidden: indicates if the game should be hidden in GOG Galaxy application
+    """
+    game_id: str
+    tags: Optional[List[str]]
+    hidden: Optional[bool]
+
+
+@dataclass
+class UserPresence:
+    """Presence information of a user.
+
+    :param presence_state: the state of the user
+    :param game_id: id of the game a user is currently in
+    :param game_title: name of the game a user is currently in
+    :param presence_status: detailed user's presence description
+    """
+    presence_state: PresenceState
+    game_id: Optional[str] = None
+    game_title: Optional[str] = None
+    presence_status: Optional[str] = None

src/galaxy/http.py

@@ -1,5 +1,37 @@
+"""
+This module standarize http traffic and the error handling for further communication with the GOG Galaxy 2.0.
+
+It is recommended to use provided convenient methods for HTTP requests, especially when dealing with authorized sessions.
+Examplary simple web service could looks like:
+
+    .. code-block:: python
+
+        import logging
+        from galaxy.http import create_client_session, handle_exception
+
+        class BackendClient:
+            AUTH_URL = 'my-integration.com/auth'
+            HEADERS = {
+                "My-Custom-Header": "true",
+            }
+            def __init__(self):
+                self._session = create_client_session(headers=self.HEADERS)
+
+            async def authenticate(self):
+                await self._session.request('POST', self.AUTH_URL)
+
+            async def close(self):
+                # to be called on plugin shutdown
+                await self._session.close()
+
+            async def _authorized_request(self, method, url, *args, **kwargs):
+                with handle_exceptions():
+                    return await self._session.request(method, url, *args, **kwargs)
+"""
+
 import asyncio
 import ssl
+from contextlib import contextmanager
 from http import HTTPStatus
 
 import aiohttp
@@ -12,44 +44,103 @@ from galaxy.api.errors import (
 )
 
 
+#: Default limit of the simultaneous connections for ssl connector.
+DEFAULT_LIMIT = 20
+#: Default timeout in seconds used for client session.
+DEFAULT_TIMEOUT = 60
+
+
 class HttpClient:
-    def __init__(self, limit=20, timeout=aiohttp.ClientTimeout(total=60), cookie_jar=None):
-        ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
-        ssl_context.load_verify_locations(certifi.where())
-        connector = aiohttp.TCPConnector(limit=limit, ssl=ssl_context)
-        self._session = aiohttp.ClientSession(connector=connector, timeout=timeout, cookie_jar=cookie_jar)
+    """
+    .. deprecated:: 0.41
+      Use http module functions instead
+    """
+    def __init__(self, limit=DEFAULT_LIMIT, timeout=aiohttp.ClientTimeout(total=DEFAULT_TIMEOUT), cookie_jar=None):
+        connector = create_tcp_connector(limit=limit)
+        self._session = create_client_session(connector=connector, timeout=timeout, cookie_jar=cookie_jar)
 
     async def close(self):
+        """Closes connection. Should be called in :meth:`~galaxy.api.plugin.Plugin.shutdown`"""
         await self._session.close()
 
     async def request(self, method, url, *args, **kwargs):
-        try:
-            response = await self._session.request(method, url, *args, **kwargs)
-        except asyncio.TimeoutError:
-            raise BackendTimeout()
-        except aiohttp.ServerDisconnectedError:
-            raise BackendNotAvailable()
-        except aiohttp.ClientConnectionError:
-            raise NetworkError()
-        except aiohttp.ContentTypeError:
-            raise UnknownBackendResponse()
-        except aiohttp.ClientError:
-            logging.exception(
-                "Caught exception while running {} request for {}".format(method, url))
-            raise UnknownError()
-        if response.status == HTTPStatus.UNAUTHORIZED:
-            raise AuthenticationRequired()
-        if response.status == HTTPStatus.FORBIDDEN:
-            raise AccessDenied()
-        if response.status == HTTPStatus.SERVICE_UNAVAILABLE:
-            raise BackendNotAvailable()
-        if response.status == HTTPStatus.TOO_MANY_REQUESTS:
-            raise TooManyRequests()
-        if response.status >= 500:
-            raise BackendError()
-        if response.status >= 400:
-            logging.warning(
-                "Got status {} while running {} request for {}".format(response.status, method, url))
-            raise UnknownError()
+        with handle_exception():
+            return await self._session.request(method, url, *args, **kwargs)
 
-        return response
+
+def create_tcp_connector(*args, **kwargs) -> aiohttp.TCPConnector:
+    """
+    Creates TCP connector with resonable defaults.
+    For details about available parameters refer to
+    `aiohttp.TCPConnector <https://docs.aiohttp.org/en/stable/client_reference.html#tcpconnector>`_
+    """
+    ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+    ssl_context.load_verify_locations(certifi.where())
+    kwargs.setdefault("ssl", ssl_context)
+    kwargs.setdefault("limit", DEFAULT_LIMIT)
+    # due to https://github.com/python/mypy/issues/4001
+    return aiohttp.TCPConnector(*args, **kwargs)  # type: ignore
+
+
+def create_client_session(*args, **kwargs) -> aiohttp.ClientSession:
+    """
+    Creates client session with resonable defaults.
+    For details about available parameters refer to
+    `aiohttp.ClientSession <https://docs.aiohttp.org/en/stable/client_reference.html>`_
+
+    Examplary customization:
+
+    .. code-block:: python
+
+        from galaxy.http import create_client_session, create_tcp_connector
+
+        session = create_client_session(
+            headers={
+                "Keep-Alive": "true"
+            },
+            connector=create_tcp_connector(limit=40),
+            timeout=100)
+    """
+    kwargs.setdefault("connector", create_tcp_connector())
+    kwargs.setdefault("timeout", aiohttp.ClientTimeout(total=DEFAULT_TIMEOUT))
+    kwargs.setdefault("raise_for_status", True)
+    # due to https://github.com/python/mypy/issues/4001
+    return aiohttp.ClientSession(*args, **kwargs)  # type: ignore
+
+
+@contextmanager
+def handle_exception():
+    """
+    Context manager translating network related exceptions
+    to custom :mod:`~galaxy.api.errors`.
+    """
+    try:
+        yield
+    except asyncio.TimeoutError:
+        raise BackendTimeout()
+    except aiohttp.ServerDisconnectedError:
+        raise BackendNotAvailable()
+    except aiohttp.ClientConnectionError:
+        raise NetworkError()
+    except aiohttp.ContentTypeError:
+        raise UnknownBackendResponse()
+    except aiohttp.ClientResponseError as error:
+        if error.status == HTTPStatus.UNAUTHORIZED:
+            raise AuthenticationRequired()
+        if error.status == HTTPStatus.FORBIDDEN:
+            raise AccessDenied()
+        if error.status == HTTPStatus.SERVICE_UNAVAILABLE:
+            raise BackendNotAvailable()
+        if error.status == HTTPStatus.TOO_MANY_REQUESTS:
+            raise TooManyRequests()
+        if error.status >= 500:
+            raise BackendError()
+        if error.status >= 400:
+            logging.warning(
+                "Got status %d while performing %s request for %s",
+                error.status, error.request_info.method, str(error.request_info.url)
+            )
+            raise UnknownError()
+    except aiohttp.ClientError:
+        logging.exception("Caught exception while performing request")
+        raise UnknownError()

src/galaxy/proc_tools.py

@@ -0,0 +1,88 @@
+import sys
+from dataclasses import dataclass
+from typing import Iterable, NewType, Optional, List, cast
+
+
+
+ProcessId = NewType("ProcessId", int)
+
+
+@dataclass
+class ProcessInfo:
+    pid: ProcessId
+    binary_path: Optional[str]
+
+
+if sys.platform == "win32":
+    from ctypes import byref, sizeof, windll, create_unicode_buffer, FormatError, WinError
+    from ctypes.wintypes import DWORD
+
+
+    def pids() -> Iterable[ProcessId]:
+        _PROC_ID_T = DWORD
+        list_size = 4096
+
+        def try_get_pids(list_size: int) -> List[ProcessId]:
+            result_size = DWORD()
+            proc_id_list = (_PROC_ID_T * list_size)()
+
+            if not windll.psapi.EnumProcesses(byref(proc_id_list), sizeof(proc_id_list), byref(result_size)):
+                raise WinError(descr="Failed to get process ID list: %s" % FormatError())  # type: ignore
+
+            return cast(List[ProcessId], proc_id_list[:int(result_size.value / sizeof(_PROC_ID_T()))])
+
+        while True:
+            proc_ids = try_get_pids(list_size)
+            if len(proc_ids) < list_size:
+                return proc_ids
+
+            list_size *= 2
+
+
+    def get_process_info(pid: ProcessId) -> Optional[ProcessInfo]:
+        _PROC_QUERY_LIMITED_INFORMATION = 0x1000
+
+        process_info = ProcessInfo(pid=pid, binary_path=None)
+
+        h_process = windll.kernel32.OpenProcess(_PROC_QUERY_LIMITED_INFORMATION, False, pid)
+        if not h_process:
+            return process_info
+
+        try:
+            def get_exe_path() -> Optional[str]:
+                _MAX_PATH = 260
+                _WIN32_PATH_FORMAT = 0x0000
+
+                exe_path_buffer = create_unicode_buffer(_MAX_PATH)
+                exe_path_len = DWORD(len(exe_path_buffer))
+
+                return cast(str, exe_path_buffer[:exe_path_len.value]) if windll.kernel32.QueryFullProcessImageNameW(
+                    h_process, _WIN32_PATH_FORMAT, exe_path_buffer, byref(exe_path_len)
+                ) else None
+
+            process_info.binary_path = get_exe_path()
+        finally:
+            windll.kernel32.CloseHandle(h_process)
+            return process_info
+else:
+    import psutil
+
+
+    def pids() -> Iterable[ProcessId]:
+        for pid in psutil.pids():
+            yield pid
+
+
+    def get_process_info(pid: ProcessId) -> Optional[ProcessInfo]:
+        process_info = ProcessInfo(pid=pid, binary_path=None)
+        try:
+            process_info.binary_path = psutil.Process(pid=pid).as_dict(attrs=["exe"])["exe"]
+        except psutil.NoSuchProcess:
+            pass
+        finally:
+            return process_info
+
+
+def process_iter() -> Iterable[Optional[ProcessInfo]]:
+    for pid in pids():
+        yield get_process_info(pid)

src/galaxy/reader.py

@@ -0,0 +1,28 @@
+from asyncio import StreamReader
+
+
+class StreamLineReader:
+    """Handles StreamReader readline without buffer limit"""
+    def __init__(self, reader: StreamReader):
+        self._reader = reader
+        self._buffer = bytes()
+        self._processed_buffer_it = 0
+
+    async def readline(self):
+        while True:
+            # check if there is no unprocessed data in the buffer
+            if not self._buffer or self._processed_buffer_it != 0:
+                chunk = await self._reader.read(1024)
+                if not chunk:
+                    return bytes() # EOF
+                self._buffer += chunk
+
+            it = self._buffer.find(b"\n", self._processed_buffer_it)
+            if it < 0:
+                self._processed_buffer_it = len(self._buffer)
+                continue
+
+            line = self._buffer[:it]
+            self._buffer = self._buffer[it+1:]
+            self._processed_buffer_it = 0
+            return line

src/galaxy/registry_monitor.py

@@ -0,0 +1,98 @@
+import sys
+if sys.platform == "win32":
+    import logging
+    import ctypes
+    from ctypes.wintypes import LONG, HKEY, LPCWSTR, DWORD, BOOL, HANDLE, LPVOID
+
+    LPSECURITY_ATTRIBUTES = LPVOID
+
+    RegOpenKeyEx = ctypes.windll.advapi32.RegOpenKeyExW
+    RegOpenKeyEx.restype = LONG
+    RegOpenKeyEx.argtypes = [HKEY, LPCWSTR, DWORD, DWORD, ctypes.POINTER(HKEY)]
+
+    RegCloseKey = ctypes.windll.advapi32.RegCloseKey
+    RegCloseKey.restype = LONG
+    RegCloseKey.argtypes = [HKEY]
+
+    RegNotifyChangeKeyValue = ctypes.windll.advapi32.RegNotifyChangeKeyValue
+    RegNotifyChangeKeyValue.restype = LONG
+    RegNotifyChangeKeyValue.argtypes = [HKEY, BOOL, DWORD, HANDLE, BOOL]
+
+    CloseHandle = ctypes.windll.kernel32.CloseHandle
+    CloseHandle.restype = BOOL
+    CloseHandle.argtypes = [HANDLE]
+
+    CreateEvent = ctypes.windll.kernel32.CreateEventW
+    CreateEvent.restype = BOOL
+    CreateEvent.argtypes = [LPSECURITY_ATTRIBUTES, BOOL, BOOL, LPCWSTR]
+
+    WaitForSingleObject = ctypes.windll.kernel32.WaitForSingleObject
+    WaitForSingleObject.restype = DWORD
+    WaitForSingleObject.argtypes = [HANDLE, DWORD]
+
+    ERROR_SUCCESS = 0x00000000
+
+    KEY_READ = 0x00020019
+    KEY_QUERY_VALUE = 0x00000001
+
+    REG_NOTIFY_CHANGE_NAME = 0x00000001
+    REG_NOTIFY_CHANGE_LAST_SET = 0x00000004
+
+    WAIT_OBJECT_0 = 0x00000000
+    WAIT_TIMEOUT = 0x00000102
+
+class RegistryMonitor:
+
+    def __init__(self, root, subkey):
+        self._root = root
+        self._subkey = subkey
+        self._event = CreateEvent(None, False, False, None)
+
+        self._key = None
+        self._open_key()
+        if self._key:
+            self._set_key_update_notification()
+
+    def close(self):
+        CloseHandle(self._event)
+        if self._key:
+            RegCloseKey(self._key)
+            self._key = None
+
+    def is_updated(self):
+        wait_result = WaitForSingleObject(self._event, 0)
+
+        # previously watched
+        if wait_result == WAIT_OBJECT_0:
+            self._set_key_update_notification()
+            return True
+
+        # no changes or no key before
+        if wait_result != WAIT_TIMEOUT:
+            # unexpected error
+            logging.warning("Unexpected WaitForSingleObject result %s", wait_result)
+            return False
+
+        if self._key is None:
+            self._open_key()
+
+        if self._key is None:
+            return False
+
+        self._set_key_update_notification()
+        return True
+
+    def _set_key_update_notification(self):
+        filter_ = REG_NOTIFY_CHANGE_NAME | REG_NOTIFY_CHANGE_LAST_SET
+        status = RegNotifyChangeKeyValue(self._key, True, filter_, self._event, True)
+        if status != ERROR_SUCCESS:
+            # key was deleted
+            RegCloseKey(self._key)
+            self._key = None
+
+    def _open_key(self):
+        access = KEY_QUERY_VALUE | KEY_READ
+        self._key = HKEY()
+        rc = RegOpenKeyEx(self._root, self._subkey, 0, access, ctypes.byref(self._key))
+        if rc != ERROR_SUCCESS:
+            self._key = None

src/galaxy/task_manager.py

@@ -0,0 +1,49 @@
+import asyncio
+import logging
+from collections import OrderedDict
+from itertools import count
+
+class TaskManager:
+    def __init__(self, name):
+        self._name = name
+        self._tasks = OrderedDict()
+        self._task_counter = count()
+
+    def create_task(self, coro, description, handle_exceptions=True):
+        """Wrapper around asyncio.create_task - takes care of canceling tasks on shutdown"""
+
+        async def task_wrapper(task_id):
+            try:
+                result = await coro
+                logging.debug("Task manager %s: finished task %d (%s)", self._name, task_id, description)
+                return result
+            except asyncio.CancelledError:
+                if handle_exceptions:
+                    logging.debug("Task manager %s: canceled task %d (%s)", self._name, task_id, description)
+                else:
+                    raise
+            except Exception:
+                if handle_exceptions:
+                    logging.exception("Task manager %s: exception raised in task %d (%s)", self._name, task_id, description)
+                else:
+                    raise
+            finally:
+                del self._tasks[task_id]
+
+        task_id = next(self._task_counter)
+        logging.debug("Task manager %s: creating task %d (%s)", self._name, task_id, description)
+        task = asyncio.create_task(task_wrapper(task_id))
+        self._tasks[task_id] = task
+        return task
+
+    def cancel(self):
+        for task in self._tasks.values():
+            task.cancel()
+
+    async def wait(self):
+        # Tasks can spawn other tasks
+        while True:
+            tasks = self._tasks.values()
+            if not tasks:
+                return
+            await asyncio.gather(*tasks, return_exceptions=True)

src/galaxy/tools.py

@@ -3,6 +3,7 @@ import os
 import zipfile
 from glob import glob
 
+
 def zip_folder(folder):
     files = glob(os.path.join(folder, "**"), recursive=True)
     files = [file.replace(folder + os.sep, "") for file in files]
@@ -14,6 +15,7 @@ def zip_folder(folder):
             zipf.write(os.path.join(folder, file), arcname=file)
     return zip_buffer
 
+
 def zip_folder_to_file(folder, filename):
     zip_content = zip_folder(folder).getbuffer()
     with open(filename, "wb") as archive:

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,16 @@
+import json
+
+
+def create_message(request):
+    return json.dumps(request).encode() + b"\n"
+
+
+def get_messages(write_mock):
+    messages = []
+    for call_args in write_mock.call_args_list:
+        data = call_args[0][0]
+        for line in data.splitlines():
+            message = json.loads(line)
+            messages.append(message)
+    return messages
+

tests/conftest.py

@@ -1,67 +1,79 @@
-from contextlib import ExitStack
 import logging
-from unittest.mock import patch, MagicMock
+from contextlib import ExitStack
+from unittest.mock import MagicMock, patch
 
 import pytest
 
-from galaxy.api.plugin import Plugin
 from galaxy.api.consts import Platform
-from galaxy.unittest.mock import AsyncMock, coroutine_mock
+from galaxy.api.plugin import Plugin
+from galaxy.unittest.mock import async_return_value
+
 
 @pytest.fixture()
 def reader():
     stream = MagicMock(name="stream_reader")
-    stream.readline = 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.side_effect = lambda: async_return_value(None)
     yield stream
 
+
 @pytest.fixture()
-def readline(reader):
-    yield reader.readline
+def read(reader):
+    yield reader.read
+
 
 @pytest.fixture()
 def write(writer):
     yield writer.write
 
+
 @pytest.fixture()
-def plugin(reader, writer):
+async 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",
+        "achievements_import_complete",
         "get_local_games",
         "launch_game",
+        "launch_platform_client",
         "install_game",
         "uninstall_game",
         "get_friends",
-        "get_users",
-        "send_message",
-        "mark_as_read",
-        "get_rooms",
-        "get_room_history_from_message",
-        "get_room_history_from_timestamp",
-        "get_game_times"
-    )
-
-    methods = (
+        "get_game_time",
+        "prepare_game_times_context",
+        "game_times_import_complete",
+        "shutdown_platform_client",
         "shutdown",
-        "tick"
+        "tick",
+        "get_game_library_settings",
+        "prepare_game_library_settings_context",
+        "game_library_settings_import_complete",
+        "get_os_compatibility",
+        "prepare_os_compatibility_context",
+        "os_compatibility_import_complete",
+        "get_user_presence",
+        "prepare_user_presence_context",
+        "user_presence_import_complete",
     )
 
     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")
+
+        async with Plugin(Platform.Generic, "0.1", reader, writer, "token") as plugin:
+            plugin.shutdown.return_value = async_return_value(None)
+            yield plugin
+
 
 @pytest.fixture(autouse=True)
 def my_caplog(caplog):

tests/test_achievements.py

@@ -1,94 +1,207 @@
-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, skip_loop
+
+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, readline, 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"]
         }
     }
-    readline.side_effect = [json.dumps(request), ""]
-    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)
+    plugin.achievements_import_complete.asert_called_with()
 
-    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, readline, 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"]
         }
     }
 
-    readline.side_effect = [json.dumps(request), ""]
-    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])
+    plugin.achievements_import_complete.asert_called_with()
 
-    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"", 10)]
 
-def test_unlock_achievement(plugin, write):
+    await plugin.run()
+
+    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)
+    plugin.get_unlocked_achievements.return_value = async_return_value([])
+    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"", 10)
+    ]
+
+    await plugin.run()
+
+    messages = get_messages(write)
+    assert {
+        "jsonrpc": "2.0",
+        "id": "3",
+        "result": None
+    } in messages
+    assert {
+        "jsonrpc": "2.0",
+        "id": "4",
+        "error": {
+            "code": 600,
+            "message": "Import already in progress"
+        }
+    } in messages
+
+
+@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)
+    await skip_loop()
     response = json.loads(write.call_args[0][0])
 
     assert response == {
@@ -102,92 +215,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, skip_loop
 
-def test_success(plugin, readline, 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"
     }
-
-    readline.side_effect = [json.dumps(request), ""]
-    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"", 10)]
+    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, readline, write):
     pytest.param(Banned, 105, "Banned", id="banned"),
     pytest.param(AccessDenied, 106, "Access denied", id="access_denied"),
 ])
-def test_failure(plugin, readline, write, error, code, message):
+async def test_failure(plugin, read, write, error, code, message):
     request = {
         "jsonrpc": "2.0",
         "id": "3",
         "method": "init_authentication"
     }
 
-    readline.side_effect = [json.dumps(request), ""]
-    plugin.authenticate.coro.side_effect = error()
-    asyncio.run(plugin.run())
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+    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, readline, write):
+
+@pytest.mark.asyncio
+async def test_stored_credentials(plugin, read, write):
     request = {
         "jsonrpc": "2.0",
         "id": "3",
@@ -77,39 +84,39 @@ def test_stored_credentials(plugin, readline, write):
             }
         }
     }
-    readline.side_effect = [json.dumps(request), ""]
-    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"", 10)]
+    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)
+    await skip_loop()
 
-    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()
+    await skip_loop()
 
-def test_lost_authentication(plugin, readline, 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_chat.py

@@ -1,354 +0,0 @@
-import asyncio
-import json
-
-import pytest
-
-from galaxy.api.types import Room, Message
-from galaxy.api.errors import (
-    UnknownError, AuthenticationRequired, BackendNotAvailable, BackendTimeout, BackendError,
-    TooManyMessagesSent, IncoherentLastMessage, MessageNotFound
-)
-
-def test_send_message_success(plugin, readline, write):
-    request = {
-        "jsonrpc": "2.0",
-        "id": "3",
-        "method": "send_message",
-        "params": {
-            "room_id": "14",
-            "message": "Hello!"
-        }
-    }
-
-    readline.side_effect = [json.dumps(request), ""]
-    plugin.send_message.coro.return_value = None
-    asyncio.run(plugin.run())
-    plugin.send_message.assert_called_with(room_id="14", message="Hello!")
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "3",
-        "result": None
-    }
-
-@pytest.mark.parametrize("error,code,message", [
-    pytest.param(UnknownError, 0, "Unknown error", id="unknown_error"),
-    pytest.param(AuthenticationRequired, 1, "Authentication required", id="not_authenticated"),
-    pytest.param(BackendNotAvailable, 2, "Backend not available", id="backend_not_available"),
-    pytest.param(BackendTimeout, 3, "Backend timed out", id="backend_timeout"),
-    pytest.param(BackendError, 4, "Backend error", id="backend_error"),
-    pytest.param(TooManyMessagesSent, 300, "Too many messages sent", id="too_many_messages")
-])
-def test_send_message_failure(plugin, readline, write, error, code, message):
-    request = {
-        "jsonrpc": "2.0",
-        "id": "6",
-        "method": "send_message",
-        "params": {
-            "room_id": "15",
-            "message": "Bye"
-        }
-    }
-
-    readline.side_effect = [json.dumps(request), ""]
-    plugin.send_message.coro.side_effect = error()
-    asyncio.run(plugin.run())
-    plugin.send_message.assert_called_with(room_id="15", message="Bye")
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "6",
-        "error": {
-            "code": code,
-            "message": message
-        }
-    }
-
-def test_mark_as_read_success(plugin, readline, write):
-    request = {
-        "jsonrpc": "2.0",
-        "id": "7",
-        "method": "mark_as_read",
-        "params": {
-            "room_id": "14",
-            "last_message_id": "67"
-        }
-    }
-
-    readline.side_effect = [json.dumps(request), ""]
-    plugin.mark_as_read.coro.return_value = None
-    asyncio.run(plugin.run())
-    plugin.mark_as_read.assert_called_with(room_id="14", last_message_id="67")
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "7",
-        "result": None
-    }
-
-@pytest.mark.parametrize("error,code,message", [
-    pytest.param(UnknownError, 0, "Unknown error", id="unknown_error"),
-    pytest.param(AuthenticationRequired, 1, "Authentication required", id="not_authenticated"),
-    pytest.param(BackendNotAvailable, 2, "Backend not available", id="backend_not_available"),
-    pytest.param(BackendTimeout, 3, "Backend timed out", id="backend_timeout"),
-    pytest.param(BackendError, 4, "Backend error", id="backend_error"),
-    pytest.param(
-        IncoherentLastMessage,
-        400,
-        "Different last message id on backend",
-        id="incoherent_last_message"
-    )
-])
-def test_mark_as_read_failure(plugin, readline, write, error, code, message):
-    request = {
-        "jsonrpc": "2.0",
-        "id": "4",
-        "method": "mark_as_read",
-        "params": {
-            "room_id": "18",
-            "last_message_id": "7"
-        }
-    }
-
-    readline.side_effect = [json.dumps(request), ""]
-    plugin.mark_as_read.coro.side_effect = error()
-    asyncio.run(plugin.run())
-    plugin.mark_as_read.assert_called_with(room_id="18", last_message_id="7")
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "4",
-        "error": {
-            "code": code,
-            "message": message
-        }
-    }
-
-def test_get_rooms_success(plugin, readline, write):
-    request = {
-        "jsonrpc": "2.0",
-        "id": "2",
-        "method": "import_rooms"
-    }
-
-    readline.side_effect = [json.dumps(request), ""]
-    plugin.get_rooms.coro.return_value = [
-        Room("13", 0, None),
-        Room("15", 34, "8")
-    ]
-    asyncio.run(plugin.run())
-    plugin.get_rooms.assert_called_with()
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "2",
-        "result": {
-            "rooms": [
-                {
-                    "room_id": "13",
-                    "unread_message_count": 0,
-                },
-                {
-                    "room_id": "15",
-                    "unread_message_count": 34,
-                    "last_message_id": "8"
-                }
-            ]
-        }
-    }
-
-def test_get_rooms_failure(plugin, readline, write):
-    request = {
-        "jsonrpc": "2.0",
-        "id": "9",
-        "method": "import_rooms"
-    }
-
-    readline.side_effect = [json.dumps(request), ""]
-    plugin.get_rooms.coro.side_effect = UnknownError()
-    asyncio.run(plugin.run())
-    plugin.get_rooms.assert_called_with()
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "9",
-        "error": {
-            "code": 0,
-            "message": "Unknown error"
-        }
-    }
-
-def test_get_room_history_from_message_success(plugin, readline, write):
-    request = {
-        "jsonrpc": "2.0",
-        "id": "2",
-        "method": "import_room_history_from_message",
-        "params": {
-            "room_id": "34",
-            "message_id": "66"
-        }
-    }
-
-    readline.side_effect = [json.dumps(request), ""]
-    plugin.get_room_history_from_message.coro.return_value = [
-        Message("13", "149", 1549454837, "Hello"),
-        Message("14", "812", 1549454899, "Hi")
-    ]
-    asyncio.run(plugin.run())
-    plugin.get_room_history_from_message.assert_called_with(room_id="34", message_id="66")
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "2",
-        "result": {
-            "messages": [
-                {
-                    "message_id": "13",
-                    "sender_id": "149",
-                    "sent_time": 1549454837,
-                    "message_text": "Hello"
-                },
-                {
-                    "message_id": "14",
-                    "sender_id": "812",
-                    "sent_time": 1549454899,
-                    "message_text": "Hi"
-                }
-            ]
-        }
-    }
-
-@pytest.mark.parametrize("error,code,message", [
-    pytest.param(UnknownError, 0, "Unknown error", id="unknown_error"),
-    pytest.param(AuthenticationRequired, 1, "Authentication required", id="not_authenticated"),
-    pytest.param(BackendNotAvailable, 2, "Backend not available", id="backend_not_available"),
-    pytest.param(BackendTimeout, 3, "Backend timed out", id="backend_timeout"),
-    pytest.param(BackendError, 4, "Backend error", id="backend_error"),
-    pytest.param(MessageNotFound, 500, "Message not found", id="message_not_found")
-])
-def test_get_room_history_from_message_failure(plugin, readline, write, error, code, message):
-    request = {
-        "jsonrpc": "2.0",
-        "id": "7",
-        "method": "import_room_history_from_message",
-        "params": {
-            "room_id": "33",
-            "message_id": "88"
-        }
-    }
-
-    readline.side_effect = [json.dumps(request), ""]
-    plugin.get_room_history_from_message.coro.side_effect = error()
-    asyncio.run(plugin.run())
-    plugin.get_room_history_from_message.assert_called_with(room_id="33", message_id="88")
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "7",
-        "error": {
-            "code": code,
-            "message": message
-        }
-    }
-
-def test_get_room_history_from_timestamp_success(plugin, readline, write):
-    request = {
-        "jsonrpc": "2.0",
-        "id": "7",
-        "method": "import_room_history_from_timestamp",
-        "params": {
-            "room_id": "12",
-            "from_timestamp": 1549454835
-        }
-    }
-
-    readline.side_effect = [json.dumps(request), ""]
-    plugin.get_room_history_from_timestamp.coro.return_value = [
-        Message("12", "155", 1549454836, "Bye")
-    ]
-    asyncio.run(plugin.run())
-    plugin.get_room_history_from_timestamp.assert_called_with(
-        room_id="12",
-        from_timestamp=1549454835
-    )
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "7",
-        "result": {
-            "messages": [
-                {
-                    "message_id": "12",
-                    "sender_id": "155",
-                    "sent_time": 1549454836,
-                    "message_text": "Bye"
-                }
-            ]
-        }
-    }
-
-def test_get_room_history_from_timestamp_failure(plugin, readline, write):
-    request = {
-        "jsonrpc": "2.0",
-        "id": "3",
-        "method": "import_room_history_from_timestamp",
-        "params": {
-            "room_id": "10",
-            "from_timestamp": 1549454800
-        }
-    }
-
-    readline.side_effect = [json.dumps(request), ""]
-    plugin.get_room_history_from_timestamp.coro.side_effect = UnknownError()
-    asyncio.run(plugin.run())
-    plugin.get_room_history_from_timestamp.assert_called_with(
-        room_id="10",
-        from_timestamp=1549454800
-    )
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "3",
-        "error": {
-            "code": 0,
-            "message": "Unknown error"
-        }
-    }
-
-def test_update_room(plugin, write):
-    messages = [
-        Message("10", "898", 1549454832, "Hi")
-    ]
-
-    async def couritine():
-        plugin.update_room("14", 15, messages)
-
-    asyncio.run(couritine())
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "method": "chat_room_updated",
-        "params": {
-            "room_id": "14",
-            "unread_message_count": 15,
-            "messages": [
-                {
-                    "message_id": "10",
-                    "sender_id": "898",
-                    "sent_time": 1549454832,
-                    "message_text": "Hi"
-                }
-            ]
-        }
-    }

tests/test_chunk_messages.py

@@ -0,0 +1,63 @@
+import json
+
+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",
+        "params": {
+            "game_id": "3"
+        }
+    }
+
+    message = json.dumps(request).encode() + b"\n"
+    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")
+
+
+@pytest.mark.asyncio
+async def test_joined_messages(plugin, read):
+    requests = [
+        {
+            "jsonrpc": "2.0",
+            "method": "install_game",
+            "params": {
+                "game_id": "3"
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "launch_game",
+            "params": {
+                "game_id": "3"
+            }
+        }
+    ]
+    data = b"".join([json.dumps(request).encode() + b"\n" for request in requests])
+
+    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")
+
+
+@pytest.mark.asyncio
+async def test_not_finished(plugin, read):
+    request = {
+        "jsonrpc": "2.0",
+        "method": "install_game",
+        "params": {
+            "game_id": "3"
+        }
+    }
+
+    message = json.dumps(request).encode() # no new line
+    read.side_effect = [async_return_value(message), async_return_value(b"")]
+    await plugin.run()
+    plugin.install_game.assert_not_called()

tests/test_features.py

@@ -1,45 +1,53 @@
+from galaxy.api.consts import Feature, Platform
 from galaxy.api.plugin import Plugin
-from galaxy.api.consts import Platform, Feature
+
 
 def test_base_class():
     plugin = Plugin(Platform.Generic, "0.1", None, None, None)
-    assert plugin.features == []
+    assert set(plugin.features) == {
+        Feature.ImportInstalledGames,
+        Feature.ImportOwnedGames,
+        Feature.LaunchGame,
+        Feature.InstallGame,
+        Feature.UninstallGame,
+        Feature.ImportAchievements,
+        Feature.ImportGameTime,
+        Feature.ImportFriends,
+        Feature.ShutdownPlatformClient,
+        Feature.LaunchPlatformClient,
+        Feature.ImportGameLibrarySettings,
+        Feature.ImportOSCompatibility,
+        Feature.ImportUserPresence
+    }
+
 
 def test_no_overloads():
-    class PluginImpl(Plugin): #pylint: disable=abstract-method
+    class PluginImpl(Plugin):  # pylint: disable=abstract-method
         pass
 
     plugin = PluginImpl(Platform.Generic, "0.1", None, None, None)
     assert plugin.features == []
 
+
 def test_one_method_feature():
-    class PluginImpl(Plugin): #pylint: disable=abstract-method
+    class PluginImpl(Plugin):  # pylint: disable=abstract-method
         async def get_owned_games(self):
             pass
 
     plugin = PluginImpl(Platform.Generic, "0.1", None, None, None)
     assert plugin.features == [Feature.ImportOwnedGames]
 
-def test_multiple_methods_feature_all():
-    class PluginImpl(Plugin): #pylint: disable=abstract-method
-        async def send_message(self, room_id, message):
+
+def test_multi_features():
+    class PluginImpl(Plugin):  # pylint: disable=abstract-method
+        async def get_owned_games(self):
             pass
-        async def mark_as_read(self, room_id, last_message_id):
+
+        async def get_unlocked_achievements(self, game_id, context):
             pass
-        async def get_rooms(self):
-            pass
-        async def get_room_history_from_message(self, room_id, message_id):
-            pass
-        async def get_room_history_from_timestamp(self, room_id, timestamp):
+
+        async def get_game_time(self, game_id, context):
             pass
 
     plugin = PluginImpl(Platform.Generic, "0.1", None, None, None)
-    assert plugin.features == [Feature.Chat]
-
-def test_multiple_methods_feature_not_all():
-    class PluginImpl(Plugin): #pylint: disable=abstract-method
-        async def send_message(self, room_id, message):
-            pass
-
-    plugin = PluginImpl(Platform.Generic, "0.1", None, None, None)
-    assert plugin.features == []
+    assert set(plugin.features) == {Feature.ImportAchievements, Feature.ImportOwnedGames, Feature.ImportGameTime}

tests/test_friends.py

@@ -1,90 +1,96 @@
-import asyncio
-import json
-
 from galaxy.api.types import FriendInfo
 from galaxy.api.errors import UnknownError
+from galaxy.unittest.mock import async_return_value, skip_loop
+
+import pytest
+
+from tests import create_message, get_messages
 
 
-def test_get_friends_success(plugin, readline, write):
+@pytest.mark.asyncio
+async def test_get_friends_success(plugin, read, write):
     request = {
         "jsonrpc": "2.0",
         "id": "3",
         "method": "import_friends"
     }
 
-    readline.side_effect = [json.dumps(request), ""]
-    plugin.get_friends.coro.return_value = [
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+    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, readline, write):
+@pytest.mark.asyncio
+async def test_get_friends_failure(plugin, read, write):
     request = {
         "jsonrpc": "2.0",
         "id": "3",
         "method": "import_friends"
     }
 
-    readline.side_effect = [json.dumps(request), ""]
-    plugin.get_friends.coro.side_effect = UnknownError()
-    asyncio.run(plugin.run())
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+    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)
+    await skip_loop()
 
-    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")
+    await skip_loop()
 
-    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_library_settings.py

@@ -0,0 +1,196 @@
+from unittest.mock import call
+
+import pytest
+from galaxy.api.types import GameLibrarySettings
+from galaxy.api.errors import BackendError
+from galaxy.unittest.mock import async_return_value
+
+from tests import create_message, get_messages
+
+
+@pytest.mark.asyncio
+async def test_get_library_settings_success(plugin, read, write):
+    plugin.prepare_game_library_settings_context.return_value = async_return_value("abc")
+    request = {
+        "jsonrpc": "2.0",
+        "id": "3",
+        "method": "start_game_library_settings_import",
+        "params": {
+            "game_ids": ["3", "5", "7"]
+        }
+    }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+    plugin.get_game_library_settings.side_effect = [
+        async_return_value(GameLibrarySettings("3", None, True)),
+        async_return_value(GameLibrarySettings("5", [], False)),
+        async_return_value(GameLibrarySettings("7", ["tag1", "tag2", "tag3"], None)),
+    ]
+    await plugin.run()
+    plugin.get_game_library_settings.assert_has_calls([
+        call("3", "abc"),
+        call("5", "abc"),
+        call("7", "abc"),
+    ])
+    plugin.game_library_settings_import_complete.assert_called_once_with()
+
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "result": None
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "game_library_settings_import_success",
+            "params": {
+                "game_library_settings": {
+                    "game_id": "3",
+                    "hidden": True
+                }
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "game_library_settings_import_success",
+            "params": {
+                "game_library_settings": {
+                    "game_id": "5",
+                    "tags": [],
+                    "hidden": False
+                }
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "game_library_settings_import_success",
+            "params": {
+                "game_library_settings": {
+                    "game_id": "7",
+                    "tags": ["tag1", "tag2", "tag3"]
+                }
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "game_library_settings_import_finished",
+            "params": None
+        }
+    ]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("exception,code,message", [
+    (BackendError, 4, "Backend error"),
+    (KeyError, 0, "Unknown error")
+])
+async def test_get_game_library_settings_error(exception, code, message, plugin, read, write):
+    plugin.prepare_game_library_settings_context.return_value = async_return_value(None)
+    request = {
+        "jsonrpc": "2.0",
+        "id": "3",
+        "method": "start_game_library_settings_import",
+        "params": {
+            "game_ids": ["6"]
+        }
+    }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+    plugin.get_game_library_settings.side_effect = exception
+    await plugin.run()
+    plugin.get_game_library_settings.assert_called()
+    plugin.game_library_settings_import_complete.assert_called_once_with()
+
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "result": None
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "game_library_settings_import_failure",
+            "params": {
+                "game_id": "6",
+                "error": {
+                    "code": code,
+                    "message": message
+                }
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "game_library_settings_import_finished",
+            "params": None
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_prepare_get_game_library_settings_context_error(plugin, read, write):
+    plugin.prepare_game_library_settings_context.side_effect = BackendError()
+    request = {
+        "jsonrpc": "2.0",
+        "id": "3",
+        "method": "start_game_library_settings_import",
+        "params": {
+            "game_ids": ["6"]
+        }
+    }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+    await plugin.run()
+
+    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_game_library_settings_context.return_value = async_return_value(None)
+    requests = [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "method": "start_game_library_settings_import",
+            "params": {
+                "game_ids": ["6"]
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "id": "4",
+            "method": "start_game_library_settings_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"", 10)
+    ]
+
+    await plugin.run()
+
+    messages = get_messages(write)
+    assert {
+        "jsonrpc": "2.0",
+        "id": "3",
+        "result": None
+    } in messages
+    assert {
+        "jsonrpc": "2.0",
+        "id": "4",
+        "error": {
+            "code": 600,
+            "message": "Import already in progress"
+        }
+    } in messages
+

tests/test_game_times.py

@@ -1,175 +1,216 @@
-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, skip_loop
 
-def test_success(plugin, readline, 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"]
+        }
     }
-
-    readline.side_effect = [json.dumps(request), ""]
-    plugin.get_game_times.coro.return_value = [
-        GameTime("3", 60, 1549550504),
-        GameTime("5", 10, 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"),
+    ])
+    plugin.game_times_import_complete.assert_called_once_with()
 
-    assert response == {
+    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",
+                    "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
+        }
+    ]
+
+
+@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",
-        "result": {
-            "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()
+    plugin.game_times_import_complete.assert_called_once_with()
+
+    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
+        }
+    ]
+
+
+@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",
+        "method": "start_game_times_import",
+        "params": {
+            "game_ids": ["6"]
+        }
+    }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+    await plugin.run()
+
+    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_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"", 10)
+    ]
+
+    await plugin.run()
+
+    messages = get_messages(write)
+    assert {
+        "jsonrpc": "2.0",
+        "id": "3",
+        "result": None
+    } in messages
+    assert {
+        "jsonrpc": "2.0",
+        "id": "4",
+        "error": {
+            "code": 600,
+            "message": "Import already in progress"
+        }
+    } in messages
+
+
+@pytest.mark.asyncio
+async def test_update_game(plugin, write):
+    game_time = GameTime("3", 60, 1549550504)
+    plugin.update_game_time(game_time)
+    await skip_loop()
+
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "method": "game_time_updated",
+            "params": {
+                "game_time": {
                     "game_id": "3",
                     "time_played": 60,
                     "last_played_time": 1549550504
-                },
-                {
-                    "game_id": "5",
-                    "time_played": 10,
-                    "last_played_time": 1549550502
                 }
-            ]
-        }
-    }
-
-def test_failure(plugin, readline, write):
-    request = {
-        "jsonrpc": "2.0",
-        "id": "3",
-        "method": "import_game_times"
-    }
-
-    readline.side_effect = [json.dumps(request), ""]
-    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 response == {
-        "jsonrpc": "2.0",
-        "id": "3",
-        "error": {
-            "code": 0,
-            "message": "Unknown error",
-        }
-    }
-
-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
             }
         }
-    }
-
-@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
-            }
-        }
-    }
-
-@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])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "method": "game_time_import_failure",
-        "params": {
-            "game_id": "134",
-            "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")
-
-    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()

tests/test_http.py

@@ -0,0 +1,39 @@
+import asyncio
+from http import HTTPStatus
+
+import aiohttp
+import pytest
+from multidict import CIMultiDict, CIMultiDictProxy
+from yarl import URL
+
+from galaxy.api.errors import (
+    AccessDenied, AuthenticationRequired, BackendTimeout, BackendNotAvailable, BackendError, NetworkError,
+    TooManyRequests, UnknownBackendResponse, UnknownError
+)
+from galaxy.http import handle_exception
+
+request_info = aiohttp.RequestInfo(URL("http://o.pl"), "GET", CIMultiDictProxy(CIMultiDict()))
+
+@pytest.mark.parametrize(
+    "aiohttp_exception,expected_exception_type",
+    [
+        (asyncio.TimeoutError(), BackendTimeout),
+        (aiohttp.ServerDisconnectedError(), BackendNotAvailable),
+        (aiohttp.ClientConnectionError(), NetworkError),
+        (aiohttp.ContentTypeError(request_info, ()), UnknownBackendResponse),
+        (aiohttp.ClientResponseError(request_info, (), status=HTTPStatus.UNAUTHORIZED), AuthenticationRequired),
+        (aiohttp.ClientResponseError(request_info, (), status=HTTPStatus.FORBIDDEN), AccessDenied),
+        (aiohttp.ClientResponseError(request_info, (), status=HTTPStatus.SERVICE_UNAVAILABLE), BackendNotAvailable),
+        (aiohttp.ClientResponseError(request_info, (), status=HTTPStatus.TOO_MANY_REQUESTS), TooManyRequests),
+        (aiohttp.ClientResponseError(request_info, (), status=HTTPStatus.INTERNAL_SERVER_ERROR), BackendError),
+        (aiohttp.ClientResponseError(request_info, (), status=HTTPStatus.NOT_IMPLEMENTED), BackendError),
+        (aiohttp.ClientResponseError(request_info, (), status=HTTPStatus.BAD_REQUEST), UnknownError),
+        (aiohttp.ClientResponseError(request_info, (), status=HTTPStatus.NOT_FOUND), UnknownError),
+        (aiohttp.ClientError(), UnknownError)
+    ]
+)
+def test_handle_exception(aiohttp_exception, expected_exception_type):
+    with pytest.raises(expected_exception_type):
+        with handle_exception():
+            raise aiohttp_exception
+

tests/test_install_game.py

@@ -1,7 +1,12 @@
-import asyncio
-import json
+import pytest
 
-def test_success(plugin, readline):
+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, readline):
         }
     }
 
-    readline.side_effect = [json.dumps(request), ""]
-    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, readline, 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,53 +20,76 @@ def test_get_capabilites(reader, writer, readline, write):
     }
     token = "token"
     plugin = PluginImpl(Platform.Generic, "0.1", reader, writer, token)
-    readline.side_effect = [json.dumps(request), ""]
-    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, readline, write):
+
+@pytest.mark.asyncio
+async def test_shutdown(plugin, read, write):
     request = {
         "jsonrpc": "2.0",
         "id": "5",
         "method": "shutdown"
     }
-    readline.side_effect = [json.dumps(request)]
-    asyncio.run(plugin.run())
+    read.side_effect = [async_return_value(create_message(request))]
+    await plugin.run()
+    await plugin.wait_closed()
     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, readline, write):
+
+@pytest.mark.asyncio
+async def test_ping(plugin, read, write):
     request = {
         "jsonrpc": "2.0",
         "id": "7",
         "method": "ping"
     }
-    readline.side_effect = [json.dumps(request), ""]
-    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(plugin, readline):
-    readline.side_effect = [""]
-    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()
+
+
+@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 = [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, readline):
+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, readline):
         }
     }
 
-    readline.side_effect = [json.dumps(request), ""]
-    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_launch_platform_client.py

@@ -0,0 +1,17 @@
+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 = {
+        "jsonrpc": "2.0",
+        "method": "launch_platform_client"
+    }
+
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    plugin.launch_platform_client.return_value = async_return_value(None)
+    await plugin.run()
+    plugin.launch_platform_client.assert_called_with()

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, skip_loop
 
-def test_success(plugin, readline, 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"", 10)]
 
-    readline.side_effect = [json.dumps(request), ""]
-
-    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,43 @@ def test_success(plugin, readline, write):
         pytest.param(FailedParsingManifest, 200, "Failed parsing manifest", id="failed_parsing")
     ],
 )
-def test_failure(plugin, readline, write, error, code, message):
+async def test_failure(plugin, read, write, error, code, message):
     request = {
         "jsonrpc": "2.0",
         "id": "3",
         "method": "import_local_games"
     }
-
-    readline.side_effect = [json.dumps(request), ""]
-    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"", 10)]
+    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)
+    await skip_loop()
+
+    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_os_compatibility.py

@@ -0,0 +1,187 @@
+from unittest.mock import call
+
+import pytest
+from galaxy.api.consts import OSCompatibility
+from galaxy.api.errors import BackendError
+from galaxy.unittest.mock import async_return_value
+
+from tests import create_message, get_messages
+
+
+@pytest.mark.asyncio
+async def test_get_os_compatibility_success(plugin, read, write):
+    context = "abc"
+    plugin.prepare_os_compatibility_context.return_value = async_return_value(context)
+    request = {
+        "jsonrpc": "2.0",
+        "id": "11",
+        "method": "start_os_compatibility_import",
+        "params": {"game_ids": ["666", "13", "42"]}
+    }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+    plugin.get_os_compatibility.side_effect = [
+        async_return_value(OSCompatibility.Linux),
+        async_return_value(None),
+        async_return_value(OSCompatibility.Windows | OSCompatibility.MacOS),
+    ]
+    await plugin.run()
+    plugin.get_os_compatibility.assert_has_calls([
+        call("666", context),
+        call("13", context),
+        call("42", context),
+    ])
+    plugin.os_compatibility_import_complete.assert_called_once_with()
+
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "11",
+            "result": None
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "os_compatibility_import_success",
+            "params": {
+                "game_id": "666",
+                "os_compatibility": OSCompatibility.Linux.value
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "os_compatibility_import_success",
+            "params": {
+                "game_id": "13",
+                "os_compatibility": None
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "os_compatibility_import_success",
+            "params": {
+                "game_id": "42",
+                "os_compatibility": (OSCompatibility.Windows | OSCompatibility.MacOS).value
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "os_compatibility_import_finished",
+            "params": None
+        }
+    ]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("exception,code,message", [
+    (BackendError, 4, "Backend error"),
+    (KeyError, 0, "Unknown error")
+])
+async def test_get_os_compatibility_error(exception, code, message, plugin, read, write):
+    game_id = "6"
+    request_id = "55"
+    plugin.prepare_os_compatibility_context.return_value = async_return_value(None)
+    request = {
+        "jsonrpc": "2.0",
+        "id": request_id,
+        "method": "start_os_compatibility_import",
+        "params": {"game_ids": [game_id]}
+    }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+    plugin.get_os_compatibility.side_effect = exception
+    await plugin.run()
+    plugin.get_os_compatibility.assert_called()
+    plugin.os_compatibility_import_complete.assert_called_once_with()
+
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": request_id,
+            "result": None
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "os_compatibility_import_failure",
+            "params": {
+                "game_id": game_id,
+                "error": {
+                    "code": code,
+                    "message": message
+                }
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "os_compatibility_import_finished",
+            "params": None
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_prepare_get_os_compatibility_context_error(plugin, read, write):
+    request_id = "31415"
+    plugin.prepare_os_compatibility_context.side_effect = BackendError()
+    request = {
+        "jsonrpc": "2.0",
+        "id": request_id,
+        "method": "start_os_compatibility_import",
+        "params": {"game_ids": ["6"]}
+    }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+    await plugin.run()
+
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": request_id,
+            "error": {
+                "code": 4,
+                "message": "Backend error"
+            }
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_import_already_in_progress_error(plugin, read, write):
+    plugin.prepare_os_compatibility_context.return_value = async_return_value(None)
+    requests = [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "method": "start_os_compatibility_import",
+            "params": {
+                "game_ids": ["42"]
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "id": "4",
+            "method": "start_os_compatibility_import",
+            "params": {
+                "game_ids": ["666"]
+            }
+        }
+    ]
+    read.side_effect = [
+        async_return_value(create_message(requests[0])),
+        async_return_value(create_message(requests[1])),
+        async_return_value(b"", 10)
+    ]
+
+    await plugin.run()
+
+    responses = get_messages(write)
+    assert {
+        "jsonrpc": "2.0",
+        "id": "3",
+        "result": None
+    } in responses
+    assert {
+        "jsonrpc": "2.0",
+        "id": "4",
+        "error": {
+            "code": 600,
+            "message": "Import already in progress"
+        }
+    } in responses
+

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, skip_loop
 
-def test_success(plugin, readline, 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"", 10)]
 
-    readline.side_effect = [json.dumps(request), ""]
-    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,129 @@ def test_success(plugin, readline, 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, readline, write):
+
+@pytest.mark.asyncio
+async def test_failure(plugin, read, write):
     request = {
         "jsonrpc": "2.0",
         "id": "3",
         "method": "import_owned_games"
     }
 
-    readline.side_effect = [json.dumps(request), ""]
-    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"", 10)]
+    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)
+    await skip_loop()
+    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")
+    await skip_loop()
+    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)
+    await skip_loop()
+    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, skip_loop
+
+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, readline, 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,34 @@ def test_initialize_cache(plugin, readline, write, cache_data):
         "method": "initialize_cache",
         "params": {"data": cache_data}
     }
-    readline.side_effect = [json.dumps(request)]
+    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()
+    await skip_loop()
 
-        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()
+    await skip_loop()
 
-        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

@@ -0,0 +1,17 @@
+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 = {
+        "jsonrpc": "2.0",
+        "method": "shutdown_platform_client"
+    }
+
+    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

@@ -0,0 +1,46 @@
+import pytest
+
+from galaxy.reader import StreamLineReader
+from galaxy.unittest.mock import async_return_value
+
+
+@pytest.fixture()
+def stream_line_reader(reader):
+    return StreamLineReader(reader)
+
+
+@pytest.mark.asyncio
+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(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(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, readline):
+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, readline):
             "game_id": "3"
         }
     }
-
-    readline.side_effect = [json.dumps(request), ""]
+    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")

tests/test_user_presence.py

@@ -0,0 +1,231 @@
+from unittest.mock import call
+
+import pytest
+
+from galaxy.api.consts import PresenceState
+from galaxy.api.errors import BackendError
+from galaxy.api.types import UserPresence
+from galaxy.unittest.mock import async_return_value
+from tests import create_message, get_messages
+
+
+@pytest.mark.asyncio
+async def test_get_user_presence_success(plugin, read, write):
+    context = "abc"
+    user_ids = ["666", "13", "42", "69"]
+    plugin.prepare_user_presence_context.return_value = async_return_value(context)
+    request = {
+        "jsonrpc": "2.0",
+        "id": "11",
+        "method": "start_user_presence_import",
+        "params": {"user_ids": user_ids}
+    }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+    plugin.get_user_presence.side_effect = [
+        async_return_value(UserPresence(
+            PresenceState.Unknown,
+            "game-id1",
+            None,
+            "unknown state"
+        )),
+        async_return_value(UserPresence(
+            PresenceState.Offline,
+            None,
+            None,
+            "Going to grandma's house"
+        )),
+        async_return_value(UserPresence(
+            PresenceState.Online,
+            "game-id3",
+            "game-title3",
+            "Pew pew"
+        )),
+        async_return_value(UserPresence(
+            PresenceState.Away,
+            None,
+            "game-title4",
+            "AFKKTHXBY"
+        )),
+    ]
+    await plugin.run()
+    plugin.get_user_presence.assert_has_calls([
+        call(user_id, context) for user_id in user_ids
+    ])
+    plugin.user_presence_import_complete.assert_called_once_with()
+
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "11",
+            "result": None
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "user_presence_import_success",
+            "params": {
+                "user_id": "666",
+                "presence": {
+                    "presence_state": PresenceState.Unknown.value,
+                    "game_id": "game-id1",
+                    "presence_status": "unknown state"
+                }
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "user_presence_import_success",
+            "params": {
+                "user_id": "13",
+                "presence": {
+                    "presence_state": PresenceState.Offline.value,
+                    "presence_status": "Going to grandma's house"
+                }
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "user_presence_import_success",
+            "params": {
+                "user_id": "42",
+                "presence": {
+                    "presence_state": PresenceState.Online.value,
+                    "game_id": "game-id3",
+                    "game_title": "game-title3",
+                    "presence_status": "Pew pew"
+                }
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "user_presence_import_success",
+            "params": {
+                "user_id": "69",
+                "presence": {
+                    "presence_state": PresenceState.Away.value,
+                    "game_title": "game-title4",
+                    "presence_status": "AFKKTHXBY"
+                }
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "user_presence_import_finished",
+            "params": None
+        }
+    ]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("exception,code,message", [
+    (BackendError, 4, "Backend error"),
+    (KeyError, 0, "Unknown error")
+])
+async def test_get_user_presence_error(exception, code, message, plugin, read, write):
+    user_id = "69"
+    request_id = "55"
+    plugin.prepare_user_presence_context.return_value = async_return_value(None)
+    request = {
+        "jsonrpc": "2.0",
+        "id": request_id,
+        "method": "start_user_presence_import",
+        "params": {"user_ids": [user_id]}
+    }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+    plugin.get_user_presence.side_effect = exception
+    await plugin.run()
+    plugin.get_user_presence.assert_called()
+    plugin.user_presence_import_complete.assert_called_once_with()
+
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": request_id,
+            "result": None
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "user_presence_import_failure",
+            "params": {
+                "user_id": user_id,
+                "error": {
+                    "code": code,
+                    "message": message
+                }
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "user_presence_import_finished",
+            "params": None
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_prepare_get_user_presence_context_error(plugin, read, write):
+    request_id = "31415"
+    plugin.prepare_user_presence_context.side_effect = BackendError()
+    request = {
+        "jsonrpc": "2.0",
+        "id": request_id,
+        "method": "start_user_presence_import",
+        "params": {"user_ids": ["6"]}
+    }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+    await plugin.run()
+
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": request_id,
+            "error": {
+                "code": 4,
+                "message": "Backend error"
+            }
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_import_already_in_progress_error(plugin, read, write):
+    plugin.prepare_user_presence_context.return_value = async_return_value(None)
+    requests = [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "method": "start_user_presence_import",
+            "params": {
+                "user_ids": ["42"]
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "id": "4",
+            "method": "start_user_presence_import",
+            "params": {
+                "user_ids": ["666"]
+            }
+        }
+    ]
+    read.side_effect = [
+        async_return_value(create_message(requests[0])),
+        async_return_value(create_message(requests[1])),
+        async_return_value(b"", 10)
+    ]
+
+    await plugin.run()
+
+    responses = get_messages(write)
+    assert {
+        "jsonrpc": "2.0",
+        "id": "3",
+        "result": None
+    } in responses
+    assert {
+         "jsonrpc": "2.0",
+         "id": "4",
+         "error": {
+             "code": 600,
+             "message": "Import already in progress"
+         }
+    } in responses

tests/test_users.py

@@ -1,69 +0,0 @@
-import asyncio
-import json
-
-from galaxy.api.types import UserInfo, Presence
-from galaxy.api.errors import UnknownError
-from galaxy.api.consts import PresenceState
-
-
-def test_get_users_success(plugin, readline, write):
-    request = {
-        "jsonrpc": "2.0",
-        "id": "8",
-        "method": "import_user_infos",
-        "params": {
-            "user_id_list": ["13"]
-        }
-    }
-
-    readline.side_effect = [json.dumps(request), ""]
-    plugin.get_users.coro.return_value = [
-        UserInfo("5", False, "Ula", "http://avatar.png", Presence(PresenceState.Offline))
-    ]
-    asyncio.run(plugin.run())
-    plugin.get_users.assert_called_with(user_id_list=["13"])
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "8",
-        "result": {
-            "user_info_list": [
-                {
-                    "user_id": "5",
-                    "is_friend": False,
-                    "user_name": "Ula",
-                    "avatar_url": "http://avatar.png",
-                    "presence": {
-                        "presence_state": "offline"
-                    }
-                }
-            ]
-        }
-    }
-
-
-def test_get_users_failure(plugin, readline, write):
-    request = {
-        "jsonrpc": "2.0",
-        "id": "12",
-        "method": "import_user_infos",
-        "params": {
-            "user_id_list": ["10", "11", "12"]
-        }
-    }
-
-    readline.side_effect = [json.dumps(request), ""]
-    plugin.get_users.coro.side_effect = UnknownError()
-    asyncio.run(plugin.run())
-    plugin.get_users.assert_called_with(user_id_list=["10", "11", "12"])
-    response = json.loads(write.call_args[0][0])
-
-    assert response == {
-        "jsonrpc": "2.0",
-        "id": "12",
-        "error": {
-            "code": 0,
-            "message": "Unknown error"
-        }
-    }