Increment version

+ other small imporvements in docs

.github/ISSUE_TEMPLATE/config.yml

@@ -0,0 +1,11 @@
+blank_issues_enabled: true
+contact_links:
+  - name: GOG GALAXY 2.0 issue
+    url: https://mantis2.gog.com/
+    about: Report issues related to GOG GALAXY 2.0, official integrations or the whole ecosystem
+  - name: Platform ID request
+    url: https://github.com/gogcom/galaxy-integrations-python-api/issues/160
+    about: Report missing platform id
+  - name: Community integrations
+    url: https://github.com/Mixaill/awesome-gog-galaxy
+    about: Find integrations and their maintainers, request new integrations or report issues related to unofficial integrations.

.github/ISSUE_TEMPLATE/problem_report.md

@@ -0,0 +1,14 @@
+---
+name: API issue
+about: Report a bug or problem with current API architecture
+
+---
+
+**Problem**
+<!-- Describe the problem you faced. -->
+
+**Solution**
+<!-- Describe the solution you'd like. -->
+
+**Alternatives**
+<!-- Optionally describe possible alternatives or current workarounds if any. -->

.gitignore

@@ -7,3 +7,5 @@ docs/build/
 Pipfile
 .idea
 docs/source/_build
+.mypy_cache
+.pytest_cache

.gitlab-ci.yml

@@ -1,4 +1,4 @@
-image: registry-gitlab.gog.com/galaxy-client/gitlab-ci-tools:latest
+image: registry-gitlab.gog.com/docker/python:3.7.3
 
 stages:
   - test
@@ -14,13 +14,19 @@ test_package:
 
 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 upload -r gog-pypi
+    - 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
+    - tags

PLATFORM_IDs.md

@@ -4,10 +4,10 @@ Platform ID list for GOG Galaxy 2.0 Integrations
 
 | ID | Name |
 | --- | --- |
+| test | Testing purposes |
 | steam	| Steam |
 | psn	| PlayStation Network |
 | xboxone	| Xbox Live |
-| generic	| Manually added games |
 | origin	| Origin |
 | uplay	| Uplay |
 | battlenet	| Battle.net |
@@ -20,7 +20,7 @@ Platform ID list for GOG Galaxy 2.0 Integrations
 | nswitch	| Nintendo Switch |
 | nwiiu	| Nintendo Wii U |
 | nwii	| Nintendo Wii |
-| ncube	| Nintendo Game Cube |
+| ncube	| Nintendo GameCube |
 | riot	| Riot |
 | wargaming	| Wargaming |
 | ngameboy	| Nintendo Game Boy |
@@ -58,25 +58,34 @@ Platform ID list for GOG Galaxy 2.0 Integrations
 | bb	| BestBuy |
 | gameuk	| Game UK |
 | fanatical	| Fanatical store |
-| playasia 	| PlayAsia |
+| playasia 	| Play-Asia |
 | stadia	| Google Stadia |
 | arc	| ARC |
 | eso	| ESO |
 | glyph	| Trion World |
 | aionl	| Aion: Legions of War |
 | aion	| Aion |
-| blade	| Blade and Soul |
+| blade	| Blade & Soul |
 | gw	| Guild Wars |
 | gw2	| Guild Wars 2 |
 | lin2	| Lineage 2 |
 | ffxi	| Final Fantasy XI |
-| ffxiv	| Final Fantasy XIV |
+| ffxiv	| Final Fantasy XIV  |
-| totalwar	| TotalWar |
+| totalwar	| Total War |
 | winstore	| Windows Store |
 | elites	| Elite Dangerous |
 | star	| Star Citizen |
-| psp	| Playstation Portable |
+| psp	| PlayStation Portable |
-| psvita	| Playstation Vita |
+| psvita	| PlayStation Vita |
 | nds	| Nintendo DS |
 | 3ds	| Nintendo 3DS |
 | pathofexile | Path of Exile |
+| twitch | Twitch |
+| minecraft | Minecraft |
+| gamesessions | GameSessions |
+| nuuvem | Nuuvem |
+| fxstore | FX Store |
+| indiegala | IndieGala |
+| playfire | Playfire |
+| oculus | Oculus |
+| rockstar | Rockstar |

README.md

@@ -36,20 +36,31 @@ Communication between an integration and the client is also possible with the us
 import sys
 from galaxy.api.plugin import Plugin, create_and_run_plugin
 from galaxy.api.consts import Platform
+from galaxy.api.types import Authentication, Game, LicenseInfo, LicenseType
+
 
 class PluginExample(Plugin):
     def __init__(self, reader, writer, token):
         super().__init__(
-            Platform.Generic, # Choose platform from available list
+            Platform.Test,  # choose platform from available list
-            "0.1", # Version
+            "0.1",  # version
             reader,
             writer,
             token
         )
 
     # implement methods
+
+    # required
     async def authenticate(self, stored_credentials=None):
-        pass
+        return Authentication('test_user_id', 'Test User Name')
+
+    # required
+    async def get_owned_games(self):
+        return [
+            Game('test', 'The Test', None, LicenseInfo(LicenseType.SinglePurchase))
+        ]
+
 
 def main():
     create_and_run_plugin(PluginExample, sys.argv)
@@ -76,6 +87,20 @@ 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`
 
+### Logging
+<a href='https://docs.python.org/3.7/howto/logging.html'>Root logger</a> is already setup by GOG Galaxy to store rotated log files in:
+
+- Windows:
+
+    `%programdata%\GOG.com\Galaxy\logs`
+
+- macOS:
+
+    `/Users/Shared/GOG.com/Galaxy/Logs`
+
+Plugin logs are kept in `plugin-<platform>-<guid>.log`.
+When debugging, inspecting the other side of communication in the `GalaxyClient.log` can be helpful as well.
+
 ### Manifest
 
 <a name="deploy-manifest"></a>
@@ -84,8 +109,8 @@ Obligatory JSON file to be placed in an integration folder.
 ```json
 {
     "name": "Example plugin",
-    "platform": "generic",
+    "platform": "test",
-    "guid": "UNIQUE-GUID",
+    "guid": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
     "version": "0.1",
     "description": "Example plugin",
     "author": "Name",
@@ -97,9 +122,8 @@ Obligatory JSON file to be placed in an integration folder.
 
 | property      | description |
 |---------------|---|
-| `guid`        |   |
+| `guid`        | custom Globally Unique Identifier |
-| `description` |   |
+| `version`     | the same string as `version` in `Plugin` constructor |
-| `url`         |   |
 | `script`      | path of the entry point module, relative to the integration folder |
 
 ### Dependencies

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,8 @@ GOG Galaxy Integrations Python API
    :includehidden:
 
    Overview <overview>
-   API <galaxy.api>
+   galaxy.api
+   galaxy.http
    Platform ID's <platforms>
 
 Index

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

@@ -1,9 +1,10 @@
 -e .
-pytest==4.2.0
+pytest==5.2.2
 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
-psutil==5.6.3; sys_platform == 'darwin'
+psutil==5.6.6; sys_platform == 'darwin'

setup.py

@@ -2,14 +2,15 @@ from setuptools import setup, find_packages
 
 setup(
     name="galaxy.plugin.api",
-    version="0.45",
+    version="0.68",
     description="GOG Galaxy Integrations Python API",
     author='Galaxy team',
     author_email='galaxy@gog.com',
     packages=find_packages("src"),
     package_dir={'': 'src'},
     install_requires=[
-        "aiohttp==3.5.4",
+        "aiohttp>=3.5.4",
-        "certifi==2019.3.9"
+        "certifi>=2019.3.9",
+        "psutil>=5.6.6; sys_platform == 'darwin'"
     ]
 )

src/galaxy/__init__.py

@@ -1 +1,6 @@
-__path__ = __import__('pkgutil').extend_path(__path__, __name__)
+import logging
+
+
+logging.getLogger(__name__).setLevel(logging.INFO)
+
+__path__: str = __import__('pkgutil').extend_path(__path__, __name__)  # type: ignore

src/galaxy/api/consts.py

@@ -81,6 +81,17 @@ class Platform(Enum):
     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):
     """Possible features that can be implemented by an integration.
@@ -99,6 +110,13 @@ class Feature(Enum):
     VerifyGame = "VerifyGame"
     ImportFriends = "ImportFriends"
     ShutdownPlatformClient = "ShutdownPlatformClient"
+    LaunchPlatformClient = "LaunchPlatformClient"
+    ImportGameLibrarySettings = "ImportGameLibrarySettings"
+    ImportOSCompatibility = "ImportOSCompatibility"
+    ImportUserPresence = "ImportUserPresence"
+    ImportLocalSize = "ImportLocalSize"
+    ImportSubscriptions = "ImportSubscriptions"
+    ImportSubscriptionGames = "ImportSubscriptionGames"
 
 
 class LicenseType(Enum):
@@ -117,3 +135,30 @@ class LocalGameState(Flag):
     None_ = 0
     Installed = 1
     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 of a user."""
+    Unknown = "unknown"
+    Online = "online"
+    Offline = "offline"
+    Away = "away"
+
+
+class SubscriptionDiscovery(Flag):
+    """Possible capabilities which inform what methods of subscriptions ownership detection are supported.
+
+    :param AUTOMATIC: integration can retrieve the proper status of subscription ownership.
+    :param USER_ENABLED: integration can handle override of ~class::`Subscription.owned` value to True
+    """
+    AUTOMATIC = 1
+    USER_ENABLED = 2

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):
@@ -20,7 +20,7 @@ class BackendError(ApplicationError):
 
 class UnknownBackendResponse(ApplicationError):
     def __init__(self, data=None):
-        super().__init__(4, "Backend responded in uknown way", data)
+        super().__init__(4, "Backend responded in unknown way", data)
 
 class TooManyRequests(ApplicationError):
     def __init__(self, data=None):

src/galaxy/api/importer.py

@@ -0,0 +1,102 @@
+import asyncio
+import logging
+from galaxy.api.jsonrpc import ApplicationError
+from galaxy.api.errors import ImportInProgress, UnknownError
+
+logger = logging.getLogger(__name__)
+
+
+class Importer:
+    def __init__(
+            self,
+            task_manger,
+            name,
+            get,
+            prepare_context,
+            notification_success,
+            notification_failure,
+            notification_finished,
+            complete,
+    ):
+        self._task_manager = task_manger
+        self._name = name
+        self._get = get
+        self._prepare_context = prepare_context
+        self._notification_success = notification_success
+        self._notification_failure = notification_failure
+        self._notification_finished = notification_finished
+        self._complete = complete
+
+        self._import_in_progress = False
+
+    async def _import_element(self, id_, context_):
+        try:
+            element = await self._get(id_, context_)
+            self._notification_success(id_, element)
+        except ApplicationError as error:
+            self._notification_failure(id_, error)
+        except asyncio.CancelledError:
+            pass
+        except Exception:
+            logger.exception("Unexpected exception raised in %s importer", self._name)
+            self._notification_failure(id_, UnknownError())
+
+    async def _import_elements(self, ids_, context_):
+        try:
+            imports = [self._import_element(id_, context_) for id_ in ids_]
+            await asyncio.gather(*imports)
+            self._notification_finished()
+            self._complete()
+        except asyncio.CancelledError:
+            logger.debug("Importing %s cancelled", self._name)
+        finally:
+            self._import_in_progress = False
+
+    async def start(self, ids):
+        if self._import_in_progress:
+            raise ImportInProgress()
+
+        self._import_in_progress = True
+        try:
+            context = await self._prepare_context(ids)
+            self._task_manager.create_task(
+                self._import_elements(ids, context),
+                "{} import".format(self._name),
+                handle_exceptions=False
+            )
+        except:
+            self._import_in_progress = False
+            raise
+
+
+class CollectionImporter(Importer):
+    def __init__(self, notification_partially_finished, *args):
+        super().__init__(*args)
+        self._notification_partially_finished = notification_partially_finished
+
+    async def _import_element(self, id_, context_):
+        try:
+            async for element in self._get(id_, context_):
+                self._notification_success(id_, element)
+        except ApplicationError as error:
+            self._notification_failure(id_, error)
+        except asyncio.CancelledError:
+            pass
+        except Exception:
+            logger.exception("Unexpected exception raised in %s importer", self._name)
+            self._notification_failure(id_, UnknownError())
+        finally:
+            self._notification_partially_finished(id_)
+
+
+class SynchroneousImporter(Importer):
+    async def _import_elements(self, ids_, context_):
+        try:
+            for id_ in ids_:
+                await self._import_element(id_, context_)
+            self._notification_finished()
+            self._complete()
+        except asyncio.CancelledError:
+            logger.debug("Importing %s cancelled", self._name)
+        finally:
+            self._import_in_progress = False

src/galaxy/api/jsonrpc.py

@@ -6,6 +6,11 @@ import inspect
 import json
 
 from galaxy.reader import StreamLineReader
+from galaxy.task_manager import TaskManager
+
+
+logger = logging.getLogger(__name__)
+
 
 class JsonRpcError(Exception):
     def __init__(self, code, message, data=None):
@@ -17,6 +22,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["data"] = self.data
+
+        return obj
+
 class ParseError(JsonRpcError):
     def __init__(self):
         super().__init__(-32700, "Parse error")
@@ -52,7 +68,9 @@ 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"])
+Response = namedtuple("Response", ["id", "result", "error"], defaults=[None, {}, {}])
+Method = namedtuple("Method", ["callback", "signature", "immediate", "sensitive_params"])
+
 
 def anonymise_sensitive_params(params, sensitive_params):
     anomized_data = "****"
@@ -66,7 +84,7 @@ def anonymise_sensitive_params(params, sensitive_params):
 
     return params
 
-class Server():
+class Connection():
     def __init__(self, reader, writer, encoder=json.JSONEncoder()):
         self._active = True
         self._reader = StreamLineReader(reader)
@@ -74,9 +92,11 @@ class Server():
         self._encoder = encoder
         self._methods = {}
         self._notifications = {}
-        self._eof_listeners = []
+        self._task_manager = TaskManager("jsonrpc server")
+        self._last_request_id = 0
+        self._requests_futures = {}
 
-    def register_method(self, name, callback, internal, sensitive_params=False):
+    def register_method(self, name, callback, immediate, sensitive_params=False):
         """
         Register method
 
@@ -86,9 +106,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
 
@@ -98,10 +118,48 @@ 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)
+        self._notifications[name] = Method(callback, inspect.signature(callback), immediate, sensitive_params)
 
-    def register_eof(self, callback):
+    async def send_request(self, method, params, sensitive_params):
-        self._eof_listeners.append(callback)
+        """
+        Send request
+
+        :param method:
+        :param params:
+        :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._last_request_id += 1
+        request_id = str(self._last_request_id)
+
+        loop = asyncio.get_running_loop()
+        future = loop.create_future()
+        self._requests_futures[self._last_request_id] = (future, sensitive_params)
+
+        logger.info(
+            "Sending request: id=%s, method=%s, params=%s",
+            request_id, method, anonymise_sensitive_params(params, sensitive_params)
+        )
+
+        self._send_request(request_id, method, params)
+        return await future
+
+    def send_notification(self, method, params, sensitive_params=False):
+        """
+        Send notification
+
+        :param method:
+        :param params:
+        :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
+        """
+
+        logger.info(
+            "Sending notification: method=%s, params=%s",
+            method, anonymise_sensitive_params(params, sensitive_params)
+        )
+
+        self._send_notification(method, params)
 
     async def run(self):
         while self._active:
@@ -114,38 +172,66 @@ class Server():
                 self._eof()
                 continue
             data = data.strip()
-            logging.debug("Received %d bytes of data", len(data))
+            logger.debug("Received %d bytes of data", len(data))
             self._handle_input(data)
             await asyncio.sleep(0) # To not starve task queue
 
-    def stop(self):
+    def close(self):
-        self._active = False
+        if self._active:
+            logger.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")
+        logger.info("Received EOF")
-        self.stop()
+        self.close()
-        for listener in self._eof_listeners:
-            listener()
 
     def _handle_input(self, data):
         try:
-            request = self._parse_request(data)
+            message = self._parse_message(data)
         except JsonRpcError as error:
             self._send_error(None, error)
             return
 
-        if request.id is not None:
+        if isinstance(message, Request):
-            self._handle_request(request)
+            if message.id is not None:
-        else:
+                self._handle_request(message)
-            self._handle_notification(request)
+            else:
+                self._handle_notification(message)
+        elif isinstance(message, Response):
+            self._handle_response(message)
+
+    def _handle_response(self, response):
+        request_future = self._requests_futures.get(int(response.id))
+        if request_future is None:
+            response_type = "response" if response.result is not None else "error"
+            logger.warning("Received %s for unknown request: %s", response_type, response.id)
+            return
+
+        future, sensitive_params = request_future
+
+        if response.error:
+            error = JsonRpcError(
+                response.error.setdefault("code", 0),
+                response.error.setdefault("message", ""),
+                response.error.setdefault("data", None)
+            )
+            self._log_error(response, error, sensitive_params)
+            future.set_exception(error)
+            return
+
+        self._log_response(response, sensitive_params)
+        future.set_result(response.result)
 
     def _handle_notification(self, request):
         method = self._notifications.get(request.method)
         if not method:
-            logging.error("Received unknown notification: %s", request.method)
+            logger.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:
@@ -153,23 +239,22 @@ class Server():
         except TypeError:
             self._send_error(request.id, InvalidParams())
 
-        if internal:
+        if immediate:
-            # internal requests are handled immediately
             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")
+                logger.exception("Unexpected exception raised in notification handler")
 
     def _handle_request(self, request):
         method = self._methods.get(request.method)
         if not method:
-            logging.error("Received unknown request: %s", request.method)
+            logger.error("Received unknown request: %s", request.method)
             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:
@@ -177,8 +262,7 @@ class Server():
         except TypeError:
             self._send_error(request.id, InvalidParams())
 
-        if internal:
+        if immediate:
-            # internal requests are handled immediately
             response = callback(*bound_args.args, **bound_args.kwargs)
             self._send_response(request.id, response)
         else:
@@ -190,34 +274,42 @@ 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")
+                    logger.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):
+    def _parse_message(data):
         try:
-            jsonrpc_request = json.loads(data, encoding="utf-8")
+            jsonrpc_message = json.loads(data, encoding="utf-8")
-            if jsonrpc_request.get("jsonrpc") != "2.0":
+            if jsonrpc_message.get("jsonrpc") != "2.0":
                 raise InvalidRequest()
-            del jsonrpc_request["jsonrpc"]
+            del jsonrpc_message["jsonrpc"]
-            return Request(**jsonrpc_request)
+            if "result" in jsonrpc_message.keys() or "error" in jsonrpc_message.keys():
+                return Response(**jsonrpc_message)
+            else:
+                return Request(**jsonrpc_message)
+
         except json.JSONDecodeError:
             raise ParseError()
         except TypeError:
             raise InvalidRequest()
 
-    def _send(self, data):
+    def _send(self, data, sensitive=True):
         try:
             line = self._encoder.encode(data)
-            logging.debug("Sending data: %s", line)
             data = (line + "\n").encode("utf-8")
+            if sensitive:
+                logger.debug("Sending %d bytes of data", len(data))
+            else:
+                logger.debug("Sending data: %s", line)
             self._writer.write(data)
-            asyncio.create_task(self._writer.drain())
         except TypeError as error:
-            logging.error(str(error))
+            logger.error(str(error))
 
     def _send_response(self, request_id, result):
         response = {
@@ -225,65 +317,51 @@ class Server():
             "id": request_id,
             "result": result
         }
-        self._send(response)
+        self._send(response, sensitive=False)
 
     def _send_error(self, request_id, error):
         response = {
             "jsonrpc": "2.0",
             "id": request_id,
-            "error": {
+            "error": error.json()
-                "code": error.code,
-                "message": error.message
-            }
         }
 
-        if error.data is not None:
+        self._send(response, sensitive=False)
-            response["error"]["data"] = error.data
 
-        self._send(response)
+    def _send_request(self, request_id, method, params):
+        request = {
+            "jsonrpc": "2.0",
+            "method": method,
+            "id": request_id,
+            "params": params
+        }
+        self._send(request, sensitive=True)
 
-    @staticmethod
+    def _send_notification(self, method, params):
-    def _log_request(request, sensitive_params):
-        params = anonymise_sensitive_params(request.params, sensitive_params)
-        if request.id is not None:
-            logging.info("Handling request: id=%s, method=%s, params=%s", request.id, request.method, params)
-        else:
-            logging.info("Handling notification: method=%s, params=%s", request.method, params)
-
-class NotificationClient():
-    def __init__(self, writer, encoder=json.JSONEncoder()):
-        self._writer = writer
-        self._encoder = encoder
-        self._methods = {}
-
-    def notify(self, method, params, sensitive_params=False):
-        """
-        Send notification
-
-        :param method:
-        :param params:
-        :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
-        """
         notification = {
             "jsonrpc": "2.0",
             "method": method,
             "params": params
         }
-        self._log(method, params, sensitive_params)
+        self._send(notification, sensitive=True)
-        self._send(notification)
-
-    def _send(self, data):
-        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())
-        except TypeError as error:
-            logging.error("Failed to parse outgoing message: %s", str(error))
 
     @staticmethod
-    def _log(method, params, sensitive_params):
+    def _log_request(request, sensitive_params):
-        params = anonymise_sensitive_params(params, sensitive_params)
+        params = anonymise_sensitive_params(request.params, sensitive_params)
-        logging.info("Sending notification: method=%s, params=%s", method, params)
+        if request.id is not None:
+            logger.info("Handling request: id=%s, method=%s, params=%s", request.id, request.method, params)
+        else:
+            logger.info("Handling notification: method=%s, params=%s", request.method, params)
+
+    @staticmethod
+    def _log_response(response, sensitive_params):
+        result = anonymise_sensitive_params(response.result, sensitive_params)
+        logger.info("Handling response: id=%s, result=%s", response.id, result)
+
+    @staticmethod
+    def _log_error(response, error, sensitive_params):
+        params = error.data if error.data is not None else {}
+        data = anonymise_sensitive_params(params, sensitive_params)
+        logger.info("Handling error: id=%s, code=%s, description=%s, data=%s",
+            response.id, error.code, error.message, data
+        )

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, SubscriptionDiscovery
 
-from galaxy.api.consts import LicenseType, LocalGameState
 
 @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:
 
@@ -58,17 +61,20 @@ class NextStep():
             if not stored_credentials:
                 return NextStep("web_session", PARAMS, cookies=COOKIES, js=JS)
 
-    :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 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.
+    :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
@@ -77,8 +83,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
@@ -89,8 +96,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
@@ -103,8 +111,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
@@ -119,8 +128,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
@@ -129,9 +139,14 @@ class LocalGame():
     game_id: str
     local_game_state: LocalGameState
 
+
 @dataclass
-class FriendInfo():
+class FriendInfo:
-    """Information about a friend of the currently authenticated user.
+    """
+    .. deprecated:: 0.56
+      Use :class:`UserInfo`.
+
+    Information about a friend of the currently authenticated user.
 
     :param user_id: id of the user
     :param user_name: username of the user
@@ -139,15 +154,104 @@ class FriendInfo():
     user_id: str
     user_name: str
 
+
 @dataclass
-class GameTime():
+class UserInfo:
+    """Information about a user of related user.
+
+    :param user_id: id of the user
+    :param user_name: username of the user
+    :param avatar_url: the URL of the user avatar
+    :param profile_url: the URL of the user profile
+    """
+    user_id: str
+    user_name: str
+    avatar_url: Optional[str] = None
+    profile_url: Optional[str] = None
+
+
+@dataclass
+class GameTime:
     """Game time of a game, defines the total time spent in the game
     and the last time the game was played.
 
     :param game_id: id of the related game
     :param time_played: the total time spent in the game in **minutes**
-    :param last_time_played: last time the game was played (**unix timestamp**)
+    :param last_played_time: last time the game was played (**unix timestamp**)
     """
     game_id: str
     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 client
+    """
+    game_id: str
+    tags: Optional[List[str]]
+    hidden: Optional[bool]
+
+
+@dataclass
+class UserPresence:
+    """Presence information of a user.
+
+    The GOG Galaxy client will prefer to generate user status basing on `game_id` (or `game_title`)
+    and `in_game_status` fields but if plugin is not capable of delivering it then the `full_status` will be used if
+    available
+
+    :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 in_game_status: status set by the game itself e.x. "In Main Menu"
+    :param full_status: full user status e.x. "Playing <title_name>: <in_game_status>"
+    """
+    presence_state: PresenceState
+    game_id: Optional[str] = None
+    game_title: Optional[str] = None
+    in_game_status: Optional[str] = None
+    full_status: Optional[str] = None
+
+
+@dataclass
+class Subscription:
+    """Information about a subscription.
+
+    :param subscription_name: name of the subscription, will also be used as its identifier.
+    :param owned: whether the subscription is owned or not, None if unknown.
+    :param end_time: unix timestamp of when the subscription ends, None if unknown.
+    :param subscription_discovery: combination of settings that can be manually
+        chosen by user to determine subscription handling behaviour. For example, if the integration cannot retrieve games
+        for subscription when user doesn't own it, then USER_ENABLED should not be used.
+        If the integration cannot determine subscription ownership for a user then AUTOMATIC should not be used.
+
+    """
+    subscription_name: str
+    owned: Optional[bool] = None
+    end_time: Optional[int] = None
+    subscription_discovery: SubscriptionDiscovery = SubscriptionDiscovery.AUTOMATIC | \
+                                                                 SubscriptionDiscovery.USER_ENABLED
+
+    def __post_init__(self):
+        assert self.subscription_discovery in [SubscriptionDiscovery.AUTOMATIC, SubscriptionDiscovery.USER_ENABLED,
+                                               SubscriptionDiscovery.AUTOMATIC | SubscriptionDiscovery.USER_ENABLED]
+
+
+@dataclass
+class SubscriptionGame:
+    """Information about a game from a subscription.
+
+    :param game_title: title of the game
+    :param game_id: id of the game
+    :param start_time: unix timestamp of when the game has been added to subscription
+    :param end_time: unix timestamp of when the game will be removed from subscription.
+    """
+    game_title: str
+    game_id: str
+    start_time: Optional[int] = None
+    end_time: Optional[int] = None

src/galaxy/http.py

@@ -1,3 +1,33 @@
+"""
+This module standardizes 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.
+Exemplary simple web service could looks like:
+
+    .. code-block:: python
+
+        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
@@ -13,17 +43,25 @@ from galaxy.api.errors import (
 )
 
 
+logger = logging.getLogger(__name__)
+
+#: Default limit of the simultaneous connections for ssl connector.
 DEFAULT_LIMIT = 20
-DEFAULT_TIMEOUT = 60  # seconds
+#: Default timeout in seconds used for client session.
+DEFAULT_TIMEOUT = 60
 
 
 class HttpClient:
-    """Deprecated"""
+    """
+    .. 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):
@@ -31,23 +69,52 @@ class HttpClient:
             return await self._session.request(method, url, *args, **kwargs)
 
 
-def create_tcp_connector(*args, **kwargs):
+def create_tcp_connector(*args, **kwargs) -> aiohttp.TCPConnector:
+    """
+    Creates TCP connector with reasonable 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)
-    return aiohttp.TCPConnector(*args, **kwargs)
+    # due to https://github.com/python/mypy/issues/4001
+    return aiohttp.TCPConnector(*args, **kwargs)  # type: ignore
 
 
-def create_client_session(*args, **kwargs):
+def create_client_session(*args, **kwargs) -> aiohttp.ClientSession:
+    """
+    Creates client session with reasonable defaults.
+    For details about available parameters refer to
+    `aiohttp.ClientSession <https://docs.aiohttp.org/en/stable/client_reference.html>`_
+
+    Exemplary 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)
-    return aiohttp.ClientSession(*args, **kwargs)
+    # 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:
@@ -56,26 +123,25 @@ def handle_exception():
         raise BackendNotAvailable()
     except aiohttp.ClientConnectionError:
         raise NetworkError()
-    except aiohttp.ContentTypeError:
+    except aiohttp.ContentTypeError as error:
-        raise UnknownBackendResponse()
+        raise UnknownBackendResponse(error.message)
     except aiohttp.ClientResponseError as error:
         if error.status == HTTPStatus.UNAUTHORIZED:
-            raise AuthenticationRequired()
+            raise AuthenticationRequired(error.message)
         if error.status == HTTPStatus.FORBIDDEN:
-            raise AccessDenied()
+            raise AccessDenied(error.message)
         if error.status == HTTPStatus.SERVICE_UNAVAILABLE:
-            raise BackendNotAvailable()
+            raise BackendNotAvailable(error.message)
         if error.status == HTTPStatus.TOO_MANY_REQUESTS:
-            raise TooManyRequests()
+            raise TooManyRequests(error.message)
         if error.status >= 500:
-            raise BackendError()
+            raise BackendError(error.message)
         if error.status >= 400:
-            logging.warning(
+            logger.warning(
                 "Got status %d while performing %s request for %s",
                 error.status, error.request_info.method, str(error.request_info.url)
             )
-            raise UnknownError()
+            raise UnknownError(error.message)
-    except aiohttp.ClientError:
+    except aiohttp.ClientError as e:
-        logging.exception("Caught exception while performing request")
+        logger.exception("Caught exception while performing request")
-        raise UnknownError()
+        raise UnknownError(repr(e))
-

src/galaxy/proc_tools.py

@@ -1,10 +1,6 @@
-import platform
+import sys
 from dataclasses import dataclass
-from typing import Iterable, NewType, Optional, Set
+from typing import Iterable, NewType, Optional, List, cast
-
-
-def is_windows():
-    return platform.system() == "Windows"
 
 
 ProcessId = NewType("ProcessId", int)
@@ -16,7 +12,7 @@ class ProcessInfo:
     binary_path: Optional[str]
 
 
-if is_windows():
+if sys.platform == "win32":
     from ctypes import byref, sizeof, windll, create_unicode_buffer, FormatError, WinError
     from ctypes.wintypes import DWORD
 
@@ -25,14 +21,14 @@ if is_windows():
         _PROC_ID_T = DWORD
         list_size = 4096
 
-        def try_get_pids(list_size: int) -> Set[ProcessId]:
+        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())
+                raise WinError(descr="Failed to get process ID list: %s" % FormatError())  # type: ignore
 
-            return proc_id_list[:int(result_size.value / sizeof(_PROC_ID_T()))]
+            return cast(List[ProcessId], proc_id_list[:int(result_size.value / sizeof(_PROC_ID_T()))])
 
         while True:
             proc_ids = try_get_pids(list_size)
@@ -59,7 +55,7 @@ if is_windows():
                 exe_path_buffer = create_unicode_buffer(_MAX_PATH)
                 exe_path_len = DWORD(len(exe_path_buffer))
 
-                return exe_path_buffer[:exe_path_len.value] if windll.kernel32.QueryFullProcessImageNameW(
+                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
 
@@ -86,6 +82,6 @@ else:
             return process_info
 
 
-def process_iter() -> Iterable[ProcessInfo]:
+def process_iter() -> Iterable[Optional[ProcessInfo]]:
     for pid in pids():
         yield get_process_info(pid)

src/galaxy/reader.py

@@ -12,7 +12,7 @@ class StreamLineReader:
         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)
+                chunk = await self._reader.read(1024*1024)
                 if not chunk:
                     return bytes() # EOF
                 self._buffer += chunk

src/galaxy/registry_monitor.py

@@ -0,0 +1,99 @@
+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 not None:
+            self._set_key_update_notification()
+
+        return False
+
+    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,53 @@
+import asyncio
+import logging
+from collections import OrderedDict
+from itertools import count
+
+
+logger = logging.getLogger(__name__)
+
+
+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
+                logger.debug("Task manager %s: finished task %d (%s)", self._name, task_id, description)
+                return result
+            except asyncio.CancelledError:
+                if handle_exceptions:
+                    logger.debug("Task manager %s: canceled task %d (%s)", self._name, task_id, description)
+                else:
+                    raise
+            except Exception:
+                if handle_exceptions:
+                    logger.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)
+        logger.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/unittest/mock.py

@@ -4,8 +4,8 @@ from unittest.mock import MagicMock
 
 class AsyncMock(MagicMock):
     """
-    ..deprecated:: 0.45
+    .. deprecated:: 0.45
-    Use: :class:`MagicMock` with meth:`~.async_return_value`.
+      Use: :class:`MagicMock` with meth:`~.async_return_value`.
     """
     async def __call__(self, *args, **kwargs):
         return super(AsyncMock, self).__call__(*args, **kwargs)
@@ -13,19 +13,27 @@ class AsyncMock(MagicMock):
 
 def coroutine_mock():
     """
-    ..deprecated:: 0.45
+    .. deprecated:: 0.45
-    Use: :class:`MagicMock` with meth:`~.async_return_value`.
+      Use: :class:`MagicMock` with meth:`~.async_return_value`.
     """
     coro = MagicMock(name="CoroutineResult")
     corofunc = MagicMock(name="CoroutineFunction", side_effect=asyncio.coroutine(coro))
     corofunc.coro = coro
     return corofunc
 
+
 async def skip_loop(iterations=1):
     for _ in range(iterations):
         await asyncio.sleep(0)
 
 
 async def async_return_value(return_value, loop_iterations_delay=0):
-    await skip_loop(loop_iterations_delay)
+    if loop_iterations_delay > 0:
+        await skip_loop(loop_iterations_delay)
     return return_value
+
+
+async def async_raise(error, loop_iterations_delay=0):
+    if loop_iterations_delay > 0:
+        await skip_loop(loop_iterations_delay)
+    raise error

tests/__init__.py

@@ -7,11 +7,8 @@ def create_message(request):
 
 def get_messages(write_mock):
     messages = []
-    print("call_args_list", write_mock.call_args_list)
     for call_args in write_mock.call_args_list:
-        print("call_args", call_args)
         data = call_args[0][0]
-        print("data", data)
         for line in data.splitlines():
             message = json.loads(line)
             messages.append(message)

tests/conftest.py

@@ -1,11 +1,13 @@
-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.api.plugin import Plugin
+from galaxy.unittest.mock import async_return_value
+
 
 @pytest.fixture()
 def reader():
@@ -13,23 +15,26 @@ def reader():
     stream.read = MagicMock()
     yield stream
 
+
 @pytest.fixture()
 async def writer():
     stream = MagicMock(name="stream_writer")
-    stream.write = MagicMock()
+    stream.drain.side_effect = lambda: async_return_value(None)
-    stream.drain = MagicMock()
     yield stream
 
+
 @pytest.fixture()
 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"""
     methods = (
         "handshake_complete",
@@ -37,22 +42,44 @@ def plugin(reader, writer):
         "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_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",
+        "get_local_size",
+        "prepare_local_size_context",
+        "local_size_import_complete",
+        "get_subscriptions",
+        "get_subscription_games",
+        "prepare_subscription_games_context",
+        "subscription_games_import_complete"
     )
 
     with ExitStack() as stack:
         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)

tests/test_achievements.py

@@ -5,7 +5,7 @@ from pytest import raises
 
 from galaxy.api.types import Achievement
 from galaxy.api.errors import BackendError
-from galaxy.unittest.mock import async_return_value
+from galaxy.unittest.mock import async_return_value, skip_loop
 
 from tests import create_message, get_messages
 
@@ -40,6 +40,7 @@ async def test_get_unlocked_achievements_success(plugin, read, write):
     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 get_messages(write) == [
         {
@@ -97,6 +98,7 @@ async def test_get_unlocked_achievements_error(exception, code, message, plugin,
     plugin.get_unlocked_achievements.side_effect = exception
     await plugin.run()
     plugin.get_unlocked_achievements.assert_called()
+    plugin.achievements_import_complete.asert_called_with()
 
     assert get_messages(write) == [
         {
@@ -133,7 +135,8 @@ async def test_prepare_get_unlocked_achievements_context_error(plugin, read, wri
             "game_ids": ["14"]
         }
     }
-    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+
     await plugin.run()
 
     assert get_messages(write) == [
@@ -151,6 +154,7 @@ async def test_prepare_get_unlocked_achievements_context_error(plugin, read, wri
 @pytest.mark.asyncio
 async def test_import_in_progress(plugin, read, write):
     plugin.prepare_achievements_context.return_value = async_return_value(None)
+    plugin.get_unlocked_achievements.return_value = async_return_value([])
     requests = [
         {
             "jsonrpc": "2.0",
@@ -172,32 +176,32 @@ async def test_import_in_progress(plugin, read, write):
     read.side_effect = [
         async_return_value(create_message(requests[0])),
         async_return_value(create_message(requests[1])),
-        async_return_value(b"")
+        async_return_value(b"", 10)
     ]
 
     await plugin.run()
 
-    assert get_messages(write) == [
+    messages = get_messages(write)
-        {
+    assert {
-            "jsonrpc": "2.0",
+        "jsonrpc": "2.0",
-            "id": "3",
+        "id": "3",
-            "result": None
+        "result": None
-        },
+    } in messages
-        {
+    assert {
-            "jsonrpc": "2.0",
+        "jsonrpc": "2.0",
-            "id": "4",
+        "id": "4",
-            "error": {
+        "error": {
-                "code": 600,
+            "code": 600,
-                "message": "Import already in progress"
+            "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)
     plugin.unlock_achievement("14", achievement)
+    await skip_loop()
     response = json.loads(write.call_args[0][0])
 
     assert response == {

tests/test_authenticate.py

@@ -5,7 +5,7 @@ from galaxy.api.errors import (
     UnknownError, InvalidCredentials, NetworkError, LoggedInElsewhere, ProtocolError,
     BackendNotAvailable, BackendTimeout, BackendError, TemporaryBlocked, Banned, AccessDenied
 )
-from galaxy.unittest.mock import async_return_value
+from galaxy.unittest.mock import async_return_value, skip_loop
 
 from tests import create_message, get_messages
 
@@ -17,7 +17,7 @@ async def test_success(plugin, read, write):
         "id": "3",
         "method": "init_authentication"
     }
-    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    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()
@@ -55,7 +55,7 @@ async def test_failure(plugin, read, write, error, code, message):
         "method": "init_authentication"
     }
 
-    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    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()
@@ -84,7 +84,7 @@ async def test_stored_credentials(plugin, read, write):
             }
         }
     }
-    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    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"})
@@ -97,6 +97,7 @@ async def test_store_credentials(plugin, write):
         "token": "ABC"
     }
     plugin.store_credentials(credentials)
+    await skip_loop()
 
     assert get_messages(write) == [
         {
@@ -110,6 +111,7 @@ async def test_store_credentials(plugin, write):
 @pytest.mark.asyncio
 async def test_lost_authentication(plugin, write):
     plugin.lost_authentication()
+    await skip_loop()
 
     assert get_messages(write) == [
         {

tests/test_features.py

@@ -13,7 +13,14 @@ def test_base_class():
         Feature.ImportAchievements,
         Feature.ImportGameTime,
         Feature.ImportFriends,
-        Feature.ShutdownPlatformClient
+        Feature.ShutdownPlatformClient,
+        Feature.LaunchPlatformClient,
+        Feature.ImportGameLibrarySettings,
+        Feature.ImportOSCompatibility,
+        Feature.ImportUserPresence,
+        Feature.ImportLocalSize,
+        Feature.ImportSubscriptions,
+        Feature.ImportSubscriptionGames
     }
 
 

tests/test_friends.py

@@ -1,6 +1,6 @@
-from galaxy.api.types import FriendInfo
+from galaxy.api.types import UserInfo
 from galaxy.api.errors import UnknownError
-from galaxy.unittest.mock import async_return_value
+from galaxy.unittest.mock import async_return_value, skip_loop
 
 import pytest
 
@@ -15,10 +15,12 @@ async def test_get_friends_success(plugin, read, write):
         "method": "import_friends"
     }
 
-    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    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"),
+        UserInfo("3", "Jan", "https://avatar.url/u3", None),
-        FriendInfo("5", "Ola")
+        UserInfo("5", "Ola", None, "https://profile.url/u5"),
+        UserInfo("6", "Ola2", None),
+        UserInfo("7", "Ola3"),
     ])
     await plugin.run()
     plugin.get_friends.assert_called_with()
@@ -29,8 +31,10 @@ async def test_get_friends_success(plugin, read, write):
             "id": "3",
             "result": {
                 "friend_info_list": [
-                    {"user_id": "3", "user_name": "Jan"},
+                    {"user_id": "3", "user_name": "Jan", "avatar_url": "https://avatar.url/u3"},
-                    {"user_id": "5", "user_name": "Ola"}
+                    {"user_id": "5", "user_name": "Ola", "profile_url": "https://profile.url/u5"},
+                    {"user_id": "6", "user_name": "Ola2"},
+                    {"user_id": "7", "user_name": "Ola3"},
                 ]
             }
         }
@@ -45,7 +49,7 @@ async def test_get_friends_failure(plugin, read, write):
         "method": "import_friends"
     }
 
-    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    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()
@@ -64,16 +68,22 @@ async def test_get_friends_failure(plugin, read, write):
 
 @pytest.mark.asyncio
 async def test_add_friend(plugin, write):
-    friend = FriendInfo("7", "Kuba")
+    friend = UserInfo("7", "Kuba", avatar_url="https://avatar.url/kuba.jpg", profile_url="https://profile.url/kuba")
 
     plugin.add_friend(friend)
+    await skip_loop()
 
     assert get_messages(write) == [
         {
             "jsonrpc": "2.0",
             "method": "friend_added",
             "params": {
-                "friend_info": {"user_id": "7", "user_name": "Kuba"}
+                "friend_info": {
+                    "user_id": "7",
+                    "user_name": "Kuba",
+                    "avatar_url": "https://avatar.url/kuba.jpg",
+                    "profile_url": "https://profile.url/kuba"
+                }
             }
         }
     ]
@@ -82,6 +92,7 @@ async def test_add_friend(plugin, write):
 @pytest.mark.asyncio
 async def test_remove_friend(plugin, write):
     plugin.remove_friend("5")
+    await skip_loop()
 
     assert get_messages(write) == [
         {
@@ -92,3 +103,26 @@ async def test_remove_friend(plugin, write):
             }
         }
     ]
+
+
+@pytest.mark.asyncio
+async def test_update_friend_info(plugin, write):
+    plugin.update_friend_info(
+        UserInfo("7", "Jakub", avatar_url="https://new-avatar.url/kuba2.jpg", profile_url="https://profile.url/kuba")
+    )
+    await skip_loop()
+
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "method": "friend_updated",
+            "params": {
+                "friend_info": {
+                    "user_id": "7",
+                    "user_name": "Jakub",
+                    "avatar_url": "https://new-avatar.url/kuba2.jpg",
+                    "profile_url": "https://profile.url/kuba"
+                }
+            }
+        }
+    ]

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

@@ -3,7 +3,7 @@ from unittest.mock import call
 import pytest
 from galaxy.api.types import GameTime
 from galaxy.api.errors import BackendError
-from galaxy.unittest.mock import async_return_value
+from galaxy.unittest.mock import async_return_value, skip_loop
 
 from tests import create_message, get_messages
 
@@ -31,6 +31,7 @@ async def test_get_game_time_success(plugin, read, write):
         call("5", "abc"),
         call("7", "abc"),
     ])
+    plugin.game_times_import_complete.assert_called_once_with()
 
     assert get_messages(write) == [
         {
@@ -96,6 +97,7 @@ async def test_get_game_time_error(exception, code, message, plugin, read, write
     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) == [
         {
@@ -133,7 +135,7 @@ async def test_prepare_get_game_time_context_error(plugin, read, write):
             "game_ids": ["6"]
         }
     }
-    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
     await plugin.run()
 
     assert get_messages(write) == [
@@ -172,32 +174,32 @@ async def test_import_in_progress(plugin, read, write):
     read.side_effect = [
         async_return_value(create_message(requests[0])),
         async_return_value(create_message(requests[1])),
-        async_return_value(b"")
+        async_return_value(b"", 10)
     ]
 
     await plugin.run()
 
-    assert get_messages(write) == [
+    messages = get_messages(write)
-        {
+    assert {
-            "jsonrpc": "2.0",
+        "jsonrpc": "2.0",
-            "id": "3",
+        "id": "3",
-            "result": None
+        "result": None
-        },
+    } in messages
-        {
+    assert {
-            "jsonrpc": "2.0",
+        "jsonrpc": "2.0",
-            "id": "4",
+        "id": "4",
-            "error": {
+        "error": {
-                "code": 600,
+            "code": 600,
-                "message": "Import already in progress"
+            "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) == [
         {

tests/test_http.py

@@ -3,6 +3,8 @@ 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,
@@ -10,7 +12,7 @@ from galaxy.api.errors import (
 )
 from galaxy.http import handle_exception
 
-request_info = aiohttp.RequestInfo("http://o.pl", "GET", {})
+request_info = aiohttp.RequestInfo(URL("http://o.pl"), "GET", CIMultiDictProxy(CIMultiDict()))
 
 @pytest.mark.parametrize(
     "aiohttp_exception,expected_exception_type",
@@ -18,15 +20,15 @@ request_info = aiohttp.RequestInfo("http://o.pl", "GET", {})
         (asyncio.TimeoutError(), BackendTimeout),
         (aiohttp.ServerDisconnectedError(), BackendNotAvailable),
         (aiohttp.ClientConnectionError(), NetworkError),
-        (aiohttp.ContentTypeError(request_info, []), UnknownBackendResponse),
+        (aiohttp.ContentTypeError(request_info, ()), UnknownBackendResponse),
-        (aiohttp.ClientResponseError(request_info, [], status=HTTPStatus.UNAUTHORIZED), AuthenticationRequired),
+        (aiohttp.ClientResponseError(request_info, (), status=HTTPStatus.UNAUTHORIZED), AuthenticationRequired),
-        (aiohttp.ClientResponseError(request_info, [], status=HTTPStatus.FORBIDDEN), AccessDenied),
+        (aiohttp.ClientResponseError(request_info, (), status=HTTPStatus.FORBIDDEN), AccessDenied),
-        (aiohttp.ClientResponseError(request_info, [], status=HTTPStatus.SERVICE_UNAVAILABLE), BackendNotAvailable),
+        (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.TOO_MANY_REQUESTS), TooManyRequests),
-        (aiohttp.ClientResponseError(request_info, [], status=HTTPStatus.INTERNAL_SERVER_ERROR), BackendError),
+        (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.NOT_IMPLEMENTED), BackendError),
-        (aiohttp.ClientResponseError(request_info, [], status=HTTPStatus.BAD_REQUEST), UnknownError),
+        (aiohttp.ClientResponseError(request_info, (), status=HTTPStatus.BAD_REQUEST), UnknownError),
-        (aiohttp.ClientResponseError(request_info, [], status=HTTPStatus.NOT_FOUND), UnknownError),
+        (aiohttp.ClientResponseError(request_info, (), status=HTTPStatus.NOT_FOUND), UnknownError),
         (aiohttp.ClientError(), UnknownError)
     ]
 )

tests/test_internal.py

@@ -46,6 +46,7 @@ async def test_shutdown(plugin, read, write):
     }
     read.side_effect = [async_return_value(create_message(request))]
     await plugin.run()
+    await plugin.wait_closed()
     plugin.shutdown.assert_called_with()
     assert get_messages(write) == [
         {

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

@@ -3,7 +3,7 @@ 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
+from galaxy.unittest.mock import async_return_value, skip_loop
 
 from tests import create_message, get_messages
 
@@ -15,7 +15,7 @@ async def test_success(plugin, read, write):
         "id": "3",
         "method": "import_local_games"
     }
-    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
 
     plugin.get_local_games.return_value = async_return_value([
         LocalGame("1", LocalGameState.Running),
@@ -63,7 +63,7 @@ async def test_failure(plugin, read, write, error, code, message):
         "id": "3",
         "method": "import_local_games"
     }
-    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    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()
@@ -83,6 +83,7 @@ async def test_failure(plugin, read, write, error, code, message):
 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) == [
         {

tests/test_local_size.py

@@ -0,0 +1,191 @@
+from unittest.mock import call
+
+import pytest
+from galaxy.api.errors import FailedParsingManifest
+from galaxy.unittest.mock import async_return_value
+
+from tests import create_message, get_messages
+
+
+@pytest.mark.asyncio
+async def test_get_local_size_success(plugin, read, write):
+    context = {'abc': 'def'}
+    plugin.prepare_local_size_context.return_value = async_return_value(context)
+    request = {
+        "jsonrpc": "2.0",
+        "id": "11",
+        "method": "start_local_size_import",
+        "params": {"game_ids": ["777", "13", "42"]}
+    }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+    plugin.get_local_size.side_effect = [
+        async_return_value(100000000000, 1),
+        async_return_value(None),
+        async_return_value(3333333)
+    ]
+    await plugin.run()
+    plugin.get_local_size.assert_has_calls([
+        call("777", context),
+        call("13", context),
+        call("42", context)
+    ])
+    plugin.local_size_import_complete.assert_called_once_with()
+
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "11",
+            "result": None
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "local_size_import_success",
+            "params": {
+                "game_id": "777",
+                "local_size": 100000000000
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "local_size_import_success",
+            "params": {
+                "game_id": "13",
+                "local_size": None
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "local_size_import_success",
+            "params": {
+                "game_id": "42",
+                "local_size": 3333333
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "local_size_import_finished",
+            "params": None
+        }
+    ]
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("exception,code,message", [
+    (FailedParsingManifest, 200, "Failed parsing manifest"),
+    (KeyError, 0, "Unknown error")
+])
+async def test_get_local_size_error(exception, code, message, plugin, read, write):
+    game_id = "6"
+    request_id = "55"
+    plugin.prepare_local_size_context.return_value = async_return_value(None)
+    request = {
+        "jsonrpc": "2.0",
+        "id": request_id,
+        "method": "start_local_size_import",
+        "params": {"game_ids": [game_id]}
+    }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+    plugin.get_local_size.side_effect = exception
+    await plugin.run()
+    plugin.get_local_size.assert_called()
+    plugin.local_size_import_complete.assert_called_once_with()
+
+    direct_response = {
+        "jsonrpc": "2.0",
+        "id": request_id,
+        "result": None
+    }
+    responses = get_messages(write)
+    assert direct_response in responses
+    responses.remove(direct_response)
+    assert responses == [
+        {
+            "jsonrpc": "2.0",
+            "method": "local_size_import_failure",
+            "params": {
+                "game_id": game_id,
+                "error": {
+                    "code": code,
+                    "message": message
+                }
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "local_size_import_finished",
+            "params": None
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_prepare_get_local_size_context_error(plugin, read, write):
+    request_id = "31415"
+    error_details = "Unexpected syntax"
+    error_message, error_code = FailedParsingManifest().message, FailedParsingManifest().code
+    plugin.prepare_local_size_context.side_effect = FailedParsingManifest(error_details)
+    request = {
+        "jsonrpc": "2.0",
+        "id": request_id,
+        "method": "start_local_size_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": error_code,
+                "message": error_message,
+                "data": error_details
+            }
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_import_already_in_progress_error(plugin, read, write):
+    plugin.prepare_local_size_context.return_value = async_return_value(None)
+    plugin.get_local_size.return_value = async_return_value(100, 5)
+    requests = [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "method": "start_local_size_import",
+            "params": {
+                "game_ids": ["42"]
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "id": "4",
+            "method": "start_local_size_import",
+            "params": {
+                "game_ids": ["13"]
+            }
+        }
+    ]
+    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_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

@@ -3,7 +3,7 @@ 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
+from galaxy.unittest.mock import async_return_value, skip_loop
 
 from tests import create_message, get_messages
 
@@ -15,7 +15,7 @@ async def test_success(plugin, read, write):
         "id": "3",
         "method": "import_owned_games"
     }
-    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
 
     plugin.get_owned_games.return_value = async_return_value([
         Game("3", "Doom", None, LicenseInfo(LicenseType.SinglePurchase, None)),
@@ -80,7 +80,7 @@ async def test_failure(plugin, read, write):
         "method": "import_owned_games"
     }
 
-    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
+    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()
@@ -100,6 +100,7 @@ async def test_failure(plugin, read, write):
 async def test_add_game(plugin, write):
     game = Game("3", "Doom", None, LicenseInfo(LicenseType.SinglePurchase, None))
     plugin.add_game(game)
+    await skip_loop()
     assert get_messages(write) == [
         {
             "jsonrpc": "2.0",
@@ -120,6 +121,7 @@ async def test_add_game(plugin, write):
 @pytest.mark.asyncio
 async def test_remove_game(plugin, write):
     plugin.remove_game("5")
+    await skip_loop()
     assert get_messages(write) == [
         {
             "jsonrpc": "2.0",
@@ -135,6 +137,7 @@ async def test_remove_game(plugin, write):
 async def test_update_game(plugin, write):
     game = Game("3", "Doom", None, LicenseInfo(LicenseType.SinglePurchase, None))
     plugin.update_game(game)
+    await skip_loop()
     assert get_messages(write) == [
         {
             "jsonrpc": "2.0",

tests/test_persistent_cache.py

@@ -1,6 +1,6 @@
 import pytest
 
-from galaxy.unittest.mock import async_return_value
+from galaxy.unittest.mock import async_return_value, skip_loop
 
 from tests import create_message, get_messages
 
@@ -57,6 +57,7 @@ async def test_set_cache(plugin, write, cache_data):
 
     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
@@ -68,6 +69,7 @@ async def test_clear_cache(plugin, write, cache_data):
 
     plugin.persistent_cache.clear()
     plugin.push_cache()
+    await skip_loop()
 
     assert_rpc_request(write, "push_cache", {})
     assert {} == plugin.persistent_cache

tests/test_refresh_credentials.py

@@ -0,0 +1,72 @@
+import pytest
+import asyncio
+
+from galaxy.unittest.mock import async_return_value
+from tests import create_message, get_messages
+from galaxy.api.errors import (
+    BackendNotAvailable, BackendTimeout, BackendError, InvalidCredentials, NetworkError, AccessDenied, UnknownError
+)
+from galaxy.api.jsonrpc import JsonRpcError
+@pytest.mark.asyncio
+async def test_refresh_credentials_success(plugin, read, write):
+
+    run_task = asyncio.create_task(plugin.run())
+
+    refreshed_credentials = {
+        "access_token": "new_access_token"
+    }
+
+    response = {
+        "jsonrpc": "2.0",
+        "id": "1",
+        "result": refreshed_credentials
+    }
+    # 2 loop iterations delay is to force sending response after request has been sent
+    read.side_effect = [async_return_value(create_message(response), loop_iterations_delay=2)]
+
+    result = await plugin.refresh_credentials({}, False)
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "method": "refresh_credentials",
+            "params": {
+            },
+            "id": "1"
+        }
+    ]
+
+    assert result == refreshed_credentials
+    await run_task
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("exception", [
+    BackendNotAvailable, BackendTimeout, BackendError, InvalidCredentials, NetworkError, AccessDenied, UnknownError
+])
+async def test_refresh_credentials_failure(exception, plugin, read, write):
+
+    run_task = asyncio.create_task(plugin.run())
+    error = exception()
+    response = {
+        "jsonrpc": "2.0",
+        "id": "1",
+        "error": error.json()
+    }
+
+    # 2 loop iterations delay is to force sending response after request has been sent
+    read.side_effect = [async_return_value(create_message(response), loop_iterations_delay=2)]
+
+    with pytest.raises(JsonRpcError) as e:
+        await plugin.refresh_credentials({}, False)
+
+    assert error == e.value
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "method": "refresh_credentials",
+            "params": {
+            },
+            "id": "1"
+        }
+    ]
+
+    await run_task

tests/test_subscriptions.py

@@ -0,0 +1,340 @@
+import pytest
+
+from galaxy.api.types import Subscription, SubscriptionGame
+from galaxy.api.consts import SubscriptionDiscovery
+from galaxy.api.errors import FailedParsingManifest, BackendError, UnknownError
+from galaxy.unittest.mock import async_return_value
+
+from tests import create_message, get_messages
+
+@pytest.mark.asyncio
+async def test_get_subscriptions_success(plugin, read, write):
+    request = {
+        "jsonrpc": "2.0",
+        "id": "3",
+        "method": "import_subscriptions"
+    }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+
+    plugin.get_subscriptions.return_value = async_return_value([
+        Subscription("1"),
+        Subscription("2", False, subscription_discovery=SubscriptionDiscovery.AUTOMATIC),
+        Subscription("3", True, 1580899100, SubscriptionDiscovery.USER_ENABLED)
+    ])
+    await plugin.run()
+    plugin.get_subscriptions.assert_called_with()
+
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "result": {
+                "subscriptions": [
+                    {
+                        "subscription_name": "1",
+                        'subscription_discovery': 3
+                    },
+                    {
+                        "subscription_name": "2",
+                        "owned": False,
+                        'subscription_discovery': 1
+                    },
+                    {
+                        "subscription_name": "3",
+                        "owned": True,
+                        "end_time": 1580899100,
+                        'subscription_discovery': 2
+                    }
+                ]
+            }
+        }
+    ]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "error,code,message",
+    [
+        pytest.param(UnknownError, 0, "Unknown error", id="unknown_error"),
+        pytest.param(FailedParsingManifest, 200, "Failed parsing manifest", id="failed_parsing")
+    ],
+)
+async def test_get_subscriptions_failure_generic(plugin, read, write, error, code, message):
+    request = {
+        "jsonrpc": "2.0",
+        "id": "3",
+        "method": "import_subscriptions"
+    }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+    plugin.get_subscriptions.side_effect = error()
+    await plugin.run()
+    plugin.get_subscriptions.assert_called_with()
+
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "error": {
+                "code": code,
+                "message": message
+            }
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_get_subscription_games_success(plugin, read, write):
+    plugin.prepare_subscription_games_context.return_value = async_return_value(5)
+    request = {
+        "jsonrpc": "2.0",
+        "id": "3",
+        "method": "start_subscription_games_import",
+        "params": {
+            "subscription_names": ["sub_a"]
+        }
+    }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+
+    async def sub_games():
+        games = [
+        SubscriptionGame(game_title="game A", game_id="game_A"),
+        SubscriptionGame(game_title="game B", game_id="game_B", start_time=1548495632),
+        SubscriptionGame(game_title="game C", game_id="game_C", end_time=1548495633),
+        SubscriptionGame(game_title="game D", game_id="game_D", start_time=1548495632, end_time=1548495633),
+       ]
+        yield [game for game in games]
+
+    plugin.get_subscription_games.return_value = sub_games()
+    await plugin.run()
+    plugin.prepare_subscription_games_context.assert_called_with(["sub_a"])
+    plugin.get_subscription_games.assert_called_with("sub_a", 5)
+    plugin.subscription_games_import_complete.asert_called_with()
+
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "result": None
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "subscription_games_import_success",
+            "params": {
+                "subscription_name": "sub_a",
+                "subscription_games": [
+                    {
+                        "game_title": "game A",
+                        "game_id": "game_A"
+                    },
+                    {
+                        "game_title": "game B",
+                        "game_id": "game_B",
+                        "start_time": 1548495632
+                    },
+                    {
+                        "game_title": "game C",
+                        "game_id": "game_C",
+                        "end_time": 1548495633
+                    },
+                    {
+                        "game_title": "game D",
+                        "game_id": "game_D",
+                        "start_time": 1548495632,
+                        "end_time": 1548495633
+                    }
+                ]
+            }
+        },
+        {
+            'jsonrpc': '2.0',
+            'method':
+                'subscription_games_partial_import_finished',
+                'params': {
+                    "subscription_name": "sub_a"
+                }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "subscription_games_import_finished",
+            "params": None
+        }
+    ]
+
+@pytest.mark.asyncio
+async def test_get_subscription_games_success_empty(plugin, read, write):
+    plugin.prepare_subscription_games_context.return_value = async_return_value(5)
+    request = {
+        "jsonrpc": "2.0",
+        "id": "3",
+        "method": "start_subscription_games_import",
+        "params": {
+            "subscription_names": ["sub_a"]
+        }
+    }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+
+    async def sub_games():
+        yield None
+
+    plugin.get_subscription_games.return_value = sub_games()
+    await plugin.run()
+    plugin.prepare_subscription_games_context.assert_called_with(["sub_a"])
+    plugin.get_subscription_games.assert_called_with("sub_a", 5)
+    plugin.subscription_games_import_complete.asert_called_with()
+
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "result": None
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "subscription_games_import_success",
+            "params": {
+                "subscription_name": "sub_a",
+                "subscription_games": None
+            }
+        },
+        {
+            'jsonrpc': '2.0',
+            'method':
+                'subscription_games_partial_import_finished',
+                'params': {
+                    "subscription_name": "sub_a"
+                }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "subscription_games_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_subscription_games_error(exception, code, message, plugin, read, write):
+    plugin.prepare_subscription_games_context.return_value = async_return_value(None)
+    request = {
+        "jsonrpc": "2.0",
+        "id": "3",
+        "method": "start_subscription_games_import",
+        "params": {
+            "subscription_names": ["sub_a"]
+        }
+    }
+
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+    plugin.get_subscription_games.side_effect = exception
+    await plugin.run()
+    plugin.get_subscription_games.assert_called()
+    plugin.subscription_games_import_complete.asert_called_with()
+
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "result": None
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "subscription_games_import_failure",
+            "params": {
+                "subscription_name": "sub_a",
+                "error": {
+                    "code": code,
+                    "message": message
+                }
+            }
+        },
+        {
+            'jsonrpc': '2.0',
+            'method':
+                'subscription_games_partial_import_finished',
+                'params': {
+                    "subscription_name": "sub_a"
+                }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "subscription_games_import_finished",
+            "params": None
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_prepare_get_subscription_games_context_error(plugin, read, write):
+    request_id = "31415"
+    error_details = "Unexpected backend error"
+    error_message, error_code = BackendError().message, BackendError().code
+    plugin.prepare_subscription_games_context.side_effect = BackendError(error_details)
+    request = {
+        "jsonrpc": "2.0",
+        "id": request_id,
+        "method": "start_subscription_games_import",
+        "params": {"subscription_names": ["sub_a", "sub_b"]}
+    }
+    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": error_code,
+                "message": error_message,
+                "data": error_details
+            }
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_import_already_in_progress_error(plugin, read, write):
+    plugin.prepare_subscription_games_context.return_value = async_return_value(None)
+    requests = [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "method": "start_subscription_games_import",
+            "params": {
+                "subscription_names": ["sub_a"]
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "id": "4",
+            "method": "start_subscription_games_import",
+            "params": {
+                "subscription_names": ["sub_a","sub_b"]
+            }
+        }
+    ]
+    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_user_presence.py

@@ -0,0 +1,276 @@
+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, skip_loop
+from tests import create_message, get_messages
+
+
+@pytest.mark.asyncio
+async def test_get_user_presence_success(plugin, read, write):
+    context = "abc"
+    user_id_list = ["666", "13", "42", "69", "22"]
+    plugin.prepare_user_presence_context.return_value = async_return_value(context)
+    request = {
+        "jsonrpc": "2.0",
+        "id": "11",
+        "method": "start_user_presence_import",
+        "params": {"user_id_list": user_id_list}
+    }
+    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",
+            None
+        )),
+        async_return_value(UserPresence(
+            PresenceState.Offline,
+            None,
+            None,
+            "Going to grandma's house",
+            None
+        )),
+        async_return_value(UserPresence(
+            PresenceState.Online,
+            "game-id3",
+            "game-title3",
+            "Pew pew",
+            None
+        )),
+        async_return_value(UserPresence(
+            PresenceState.Away,
+            None,
+            "game-title4",
+            "AFKKTHXBY",
+            None
+        )),
+        async_return_value(UserPresence(
+            PresenceState.Away,
+            None,
+            "game-title5",
+            None,
+            "Playing game-title5: In Menu"
+        )),
+    ]
+    await plugin.run()
+    plugin.get_user_presence.assert_has_calls([
+        call(user_id, context) for user_id in user_id_list
+    ])
+    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",
+                    "in_game_status": "unknown state"
+                }
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "user_presence_import_success",
+            "params": {
+                "user_id": "13",
+                "presence": {
+                    "presence_state": PresenceState.Offline.value,
+                    "in_game_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",
+                    "in_game_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",
+                    "in_game_status": "AFKKTHXBY"
+                }
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "user_presence_import_success",
+            "params": {
+                "user_id": "22",
+                "presence": {
+                    "presence_state": PresenceState.Away.value,
+                    "game_title": "game-title5",
+                    "full_status": "Playing game-title5: In Menu"
+                }
+            }
+        },
+        {
+            "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_id_list": [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_id_list": ["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_id_list": ["42"]
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "id": "4",
+            "method": "start_user_presence_import",
+            "params": {
+                "user_id_list": ["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
+
+
+@pytest.mark.asyncio
+async def test_update_user_presence(plugin, write):
+    plugin.update_user_presence("42", UserPresence(PresenceState.Online, "game-id", "game-title", "Pew pew"))
+    await skip_loop()
+
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "method": "user_presence_updated",
+            "params": {
+                "user_id": "42",
+                "presence": {
+                    "presence_state": PresenceState.Online.value,
+                    "game_id": "game-id",
+                    "game_title": "game-title",
+                    "in_game_status": "Pew pew"
+                }
+            }
+        }
+    ]