adhere to comments

.gitignore

@@ -8,3 +8,4 @@ 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

setup.py

@@ -2,14 +2,15 @@ from setuptools import setup, find_packages
 
 setup(
     name="galaxy.plugin.api",
-    version="0.60",
+    version="0.64",
     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",
-        "certifi==2019.3.9"
+        "aiohttp>=3.5.4",
+        "certifi>=2019.3.9",
+        "psutil>=5.6.3; sys_platform == 'darwin'"
     ]
 )

src/galaxy/api/consts.py

@@ -114,6 +114,9 @@ class Feature(Enum):
     ImportGameLibrarySettings = "ImportGameLibrarySettings"
     ImportOSCompatibility = "ImportOSCompatibility"
     ImportUserPresence = "ImportUserPresence"
+    ImportLocalSize = "ImportLocalSize"
+    ImportSubscriptions = "ImportSubscriptions"
+    ImportSubscriptionGames = "ImportSubscriptionGames"
 
 
 class LicenseType(Enum):

src/galaxy/api/errors.py

@@ -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,89 @@
+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_)

src/galaxy/api/jsonrpc.py

@@ -299,11 +299,14 @@ class Connection():
         except TypeError:
             raise InvalidRequest()
 
-    def _send(self, data):
+    def _send(self, data, sensitive=True):
         try:
             line = self._encoder.encode(data)
             data = (line + "\n").encode("utf-8")
-            logger.debug("Sending %d byte of data", len(data))
+            if sensitive:
+                logger.debug("Sending %d bytes of data", len(data))
+            else:
+                logging.debug("Sending data: %s", line)
             self._writer.write(data)
         except TypeError as error:
             logger.error(str(error))
@@ -314,7 +317,7 @@ class Connection():
             "id": request_id,
             "result": result
         }
-        self._send(response)
+        self._send(response, sensitive=False)
 
     def _send_error(self, request_id, error):
         response = {
@@ -323,7 +326,7 @@ class Connection():
             "error": error.json()
         }
 
-        self._send(response)
+        self._send(response, sensitive=False)
 
     def _send_request(self, request_id, method, params):
         request = {
@@ -332,7 +335,7 @@ class Connection():
             "id": request_id,
             "params": params
         }
-        self._send(request)
+        self._send(request, sensitive=True)
 
     def _send_notification(self, method, params):
         notification = {
@@ -340,7 +343,7 @@ class Connection():
             "method": method,
             "params": params
         }
-        self._send(notification)
+        self._send(notification, sensitive=True)
 
     @staticmethod
     def _log_request(request, sensitive_params):

src/galaxy/api/plugin.py

@@ -4,15 +4,16 @@ import json
 import logging
 import sys
 from enum import Enum
-from typing import Any, Dict, List, Optional, Set, Union
+from typing import Any, Dict, List, Optional, Set, Union, AsyncGenerator
 
 from galaxy.api.consts import Feature, OSCompatibility
-from galaxy.api.errors import ImportInProgress, UnknownError
 from galaxy.api.jsonrpc import ApplicationError, Connection
 from galaxy.api.types import (
-    Achievement, Authentication, Game, GameLibrarySettings, GameTime, LocalGame, NextStep, UserInfo, UserPresence
+    Achievement, Authentication, Game, GameLibrarySettings, GameTime, LocalGame, NextStep, UserInfo, UserPresence,
+    Subscription, SubscriptionGame
 )
 from galaxy.task_manager import TaskManager
+from galaxy.api.importer import Importer, CollectionImporter
 
 
 logger = logging.getLogger(__name__)
@@ -48,17 +49,83 @@ class Plugin:
         encoder = JSONEncoder()
         self._connection = Connection(self._reader, self._writer, encoder)
 
-        self._achievements_import_in_progress = False
-        self._game_times_import_in_progress = False
-        self._game_library_settings_import_in_progress = False
-        self._os_compatibility_import_in_progress = False
-        self._user_presence_import_in_progress = False
-
         self._persistent_cache = dict()
 
         self._internal_task_manager = TaskManager("plugin internal")
         self._external_task_manager = TaskManager("plugin external")
 
+        self._achievements_importer = Importer(
+            self._external_task_manager,
+            "achievements",
+            self.get_unlocked_achievements,
+            self.prepare_achievements_context,
+            self._game_achievements_import_success,
+            self._game_achievements_import_failure,
+            self._achievements_import_finished,
+            self.achievements_import_complete
+        )
+        self._game_time_importer = Importer(
+            self._external_task_manager,
+            "game times",
+            self.get_game_time,
+            self.prepare_game_times_context,
+            self._game_time_import_success,
+            self._game_time_import_failure,
+            self._game_times_import_finished,
+            self.game_times_import_complete
+        )
+        self._game_library_settings_importer = Importer(
+            self._external_task_manager,
+            "game library settings",
+            self.get_game_library_settings,
+            self.prepare_game_library_settings_context,
+            self._game_library_settings_import_success,
+            self._game_library_settings_import_failure,
+            self._game_library_settings_import_finished,
+            self.game_library_settings_import_complete
+        )
+        self._os_compatibility_importer = Importer(
+            self._external_task_manager,
+            "os compatibility",
+            self.get_os_compatibility,
+            self.prepare_os_compatibility_context,
+            self._os_compatibility_import_success,
+            self._os_compatibility_import_failure,
+            self._os_compatibility_import_finished,
+            self.os_compatibility_import_complete
+        )
+        self._user_presence_importer = Importer(
+            self._external_task_manager,
+            "users presence",
+            self.get_user_presence,
+            self.prepare_user_presence_context,
+            self._user_presence_import_success,
+            self._user_presence_import_failure,
+            self._user_presence_import_finished,
+            self.user_presence_import_complete
+        )
+        self._local_size_importer = Importer(
+            self._external_task_manager,
+            "local size",
+            self.get_local_size,
+            self.prepare_local_size_context,
+            self._local_size_import_success,
+            self._local_size_import_failure,
+            self._local_size_import_finished,
+            self.local_size_import_complete
+        )
+        self._subscription_games_importer = CollectionImporter(
+            self._subscriptions_games_partial_import_finished,
+            self._external_task_manager,
+            "subscription games",
+            self.get_subscription_games,
+            self.prepare_subscription_games_context,
+            self._subscription_games_import_success,
+            self._subscription_games_import_failure,
+            self._subscription_games_import_finished,
+            self.subscription_games_import_complete
+        )
+
         # internal
         self._register_method("shutdown", self._shutdown, internal=True)
         self._register_method("get_capabilities", self._get_capabilities, internal=True, immediate=True)
@@ -125,6 +192,15 @@ class Plugin:
         self._register_method("start_user_presence_import", self._start_user_presence_import)
         self._detect_feature(Feature.ImportUserPresence, ["get_user_presence"])
 
+        self._register_method("start_local_size_import", self._start_local_size_import)
+        self._detect_feature(Feature.ImportLocalSize, ["get_local_size"])
+
+        self._register_method("import_subscriptions", self.get_subscriptions, result_name="subscriptions")
+        self._detect_feature(Feature.ImportSubscriptions, ["get_subscriptions"])
+
+        self._register_method("start_subscription_games_import", self._start_subscription_games_import)
+        self._detect_feature(Feature.ImportSubscriptionGames, ["get_subscription_games"])
+
     async def __aenter__(self):
         return self
 
@@ -152,7 +228,8 @@ class Plugin:
         if self._implements(methods):
             self._features.add(feature)
 
-    def _register_method(self, name, handler, result_name=None, internal=False, immediate=False, sensitive_params=False):
+    def _register_method(self, name, handler, result_name=None, internal=False, immediate=False,
+                         sensitive_params=False):
         def wrap_result(result):
             if result_name:
                 result = {
@@ -435,7 +512,7 @@ class Plugin:
             }
         )
 
-    def _game_time_import_success(self, game_time: GameTime) -> None:
+    def _game_time_import_success(self, game_id: str, game_time: GameTime) -> None:
         params = {"game_time": game_time}
         self._connection.send_notification("game_time_import_success", params)
 
@@ -449,7 +526,7 @@ class Plugin:
     def _game_times_import_finished(self) -> None:
         self._connection.send_notification("game_times_import_finished", None)
 
-    def _game_library_settings_import_success(self, game_library_settings: GameLibrarySettings) -> None:
+    def _game_library_settings_import_success(self, game_id: str, game_library_settings: GameLibrarySettings) -> None:
         params = {"game_library_settings": game_library_settings}
         self._connection.send_notification("game_library_settings_import_success", params)
 
@@ -505,6 +582,57 @@ class Plugin:
     def _user_presence_import_finished(self) -> None:
         self._connection.send_notification("user_presence_import_finished", None)
 
+    def _local_size_import_success(self, game_id: str, size: Optional[int]) -> None:
+        self._connection.send_notification(
+            "local_size_import_success",
+            {
+                "game_id": game_id,
+                "local_size": size
+            }
+        )
+
+    def _local_size_import_failure(self, game_id: str, error: ApplicationError) -> None:
+        self._connection.send_notification(
+            "local_size_import_failure",
+            {
+                "game_id": game_id,
+                "error": error.json()
+            }
+        )
+
+    def _local_size_import_finished(self) -> None:
+        self._connection.send_notification("local_size_import_finished", None)
+
+    def _subscription_games_import_success(self, subscription_name: str,
+                                           subscription_games: Optional[List[SubscriptionGame]]) -> None:
+        self._connection.send_notification(
+            "subscription_games_import_success",
+            {
+                "subscription_name": subscription_name,
+                "subscription_games": subscription_games
+            }
+        )
+
+    def _subscription_games_import_failure(self, subscription_name: str, error: ApplicationError) -> None:
+        self._connection.send_notification(
+            "subscription_games_import_failure",
+            {
+                "subscription_name": subscription_name,
+                "error": error.json()
+            }
+        )
+
+    def _subscriptions_games_partial_import_finished(self, subscription_name: str) -> None:
+        self._connection.send_notification(
+            "subscription_games_partial_import_finished",
+            {
+               "subscription_name": subscription_name
+            }
+        )
+
+    def _subscription_games_import_finished(self) -> None:
+        self._connection.send_notification("subscription_games_import_finished", None)
+
     def lost_authentication(self) -> None:
         """Notify the client that integration has lost authentication for the
          current user and is unable to perform actions which would require it.
@@ -564,7 +692,7 @@ class Plugin:
         This method is called by the GOG Galaxy Client.
 
         :param stored_credentials: If the client received any credentials to store locally
-         in the previous session they will be passed here as a parameter.
+            in the previous session they will be passed here as a parameter.
 
 
         Example of possible override of the method:
@@ -586,7 +714,7 @@ class Plugin:
         raise NotImplementedError()
 
     async def pass_login_credentials(self, step: str, credentials: Dict[str, str], cookies: List[Dict[str, str]]) \
-        -> Union[NextStep, Authentication]:
+            -> Union[NextStep, Authentication]:
         """This method is called if we return :class:`~galaxy.api.types.NextStep` from :meth:`.authenticate`
         or :meth:`.pass_login_credentials`.
         This method's parameters provide the data extracted from the web page navigation that previous NextStep finished on.
@@ -636,36 +764,7 @@ class Plugin:
         raise NotImplementedError()
 
     async def _start_achievements_import(self, game_ids: List[str]) -> None:
-        if self._achievements_import_in_progress:
-            raise ImportInProgress()
-
-        context = await self.prepare_achievements_context(game_ids)
-
-        async def import_game_achievements(game_id, context_):
-            try:
-                achievements = await self.get_unlocked_achievements(game_id, context_)
-                self._game_achievements_import_success(game_id, achievements)
-            except ApplicationError as error:
-                self._game_achievements_import_failure(game_id, error)
-            except Exception:
-                logger.exception("Unexpected exception raised in import_game_achievements")
-                self._game_achievements_import_failure(game_id, UnknownError())
-
-        async def import_games_achievements(game_ids_, context_):
-            try:
-                imports = [import_game_achievements(game_id, context_) for game_id in game_ids_]
-                await asyncio.gather(*imports)
-            finally:
-                self._achievements_import_finished()
-                self._achievements_import_in_progress = False
-                self.achievements_import_complete()
-
-        self._external_task_manager.create_task(
-            import_games_achievements(game_ids, context),
-            "unlocked achievements import",
-            handle_exceptions=False
-        )
-        self._achievements_import_in_progress = True
+        await self._achievements_importer.start(game_ids)
 
     async def prepare_achievements_context(self, game_ids: List[str]) -> Any:
         """Override this method to prepare context for get_unlocked_achievements.
@@ -800,36 +899,7 @@ class Plugin:
         raise NotImplementedError()
 
     async def _start_game_times_import(self, game_ids: List[str]) -> None:
-        if self._game_times_import_in_progress:
-            raise ImportInProgress()
-
-        context = await self.prepare_game_times_context(game_ids)
-
-        async def import_game_time(game_id, context_):
-            try:
-                game_time = await self.get_game_time(game_id, context_)
-                self._game_time_import_success(game_time)
-            except ApplicationError as error:
-                self._game_time_import_failure(game_id, error)
-            except Exception:
-                logger.exception("Unexpected exception raised in import_game_time")
-                self._game_time_import_failure(game_id, UnknownError())
-
-        async def import_game_times(game_ids_, context_):
-            try:
-                imports = [import_game_time(game_id, context_) for game_id in game_ids_]
-                await asyncio.gather(*imports)
-            finally:
-                self._game_times_import_finished()
-                self._game_times_import_in_progress = False
-                self.game_times_import_complete()
-
-        self._external_task_manager.create_task(
-            import_game_times(game_ids, context),
-            "game times import",
-            handle_exceptions=False
-        )
-        self._game_times_import_in_progress = True
+        await self._game_time_importer.start(game_ids)
 
     async def prepare_game_times_context(self, game_ids: List[str]) -> Any:
         """Override this method to prepare context for get_game_time.
@@ -858,36 +928,7 @@ class Plugin:
         """
 
     async def _start_game_library_settings_import(self, game_ids: List[str]) -> None:
-        if self._game_library_settings_import_in_progress:
-            raise ImportInProgress()
-
-        context = await self.prepare_game_library_settings_context(game_ids)
-
-        async def import_game_library_settings(game_id, context_):
-            try:
-                game_library_settings = await self.get_game_library_settings(game_id, context_)
-                self._game_library_settings_import_success(game_library_settings)
-            except ApplicationError as error:
-                self._game_library_settings_import_failure(game_id, error)
-            except Exception:
-                logger.exception("Unexpected exception raised in import_game_library_settings")
-                self._game_library_settings_import_failure(game_id, UnknownError())
-
-        async def import_game_library_settings_set(game_ids_, context_):
-            try:
-                imports = [import_game_library_settings(game_id, context_) for game_id in game_ids_]
-                await asyncio.gather(*imports)
-            finally:
-                self._game_library_settings_import_finished()
-                self._game_library_settings_import_in_progress = False
-                self.game_library_settings_import_complete()
-
-        self._external_task_manager.create_task(
-            import_game_library_settings_set(game_ids, context),
-            "game library settings import",
-            handle_exceptions=False
-        )
-        self._game_library_settings_import_in_progress = True
+        await self._game_library_settings_importer.start(game_ids)
 
     async def prepare_game_library_settings_context(self, game_ids: List[str]) -> Any:
         """Override this method to prepare context for get_game_library_settings.
@@ -916,37 +957,7 @@ class Plugin:
         """
 
     async def _start_os_compatibility_import(self, game_ids: List[str]) -> None:
-        if self._os_compatibility_import_in_progress:
-            raise ImportInProgress()
-
-        context = await self.prepare_os_compatibility_context(game_ids)
-
-        async def import_os_compatibility(game_id, context_):
-            try:
-                os_compatibility = await self.get_os_compatibility(game_id, context_)
-                self._os_compatibility_import_success(game_id, os_compatibility)
-            except ApplicationError as error:
-                self._os_compatibility_import_failure(game_id, error)
-            except Exception:
-                logger.exception("Unexpected exception raised in import_os_compatibility")
-                self._os_compatibility_import_failure(game_id, UnknownError())
-
-        async def import_os_compatibility_set(game_ids_, context_):
-            try:
-                await asyncio.gather(*[
-                    import_os_compatibility(game_id, context_) for game_id in game_ids_
-                ])
-            finally:
-                self._os_compatibility_import_finished()
-                self._os_compatibility_import_in_progress = False
-                self.os_compatibility_import_complete()
-
-        self._external_task_manager.create_task(
-            import_os_compatibility_set(game_ids, context),
-            "game OS compatibility import",
-            handle_exceptions=False
-        )
-        self._os_compatibility_import_in_progress = True
+        await self._os_compatibility_importer.start(game_ids)
 
     async def prepare_os_compatibility_context(self, game_ids: List[str]) -> Any:
         """Override this method to prepare context for get_os_compatibility.
@@ -972,40 +983,10 @@ class Plugin:
         """Override this method to handle operations after OS compatibility import is finished (like updating cache)."""
 
     async def _start_user_presence_import(self, user_id_list: List[str]) -> None:
-        if self._user_presence_import_in_progress:
-            raise ImportInProgress()
-
-        context = await self.prepare_user_presence_context(user_id_list)
-
-        async def import_user_presence(user_id, context_) -> None:
-            try:
-                self._user_presence_import_success(user_id, await self.get_user_presence(user_id, context_))
-            except ApplicationError as error:
-                self._user_presence_import_failure(user_id, error)
-            except Exception:
-                logger.exception("Unexpected exception raised in import_user_presence")
-                self._user_presence_import_failure(user_id, UnknownError())
-
-        async def import_user_presence_set(user_id_list_, context_) -> None:
-            try:
-                await asyncio.gather(*[
-                    import_user_presence(user_id, context_)
-                    for user_id in user_id_list_
-                ])
-            finally:
-                self._user_presence_import_finished()
-                self._user_presence_import_in_progress = False
-                self.user_presence_import_complete()
-
-        self._external_task_manager.create_task(
-            import_user_presence_set(user_id_list, context),
-            "user presence import",
-            handle_exceptions=False
-        )
-        self._user_presence_import_in_progress = True
+        await self._user_presence_importer.start(user_id_list)
 
     async def prepare_user_presence_context(self, user_id_list: List[str]) -> Any:
-        """Override this method to prepare context for get_user_presence.
+        """Override this method to prepare context for :meth:`get_user_presence`.
         This allows for optimizations like batch requests to platform API.
         Default implementation returns None.
 
@@ -1027,6 +1008,82 @@ class Plugin:
     def user_presence_import_complete(self) -> None:
         """Override this method to handle operations after presence import is finished (like updating cache)."""
 
+    async def _start_local_size_import(self, game_ids: List[str]) -> None:
+        await self._local_size_importer.start(game_ids)
+
+    async def prepare_local_size_context(self, game_ids: List[str]) -> Any:
+        """Override this method to prepare context for :meth:`get_local_size`
+        Default implementation returns None.
+
+        :param game_ids: the ids of the games for which information about size is imported
+        :return: context
+        """
+        return None
+
+    async def get_local_size(self, game_id: str, context: Any) -> Optional[int]:
+        """Override this method to return installed game size.
+
+        .. note::
+          It is preferable to avoid iterating over local game files when overriding this method.
+          If possible, please use a more efficient way of game size retrieval.
+
+        :param context: the value returned from :meth:`prepare_local_size_context`
+        :return: game size (in bytes) or `None` if game size cannot be determined;
+            '0' if the game is not installed, or if it is not present locally (e.g. installed
+            on another machine and accessible via remote connection, playable via web browser etc.)
+        """
+        raise NotImplementedError()
+
+    def local_size_import_complete(self) -> None:
+        """Override this method to handle operations after local game size import is finished (like updating cache)."""
+
+    async def get_subscriptions(self) -> List[Subscription]:
+        """Override this method to return a list of
+        Subscriptions available on platform.
+        This method is called by the GOG Galaxy Client.
+        """
+        raise NotImplementedError()
+
+    async def _start_subscription_games_import(self, subscription_names: List[str]) -> None:
+        await self._subscription_games_importer.start(subscription_names)
+
+    async def prepare_subscription_games_context(self, subscription_names: List[str]) -> Any:
+        """Override this method to prepare context for :meth:`get_subscription_games`
+        Default implementation returns None.
+
+        :param subscription_names: the names of the subscriptions' for which subscriptions games are imported
+        :return: context
+        """
+        return None
+
+    async def get_subscription_games(self, subscription_name: str, context: Any) -> AsyncGenerator[
+        List[SubscriptionGame], None]:
+        """Override this method to provide SubscriptionGames for a given subscription.
+        This method should `yield` a list of SubscriptionGames -> yield [sub_games]
+
+        This method will only be used if :meth:`get_subscriptions` has been implemented.
+
+        :param context: the value returned from :meth:`prepare_subscription_games_context`
+        :return a generator object that yields SubscriptionGames
+
+        .. code-block:: python
+            :linenos:
+
+            async def get_subscription_games(subscription_name: str, context: Any):
+                while True:
+                    games_page = await self._get_subscriptions_from_backend(subscription_name, i)
+                    if not games_pages:
+                        yield None
+                    yield [SubGame(game['game_id'], game['game_title']) for game in games_page]
+
+        """
+        raise NotImplementedError()
+
+    def subscription_games_import_complete(self) -> None:
+        """Override this method to handle operations after
+        subscription games import is finished (like updating cache).
+        """
+
 
 def create_and_run_plugin(plugin_class, argv):
     """Call this method as an entry point for the implemented integration.
@@ -1067,10 +1124,14 @@ def create_and_run_plugin(plugin_class, argv):
 
     async def coroutine():
         reader, writer = await asyncio.open_connection("127.0.0.1", port)
-        extra_info = writer.get_extra_info("sockname")
-        logger.info("Using local address: %s:%u", *extra_info)
-        async with plugin_class(reader, writer, token) as plugin:
-            await plugin.run()
+        try:
+            extra_info = writer.get_extra_info("sockname")
+            logger.info("Using local address: %s:%u", *extra_info)
+            async with plugin_class(reader, writer, token) as plugin:
+                await plugin.run()
+        finally:
+            writer.close()
+            await writer.wait_closed()
 
     try:
         if sys.platform == "win32":

src/galaxy/api/types.py

@@ -62,10 +62,10 @@ class NextStep:
                 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`}
+        "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.
+        on every document at given step of internal browser authentication.
     """
     next_step: str
     auth_params: Dict[str, str]
@@ -216,3 +216,30 @@ class UserPresence:
     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.
+    """
+    subscription_name: str
+    owned: Optional[bool] = None
+    end_time: Optional[int] = None
+
+
+@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,8 +1,8 @@
 """
-This module standarize http traffic and the error handling for further communication with the GOG Galaxy 2.0.
+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.
-Examplary simple web service could looks like:
+Exemplary simple web service could looks like:
 
     .. code-block:: python
 
@@ -72,7 +72,7 @@ class HttpClient:
 
 def create_tcp_connector(*args, **kwargs) -> aiohttp.TCPConnector:
     """
-    Creates TCP connector with resonable defaults.
+    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>`_
     """
@@ -86,11 +86,11 @@ def create_tcp_connector(*args, **kwargs) -> aiohttp.TCPConnector:
 
 def create_client_session(*args, **kwargs) -> aiohttp.ClientSession:
     """
-    Creates client session with resonable defaults.
+    Creates client session with reasonable defaults.
     For details about available parameters refer to
     `aiohttp.ClientSession <https://docs.aiohttp.org/en/stable/client_reference.html>`_
 
-    Examplary customization:
+    Exemplary customization:
 
     .. code-block:: python
 
@@ -124,25 +124,25 @@ def handle_exception():
         raise BackendNotAvailable()
     except aiohttp.ClientConnectionError:
         raise NetworkError()
-    except aiohttp.ContentTypeError:
-        raise UnknownBackendResponse()
+    except aiohttp.ContentTypeError as error:
+        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:
             logger.warning(
                 "Got status %d while performing %s request for %s",
                 error.status, error.request_info.method, str(error.request_info.url)
             )
-            raise UnknownError()
-    except aiohttp.ClientError:
+            raise UnknownError(error.message)
+    except aiohttp.ClientError as e:
         logger.exception("Caught exception while performing request")
-        raise UnknownError()
+        raise UnknownError(repr(e))

tests/conftest.py

@@ -64,6 +64,13 @@ async def plugin(reader, writer):
         "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:

tests/test_features.py

@@ -17,7 +17,10 @@ def test_base_class():
         Feature.LaunchPlatformClient,
         Feature.ImportGameLibrarySettings,
         Feature.ImportOSCompatibility,
-        Feature.ImportUserPresence
+        Feature.ImportUserPresence,
+        Feature.ImportLocalSize,
+        Feature.ImportSubscriptions,
+        Feature.ImportSubscriptionGames
     }
 
 

tests/test_local_size.py

@@ -0,0 +1,188 @@
+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),
+        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()
+
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": request_id,
+            "result": None
+        },
+        {
+            "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)
+    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_subscriptions.py

@@ -0,0 +1,335 @@
+import pytest
+
+from galaxy.api.types import Subscription, SubscriptionGame
+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("3", True, 1580899100)
+    ])
+    await plugin.run()
+    plugin.get_subscriptions.assert_called_with()
+
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "result": {
+                "subscriptions": [
+                    {
+                        "subscription_name": "1"
+                    },
+                    {
+                        "subscription_name": "2",
+                        "owned": False
+                    },
+                    {
+                        "subscription_name": "3",
+                        "owned": True,
+                        "end_time": 1580899100
+                    }
+                ]
+            }
+        }
+    ]
+
+
+@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
+