Increment version

This reverts commit 21ab8bf33d3c2714b8c7af2da8036fe8baae20ea.

.gitignore

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

requirements.txt

@@ -1,5 +1,5 @@
 -e .
-pytest==4.2.0
+pytest==5.2.2
 pytest-asyncio==0.10.0
 pytest-mock==1.10.3
 pytest-mypy==0.4.1

setup.py

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

src/galaxy/api/consts.py

@@ -90,6 +90,7 @@ class Platform(Enum):
     Playfire = "playfire"
     Oculus = "oculus"
     Test = "test"
+    Rockstar = "rockstar"
 
 
 class Feature(Enum):
@@ -111,6 +112,8 @@ class Feature(Enum):
     ShutdownPlatformClient = "ShutdownPlatformClient"
     LaunchPlatformClient = "LaunchPlatformClient"
     ImportGameLibrarySettings = "ImportGameLibrarySettings"
+    ImportOSCompatibility = "ImportOSCompatibility"
+    ImportUserPresence = "ImportUserPresence"
 
 
 class LicenseType(Enum):
@@ -129,3 +132,20 @@ 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"

src/galaxy/api/jsonrpc.py

@@ -64,6 +64,7 @@ class UnknownError(ApplicationError):
         super().__init__(0, "Unknown error", data)
 
 Request = namedtuple("Request", ["method", "params", "id"], defaults=[{}, None])
+Response = namedtuple("Response", ["id", "result", "error"], defaults=[None, {}, {}])
 Method = namedtuple("Method", ["callback", "signature", "immediate", "sensitive_params"])
 
 
@@ -79,7 +80,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)
@@ -88,6 +89,9 @@ class Server():
         self._methods = {}
         self._notifications = {}
         self._task_manager = TaskManager("jsonrpc server")
+        self._write_lock = asyncio.Lock()
+        self._last_request_id = 0
+        self._requests_futures = {}
 
     def register_method(self, name, callback, immediate, sensitive_params=False):
         """
@@ -113,6 +117,47 @@ class Server():
         """
         self._notifications[name] = Method(callback, inspect.signature(callback), immediate, sensitive_params)
 
+    async def send_request(self, method, params, sensitive_params):
+        """
+        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)
+
+        logging.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
+        """
+
+        logging.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:
             try:
@@ -129,8 +174,9 @@ class Server():
             await asyncio.sleep(0) # To not starve task queue
 
     def close(self):
-        logging.info("Closing JSON-RPC server - not more messages will be read")
-        self._active = False
+        if self._active:
+            logging.info("Closing JSON-RPC server - not more messages will be read")
+            self._active = False
 
     async def wait_closed(self):
         await self._task_manager.wait()
@@ -141,15 +187,40 @@ class Server():
 
     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:
-            self._handle_request(request)
-        else:
-            self._handle_notification(request)
+        if isinstance(message, Request):
+            if message.id is not None:
+                self._handle_request(message)
+            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"
+            logging.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)
@@ -209,25 +280,33 @@ class Server():
             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")
-            if jsonrpc_request.get("jsonrpc") != "2.0":
+            jsonrpc_message = json.loads(data, encoding="utf-8")
+            if jsonrpc_message.get("jsonrpc") != "2.0":
                 raise InvalidRequest()
-            del jsonrpc_request["jsonrpc"]
-            return Request(**jsonrpc_request)
+            del jsonrpc_message["jsonrpc"]
+            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):
+        async def send_task(data_):
+            async with self._write_lock:
+                self._writer.write(data_)
+                await self._writer.drain()
+
         try:
             line = self._encoder.encode(data)
             logging.debug("Sending data: %s", line)
             data = (line + "\n").encode("utf-8")
-            self._writer.write(data)
-            self._task_manager.create_task(self._writer.drain(), "drain")
+            self._task_manager.create_task(send_task(data), "send")
         except TypeError as error:
             logging.error(str(error))
 
@@ -248,6 +327,23 @@ class Server():
 
         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)
+
+    def _send_notification(self, method, params):
+        notification = {
+            "jsonrpc": "2.0",
+            "method": method,
+            "params": params
+        }
+        self._send(notification)
+
     @staticmethod
     def _log_request(request, sensitive_params):
         params = anonymise_sensitive_params(request.params, sensitive_params)
@@ -256,44 +352,14 @@ class Server():
         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 = {}
-        self._task_manager = TaskManager("notification client")
-
-    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)
-
-    async def close(self):
-        await self._task_manager.wait()
-
-    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)
-            self._task_manager.create_task(self._writer.drain(), "drain")
-        except TypeError as error:
-            logging.error("Failed to parse outgoing message: %s", str(error))
+    @staticmethod
+    def _log_response(response, sensitive_params):
+        result = anonymise_sensitive_params(response.result, sensitive_params)
+        logging.info("Handling response: id=%s, result=%s", response.id, result)
 
     @staticmethod
-    def _log(method, params, sensitive_params):
-        params = anonymise_sensitive_params(params, sensitive_params)
-        logging.info("Sending notification: method=%s, params=%s", method, params)
+    def _log_error(response, error, sensitive_params):
+        data = anonymise_sensitive_params(error.data, sensitive_params)
+        logging.info("Handling error: id=%s, code=%s, description=%s, data=%s",
+            response.id, error.code, error.message, data
+        )

src/galaxy/api/plugin.py

@@ -7,12 +7,15 @@ import sys
 from enum import Enum
 from typing import Any, Dict, List, Optional, Set, Union
 
-from galaxy.api.consts import Feature
+from galaxy.api.consts import Feature, OSCompatibility
 from galaxy.api.errors import ImportInProgress, UnknownError
-from galaxy.api.jsonrpc import ApplicationError, NotificationClient, Server
-from galaxy.api.types import Achievement, Authentication, FriendInfo, Game, GameTime, LocalGame, NextStep, GameLibrarySettings
+from galaxy.api.jsonrpc import ApplicationError, Connection
+from galaxy.api.types import (
+    Achievement, Authentication, Game, GameLibrarySettings, GameTime, LocalGame, NextStep, UserInfo, UserPresence
+)
 from galaxy.task_manager import TaskManager
 
+
 class JSONEncoder(json.JSONEncoder):
     def default(self, o):  # pylint: disable=method-hidden
         if dataclasses.is_dataclass(o):
@@ -41,12 +44,13 @@ class Plugin:
         self._handshake_token = handshake_token
 
         encoder = JSONEncoder()
-        self._server = Server(self._reader, self._writer, encoder)
-        self._notification_client = NotificationClient(self._writer, encoder)
+        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()
 
@@ -113,6 +117,12 @@ class Plugin:
         self._register_method("start_game_library_settings_import", self._start_game_library_settings_import)
         self._detect_feature(Feature.ImportGameLibrarySettings, ["get_game_library_settings"])
 
+        self._register_method("start_os_compatibility_import", self._start_os_compatibility_import)
+        self._detect_feature(Feature.ImportOSCompatibility, ["get_os_compatibility"])
+
+        self._register_method("start_user_presence_import", self._start_user_presence_import)
+        self._detect_feature(Feature.ImportUserPresence, ["get_user_presence"])
+
     async def __aenter__(self):
         return self
 
@@ -153,7 +163,7 @@ class Plugin:
                 result = handler(*args, **kwargs)
                 return wrap_result(result)
 
-            self._server.register_method(name, method, True, sensitive_params)
+            self._connection.register_method(name, method, True, sensitive_params)
         else:
             async def method(*args, **kwargs):
                 if not internal:
@@ -163,37 +173,40 @@ class Plugin:
                 result = await handler_(*args, **kwargs)
                 return wrap_result(result)
 
-            self._server.register_method(name, method, False, sensitive_params)
+            self._connection.register_method(name, method, False, sensitive_params)
 
     def _register_notification(self, name, handler, internal=False, immediate=False, sensitive_params=False):
         if not internal and not immediate:
             handler = self._wrap_external_method(handler, name)
-        self._server.register_notification(name, handler, immediate, sensitive_params)
+        self._connection.register_notification(name, handler, immediate, sensitive_params)
 
     def _wrap_external_method(self, handler, name: str):
         async def wrapper(*args, **kwargs):
             return await self._external_task_manager.create_task(handler(*args, **kwargs), name, False)
+
         return wrapper
 
     async def run(self):
         """Plugin's main coroutine."""
-        await self._server.run()
+        await self._connection.run()
+        logging.debug("Plugin run loop finished")
 
     def close(self) -> None:
         if not self._active:
             return
 
         logging.info("Closing plugin")
-        self._server.close()
+        self._connection.close()
         self._external_task_manager.cancel()
         self._internal_task_manager.create_task(self.shutdown(), "shutdown")
         self._active = False
 
     async def wait_closed(self) -> None:
+        logging.debug("Waiting for plugin to close")
         await self._external_task_manager.wait()
         await self._internal_task_manager.wait()
-        await self._server.wait_closed()
-        await self._notification_client.close()
+        await self._connection.wait_closed()
+        logging.debug("Plugin closed")
 
     def create_task(self, coro, description):
         """Wrapper around asyncio.create_task - takes care of canceling tasks on shutdown"""
@@ -256,9 +269,9 @@ class Plugin:
 
         """
         # temporary solution for persistent_cache vs credentials issue
-        self.persistent_cache['credentials'] = credentials  # type: ignore
+        self.persistent_cache["credentials"] = credentials  # type: ignore
 
-        self._notification_client.notify("store_credentials", credentials, sensitive_params=True)
+        self._connection.send_notification("store_credentials", credentials, sensitive_params=True)
 
     def add_game(self, game: Game) -> None:
         """Notify the client to add game to the list of owned games
@@ -280,7 +293,7 @@ class Plugin:
 
         """
         params = {"owned_game": game}
-        self._notification_client.notify("owned_game_added", params)
+        self._connection.send_notification("owned_game_added", params)
 
     def remove_game(self, game_id: str) -> None:
         """Notify the client to remove game from the list of owned games
@@ -302,7 +315,7 @@ class Plugin:
 
         """
         params = {"game_id": game_id}
-        self._notification_client.notify("owned_game_removed", params)
+        self._connection.send_notification("owned_game_removed", params)
 
     def update_game(self, game: Game) -> None:
         """Notify the client to update the status of a game
@@ -311,7 +324,7 @@ class Plugin:
         :param game: Game to update
         """
         params = {"owned_game": game}
-        self._notification_client.notify("owned_game_updated", params)
+        self._connection.send_notification("owned_game_updated", params)
 
     def unlock_achievement(self, game_id: str, achievement: Achievement) -> None:
         """Notify the client to unlock an achievement for a specific game.
@@ -323,24 +336,24 @@ class Plugin:
             "game_id": game_id,
             "achievement": achievement
         }
-        self._notification_client.notify("achievement_unlocked", params)
+        self._connection.send_notification("achievement_unlocked", params)
 
     def _game_achievements_import_success(self, game_id: str, achievements: List[Achievement]) -> None:
         params = {
             "game_id": game_id,
             "unlocked_achievements": achievements
         }
-        self._notification_client.notify("game_achievements_import_success", params)
+        self._connection.send_notification("game_achievements_import_success", params)
 
     def _game_achievements_import_failure(self, game_id: str, error: ApplicationError) -> None:
         params = {
             "game_id": game_id,
             "error": error.json()
         }
-        self._notification_client.notify("game_achievements_import_failure", params)
+        self._connection.send_notification("game_achievements_import_failure", params)
 
     def _achievements_import_finished(self) -> None:
-        self._notification_client.notify("achievements_import_finished", None)
+        self._connection.send_notification("achievements_import_finished", None)
 
     def update_local_game_status(self, local_game: LocalGame) -> None:
         """Notify the client to update the status of a local game.
@@ -366,15 +379,15 @@ class Plugin:
                     self._check_statuses_task = asyncio.create_task(self._check_statuses())
         """
         params = {"local_game": local_game}
-        self._notification_client.notify("local_game_status_changed", params)
+        self._connection.send_notification("local_game_status_changed", params)
 
-    def add_friend(self, user: FriendInfo) -> None:
+    def add_friend(self, user: UserInfo) -> None:
         """Notify the client to add a user to friends list of the currently authenticated user.
 
-        :param user: FriendInfo of a user that the client will add to friends list
+        :param user: UserInfo of a user that the client will add to friends list
         """
         params = {"friend_info": user}
-        self._notification_client.notify("friend_added", params)
+        self._connection.send_notification("friend_added", params)
 
     def remove_friend(self, user_id: str) -> None:
         """Notify the client to remove a user from friends list of the currently authenticated user.
@@ -382,7 +395,7 @@ class Plugin:
         :param user_id: id of the user to remove from friends list
         """
         params = {"user_id": user_id}
-        self._notification_client.notify("friend_removed", params)
+        self._connection.send_notification("friend_removed", params)
 
     def update_game_time(self, game_time: GameTime) -> None:
         """Notify the client to update game time for a game.
@@ -390,51 +403,96 @@ class Plugin:
         :param game_time: game time to update
         """
         params = {"game_time": game_time}
-        self._notification_client.notify("game_time_updated", params)
+        self._connection.send_notification("game_time_updated", params)
 
     def _game_time_import_success(self, game_time: GameTime) -> None:
         params = {"game_time": game_time}
-        self._notification_client.notify("game_time_import_success", params)
+        self._connection.send_notification("game_time_import_success", params)
 
     def _game_time_import_failure(self, game_id: str, error: ApplicationError) -> None:
         params = {
             "game_id": game_id,
             "error": error.json()
         }
-        self._notification_client.notify("game_time_import_failure", params)
+        self._connection.send_notification("game_time_import_failure", params)
 
     def _game_times_import_finished(self) -> None:
-        self._notification_client.notify("game_times_import_finished", None)
+        self._connection.send_notification("game_times_import_finished", None)
 
     def _game_library_settings_import_success(self, game_library_settings: GameLibrarySettings) -> None:
         params = {"game_library_settings": game_library_settings}
-        self._notification_client.notify("game_library_settings_import_success", params)
+        self._connection.send_notification("game_library_settings_import_success", params)
 
     def _game_library_settings_import_failure(self, game_id: str, error: ApplicationError) -> None:
         params = {
             "game_id": game_id,
             "error": error.json()
         }
-        self._notification_client.notify("game_library_settings_import_failure", params)
+        self._connection.send_notification("game_library_settings_import_failure", params)
 
     def _game_library_settings_import_finished(self) -> None:
-        self._notification_client.notify("game_library_settings_import_finished", None)
+        self._connection.send_notification("game_library_settings_import_finished", None)
+
+    def _os_compatibility_import_success(self, game_id: str, os_compatibility: Optional[OSCompatibility]) -> None:
+        self._connection.send_notification(
+            "os_compatibility_import_success",
+            {
+                "game_id": game_id,
+                "os_compatibility": os_compatibility
+            }
+        )
+
+    def _os_compatibility_import_failure(self, game_id: str, error: ApplicationError) -> None:
+        self._connection.send_notification(
+            "os_compatibility_import_failure",
+            {
+                "game_id": game_id,
+                "error": error.json()
+            }
+        )
+
+    def _os_compatibility_import_finished(self) -> None:
+        self._connection.send_notification("os_compatibility_import_finished", None)
+
+    def _user_presence_import_success(self, user_id: str, user_presence: UserPresence) -> None:
+        self._connection.send_notification(
+            "user_presence_import_success",
+            {
+                "user_id": user_id,
+                "presence": user_presence
+            }
+        )
+
+    def _user_presence_import_failure(self, user_id: str, error: ApplicationError) -> None:
+        self._connection.send_notification(
+            "user_presence_import_failure",
+            {
+                "user_id": user_id,
+                "error": error.json()
+            }
+        )
+
+    def _user_presence_import_finished(self) -> None:
+        self._connection.send_notification("user_presence_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.
          """
-        self._notification_client.notify("authentication_lost", None)
+        self._connection.send_notification("authentication_lost", None)
 
     def push_cache(self) -> None:
         """Push local copy of the persistent cache to the GOG Galaxy Client replacing existing one.
         """
-        self._notification_client.notify(
+        self._connection.send_notification(
             "push_cache",
             params={"data": self._persistent_cache},
             sensitive_params="data"
         )
 
+    async def refresh_credentials(self, params: Dict[str, Any], sensitive_params) -> Dict[str, Any]:
+        return await self._connection.send_request("refresh_credentials", params, sensitive_params)
+
     # handlers
     def handshake_complete(self) -> None:
         """This method is called right after the handshake with the GOG Galaxy Client is complete and
@@ -690,7 +748,7 @@ class Plugin:
         This method is called by the GOG Galaxy Client."""
         raise NotImplementedError()
 
-    async def get_friends(self) -> List[FriendInfo]:
+    async def get_friends(self) -> List[UserInfo]:
         """Override this method to return the friends list
         of the currently authenticated user.
         This method is called by the GOG Galaxy Client.
@@ -805,7 +863,7 @@ class Plugin:
         This allows for optimizations like batch requests to platform API.
         Default implementation returns None.
 
-        :param game_ids: the ids of the games for which game time are imported
+        :param game_ids: the ids of the games for which game library settings are imported
         :return: context
         """
         return None
@@ -815,17 +873,130 @@ class Plugin:
         identified by the provided game_id.
         This method is called by import task initialized by GOG Galaxy Client.
 
-        :param game_id: the id of the game for which the game time is returned
+        :param game_id: the id of the game for which the game library settings are imported
         :param context: the value returned from :meth:`prepare_game_library_settings_context`
         :return: GameLibrarySettings object
         """
         raise NotImplementedError()
 
     def game_library_settings_import_complete(self) -> None:
-        """Override this method to handle operations after game times import is finished
+        """Override this method to handle operations after game library settings import is finished
         (like updating cache).
         """
 
+    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:
+                logging.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
+
+    async def prepare_os_compatibility_context(self, game_ids: List[str]) -> Any:
+        """Override this method to prepare context for get_os_compatibility.
+        This allows for optimizations like batch requests to platform API.
+        Default implementation returns None.
+
+        :param game_ids: the ids of the games for which game os compatibility is imported
+        :return: context
+        """
+        return None
+
+    async def get_os_compatibility(self, game_id: str, context: Any) -> Optional[OSCompatibility]:
+        """Override this method to return the OS compatibility for the game with the provided game_id.
+        This method is called by import task initialized by GOG Galaxy Client.
+
+        :param game_id: the id of the game for which the game os compatibility is imported
+        :param context: the value returned from :meth:`prepare_os_compatibility_context`
+        :return: OSCompatibility flags indicating compatible OSs, or None if compatibility is not know
+        """
+        raise NotImplementedError()
+
+    def os_compatibility_import_complete(self) -> None:
+        """Override this method to handle operations after OS compatibility import is finished (like updating cache)."""
+
+    async def _start_user_presence_import(self, user_ids: List[str]) -> None:
+        if self._user_presence_import_in_progress:
+            raise ImportInProgress()
+
+        context = await self.prepare_user_presence_context(user_ids)
+
+        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:
+                logging.exception("Unexpected exception raised in import_user_presence")
+                self._user_presence_import_failure(user_id, UnknownError())
+
+        async def import_user_presence_set(user_ids_, context_) -> None:
+            try:
+                await asyncio.gather(*[
+                    import_user_presence(user_id, context_)
+                    for user_id in user_ids_
+                ])
+            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_ids, context),
+            "user presence import",
+            handle_exceptions=False
+        )
+        self._user_presence_import_in_progress = True
+
+    async def prepare_user_presence_context(self, user_ids: List[str]) -> Any:
+        """Override this method to prepare context for get_user_presence.
+        This allows for optimizations like batch requests to platform API.
+        Default implementation returns None.
+
+        :param user_ids: the ids of the users for whom presence information is imported
+        :return: context
+        """
+        return None
+
+    async def get_user_presence(self, user_id: str, context: Any) -> UserPresence:
+        """Override this method to return presence information for the user with the provided user_id.
+        This method is called by import task initialized by GOG Galaxy Client.
+
+        :param user_id: the id of the user for whom presence information is imported
+        :param context: the value returned from :meth:`prepare_user_presence_context`
+        :return: UserPresence presence information of the provided user
+        """
+        raise NotImplementedError()
+
+    def user_presence_import_complete(self) -> None:
+        """Override this method to handle operations after presence 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.
 

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
 
-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:
 
@@ -67,8 +70,9 @@ class NextStep():
     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 +81,9 @@ class LicenseInfo():
     license_type: LicenseType
     owner: Optional[str] = None
 
+
 @dataclass
-class Dlc():
+class Dlc:
     """Downloadable content object.
 
     :param dlc_id: id of the dlc
@@ -89,8 +94,9 @@ class Dlc():
     dlc_title: str
     license_info: LicenseInfo
 
+
 @dataclass
-class Game():
+class Game:
     """Game object.
 
     :param game_id: unique identifier of the game, this will be passed as parameter for methods such as launch_game
@@ -103,8 +109,9 @@ class Game():
     dlcs: Optional[List[Dlc]]
     license_info: LicenseInfo
 
+
 @dataclass
-class Achievement():
+class Achievement:
     """Achievement, has to be initialized with either id or name.
 
     :param unlock_time: unlock time of the achievement
@@ -119,8 +126,9 @@ class Achievement():
         assert self.achievement_id or self.achievement_name, \
             "One of achievement_id or achievement_name is required"
 
+
 @dataclass
-class LocalGame():
+class LocalGame:
     """Game locally present on the authenticated user's computer.
 
     :param game_id: id of the game
@@ -129,9 +137,24 @@ class LocalGame():
     game_id: str
     local_game_state: LocalGameState
 
+
 @dataclass
-class FriendInfo():
-    """Information about a friend of the currently authenticated user.
+class FriendInfo:
+    """
+    .. 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
+    """
+    user_id: str
+    user_name: str
+
+
+@dataclass
+class UserInfo:
+    """Information about a user of related user.
 
     :param user_id: id of the user
     :param user_name: username of the user
@@ -140,7 +163,7 @@ class FriendInfo():
     user_name: str
 
 @dataclass
-class GameTime():
+class GameTime:
     """Game time of a game, defines the total time spent in the game
     and the last time the game was played.
 
@@ -152,8 +175,9 @@ class GameTime():
     time_played: Optional[int]
     last_played_time: Optional[int]
 
+
 @dataclass
-class GameLibrarySettings():
+class GameLibrarySettings:
     """Library settings of a game, defines assigned tags and visibility flag.
 
     :param game_id: id of the related game
@@ -163,3 +187,18 @@ class GameLibrarySettings():
     game_id: str
     tags: Optional[List[str]]
     hidden: Optional[bool]
+
+
+@dataclass
+class UserPresence:
+    """Presence information of a user.
+
+    :param presence_state: the state of the user
+    :param game_id: id of the game a user is currently in
+    :param game_title: name of the game a user is currently in
+    :param presence_status: detailed user's presence description
+    """
+    presence_state: PresenceState
+    game_id: Optional[str] = None
+    game_title: Optional[str] = None
+    presence_status: Optional[str] = None

src/galaxy/http.py

@@ -78,7 +78,8 @@ def create_tcp_connector(*args, **kwargs) -> aiohttp.TCPConnector:
     ssl_context.load_verify_locations(certifi.where())
     kwargs.setdefault("ssl", ssl_context)
     kwargs.setdefault("limit", DEFAULT_LIMIT)
-    return aiohttp.TCPConnector(*args, **kwargs)  # type: ignore due to https://github.com/python/mypy/issues/4001
+    # due to https://github.com/python/mypy/issues/4001
+    return aiohttp.TCPConnector(*args, **kwargs)  # type: ignore
 
 
 def create_client_session(*args, **kwargs) -> aiohttp.ClientSession:
@@ -103,7 +104,8 @@ def create_client_session(*args, **kwargs) -> aiohttp.ClientSession:
     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)  # type: ignore due to https://github.com/python/mypy/issues/4001
+    # due to https://github.com/python/mypy/issues/4001
+    return aiohttp.ClientSession(*args, **kwargs)  # type: ignore
 
 
 @contextmanager

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

@@ -1,4 +1,6 @@
 import sys
+
+
 if sys.platform == "win32":
     import logging
     import ctypes
@@ -76,11 +78,10 @@ class RegistryMonitor:
         if self._key is None:
             self._open_key()
 
-        if self._key is None:
-            return False
+        if self._key is not None:
+            self._set_key_update_notification()
 
-        self._set_key_update_notification()
-        return True
+        return False
 
     def _set_key_update_notification(self):
         filter_ = REG_NOTIFY_CHANGE_NAME | REG_NOTIFY_CHANGE_LAST_SET

tests/conftest.py

@@ -1,33 +1,38 @@
-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():
     stream = MagicMock(name="stream_reader")
     stream.read = MagicMock()
     yield stream
 
+
 @pytest.fixture()
 async def writer():
     stream = MagicMock(name="stream_writer")
     stream.drain.side_effect = lambda: async_return_value(None)
     yield stream
 
+
 @pytest.fixture()
 def read(reader):
     yield reader.read
 
+
 @pytest.fixture()
 def write(writer):
     yield writer.write
 
+
 @pytest.fixture()
 async def plugin(reader, writer):
     """Return plugin instance with all feature methods mocked"""
@@ -53,6 +58,12 @@ async def plugin(reader, writer):
         "get_game_library_settings",
         "prepare_game_library_settings_context",
         "game_library_settings_import_complete",
+        "get_os_compatibility",
+        "prepare_os_compatibility_context",
+        "os_compatibility_import_complete",
+        "get_user_presence",
+        "prepare_user_presence_context",
+        "user_presence_import_complete",
     )
 
     with ExitStack() as stack:

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
 
@@ -201,6 +201,7 @@ async def test_import_in_progress(plugin, read, write):
 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
 
@@ -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

@@ -15,7 +15,9 @@ def test_base_class():
         Feature.ImportFriends,
         Feature.ShutdownPlatformClient,
         Feature.LaunchPlatformClient,
-        Feature.ImportGameLibrarySettings
+        Feature.ImportGameLibrarySettings,
+        Feature.ImportOSCompatibility,
+        Feature.ImportUserPresence
     }
 
 

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
 
@@ -17,8 +17,8 @@ async def test_get_friends_success(plugin, read, write):
 
     read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
     plugin.get_friends.return_value = async_return_value([
-        FriendInfo("3", "Jan"),
-        FriendInfo("5", "Ola")
+        UserInfo("3", "Jan"),
+        UserInfo("5", "Ola")
     ])
     await plugin.run()
     plugin.get_friends.assert_called_with()
@@ -64,9 +64,10 @@ 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")
 
     plugin.add_friend(friend)
+    await skip_loop()
 
     assert get_messages(write) == [
         {
@@ -82,6 +83,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) == [
         {

tests/test_game_library_settings.py

@@ -9,7 +9,7 @@ from tests import create_message, get_messages
 
 
 @pytest.mark.asyncio
-async def test_get_game_time_success(plugin, read, write):
+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",

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
 
@@ -199,6 +199,7 @@ async def test_import_in_progress(plugin, read, write):
 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_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
 
@@ -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_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
 
@@ -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
+)
+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
+])
+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_user_presence.py

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