Compare commits

..

118 Commits
0.28.1 ... 0.48

Author SHA1 Message Date
Romuald Juchnowicz-Bierbasz
bab0be9994 Increment version 2019-09-02 17:31:47 +02:00
Romuald Juchnowicz-Bierbasz
0294e2a1f1 SDK-3023: Introduce task managers 2019-09-02 17:28:13 +02:00
Mieszko Banczerowski
0ab00e4119 Fix typing in persistent cache 2019-08-23 17:48:42 +02:00
Mieszko Banczerowski
b20fce057b GPI-661: Add docs to http module 2019-08-23 17:46:08 +02:00
Romuald Juchnowicz-Bierbasz
dec59f47dd Add more detials to launch_platform_client method 2019-08-22 09:42:10 +02:00
Mieszko Banczerowski
ca85b2428b GPI-517: Mark coroutines in docs 2019-08-20 14:35:54 +02:00
Pawel Czoppa
d8a00d58a6 Merge branch 'master' of https://gitlab.gog.com/galaxy-client/galaxy-plugin-api into platform_id_update
# Conflicts:
#	PLATFORM_IDs.md
2019-08-20 14:22:36 +02:00
Romuald Juchnowicz-Bierbasz
d4cd1cedfd Install wheel in deploy 2019-08-14 16:07:01 +02:00
Romuald Juchnowicz-Bierbasz
161122b94d SDK-2988: Upload to official pypi server 2019-08-14 13:48:35 +02:00
Romuald Juchnowicz-Bierbasz
cec36695b6 Increment version 2019-08-13 16:08:59 +02:00
Romuald Juchnowicz-Bierbasz
4cc8be8f5d SDK-2997: Add launch_platform_client method 2019-08-13 15:23:20 +02:00
Vadim Suharnikov
f5eb32aa19 Fix a comment typo 2019-08-04 01:06:22 +02:00
Romuald Bierbasz
a76345ff6b Docs fixes 2019-08-02 17:23:56 +02:00
Romuald Bierbasz
c3bbeee54d Turn on type checking 2019-08-02 15:15:29 +02:00
Romuald Juchnowicz-Bierbasz
13a3f7577b Increment version 2019-08-02 15:13:11 +02:00
Romuald Bierbasz
f5b9adfbd5 Add achievements_import_complete and game_times_import_complete 2019-08-02 15:11:05 +02:00
Romuald Juchnowicz-Bierbasz
e33dd09a8d Increment version 2019-08-02 12:57:13 +02:00
Aleksej Pawlowskij
f4ea2af924 Make 'handshake_complete' safe 2019-08-02 12:29:48 +02:00
Romuald Bierbasz
3bcc674518 Use MagicMocks with async_return_value and pytest.mark.asyncio 2019-08-02 12:28:13 +02:00
Romuald Juchnowicz-Bierbasz
b14595bef5 Fix types 2019-08-02 12:20:27 +02:00
Romuald Bierbasz
a9acb7a0db SDK-2931: Refactor import methods 2019-08-02 10:10:58 +02:00
Mesco
d95aacb9d8 Fix typo in docstring 2019-08-01 09:53:50 +02:00
Aleksej Pawlowskij
49ae2beab9 SDK-2983: Split entry methods from feature detection 2019-07-29 12:59:45 +02:00
Aleksej Pawlowskij
c9e190772c Increment version 2019-07-26 16:28:47 +02:00
Aleksej Pawlowskij
789415d31b SDK-2974: add process info tools 2019-07-26 13:41:43 +02:00
Rafal Makagon
223adf6a38 Remove chat and users 2019-07-24 10:34:22 +02:00
Romuald Juchnowicz-Bierbasz
bfb63a42bd SDK-2951: Add create_task method 2019-07-23 17:29:20 +02:00
Romuald Juchnowicz-Bierbasz
53b3062719 SDK-2951: Use WindowsProactorEventLoopPolicy on Windows 2019-07-23 17:08:48 +02:00
Steven M. Vascellaro
49eb10ac8a Add MIT LICENSE (#19)
Based on license from 'galaxy-csharp-demo-game'
da1b7f1453/LICENSE.md
2019-07-18 11:26:47 +02:00
Romuald Juchnowicz-Bierbasz
10ecef791f Increment version 2019-07-17 15:35:38 +02:00
Romuald Bierbasz
ce193f39bc SDK-2933: Add shutdown_client notification 2019-07-17 15:27:27 +02:00
Romuald Juchnowicz-Bierbasz
630d878a3c Use constants 2019-07-12 11:46:48 +02:00
Romuald Juchnowicz-Bierbasz
cc63c24bde Increment version 2019-07-12 11:46:48 +02:00
Romuald Juchnowicz-Bierbasz
0d0f657240 SDK-2930: Refactor http module 2019-07-12 11:46:48 +02:00
Steven M. Vascellaro
33c630225d README.md cleanup (#17) 2019-07-12 10:39:37 +02:00
apaulouski
f4bd18a8ab Merge pull request #2 from rogersachan/rogersachan-patch-1
Fix spelling errors in the README
2019-07-10 12:12:07 +02:00
Roger
fa4541434f Merge branch 'master' into rogersachan-patch-1 2019-07-09 13:30:35 -04:00
rbierbasz-gog
c083a3089a Create .travis.yml 2019-07-08 15:43:12 +02:00
Romuald Juchnowicz-Bierbasz
f6b5a12b24 SDK-2932: Remove github deployment (use mirroring) 2019-07-03 13:42:06 +02:00
Romuald Juchnowicz-Bierbasz
8a67747df5 Merge remote-tracking branch 'github/master'
* github/master:
  version 0.38
  version 0.35.2
  version 0.35.1
  version 0.34
  version 0.33.1
  version 0.33
  version 0.32.1
  version 0.32.0
  version 0.31.3
  version 0.31.2
  version 0.31.1
  Initial commit
2019-07-03 13:35:34 +02:00
Romuald Juchnowicz-Bierbasz
2db9d0f383 Increment version 2019-07-01 14:35:05 +02:00
Mieszko Banczerowski
9d93762867 Workaround for removing creds on push_cache 2019-07-01 14:32:23 +02:00
Romuald Juchnowicz-Bierbasz
c364b716f4 Increment version 2019-07-01 13:16:08 +02:00
Romuald Juchnowicz-Bierbasz
48e1782484 SDK-2893: Optional game time and last played 2019-07-01 13:14:07 +02:00
Romuald Juchnowicz-Bierbasz
ff30675a25 Do not invoke tick before handshake 2019-07-01 12:26:05 +02:00
Aliaksei Paulouski
7b3965ff4b Add poe platform 2019-06-28 15:09:46 +02:00
Piotr Marzec
2ebdfabd9b Path of Exile added 2019-06-28 14:49:56 +02:00
Romuald Juchnowicz-Bierbasz
4e1ea8056d Add StreamLineReader with unit tests 2019-06-28 14:00:44 +02:00
Romuald Juchnowicz-Bierbasz
67e8681de6 Increment version 2019-06-28 11:59:01 +02:00
Romuald Juchnowicz-Bierbasz
77d742ce18 SDK-2910: Fix readline 2019-06-28 11:58:32 +02:00
GOG Galaxy SDK Team
f1fd00fcd3 version 0.38 2019-06-26 12:46:23 +02:00
Romuald Juchnowicz-Bierbasz
692bdbf370 Increment version 2019-06-26 12:07:28 +02:00
Romuald Juchnowicz-Bierbasz
207b1e1313 SDK-2880: Fix readline 2019-06-26 12:02:25 +02:00
Romuald Juchnowicz-Bierbasz
05042fe430 Increment version 2019-06-25 17:53:35 +02:00
Romuald Juchnowicz-Bierbasz
58b17d94fa SDK-2880: Read in chunks 2019-06-25 17:53:16 +02:00
Romuald Juchnowicz-Bierbasz
be03c83d45 SDK-2873: Add typing to API 2019-06-24 10:20:04 +02:00
Roger
1edf4ff5ba Fix spelling errors 2019-06-19 15:39:22 -04:00
Piotr Marzec
8d210e7f3e platform name fixing 2019-06-19 20:43:25 +02:00
Rafal Makagon
c07c7a2c2a Platform list 2019-06-18 15:56:45 +02:00
GOG Galaxy SDK Team
9d5d48032e version 0.35.2 2019-06-17 18:11:24 +02:00
Pawel Czoppa
cb1a5fa5e4 fix for merge request 2019-06-17 18:01:09 +02:00
GOG Galaxy SDK Team
179fd147c1 version 0.35.1 2019-06-17 17:41:44 +02:00
Pawel Czoppa
4790238638 Update setup.py 2019-06-17 17:38:43 +02:00
Aliaksei Paulouski
5d90ba0c09 Increment version 2019-06-17 10:39:36 +02:00
Aliaksei Paulouski
d74ed3a4b5 Hide persistent cache data 2019-06-17 10:39:30 +02:00
Pawel Czoppa
d6f2d00fb9 typo 2019-06-15 15:11:05 +02:00
Piotr Marzec
ce9f33f5d0 platform names clieanup 2019-06-15 15:05:12 +02:00
Pawel Czoppa
b28fc60088 include Platform id's in sphinx generated documentation + some fixed 2019-06-15 14:57:20 +02:00
Piotr Marzec
be3d3bb7e5 text cleanup 2019-06-15 10:16:04 +02:00
Piotr Marzec
6dec4a99d3 add platforms list 2019-06-14 19:30:40 +02:00
Piotr Marzec
69ffef2fde create platform id list 2019-06-14 19:25:08 +02:00
GOG Galaxy SDK Team
7789927ed9 version 0.34 2019-06-14 16:54:11 +02:00
Rafal Makagon
da59670d8e Increment version 2019-06-14 16:35:20 +02:00
Rafal Makagon
ed1049b543 Write down unexpected responses from http client 2019-06-14 16:32:07 +02:00
GOG Galaxy SDK Team
e2f26271cb version 0.33.1 2019-06-14 14:49:25 +02:00
Rafal Makagon
9e8748b032 Increment version 2019-06-14 14:41:34 +02:00
Mieszko Banczerowski
bb482d4ed6 SDK-2872 update documentation about plugin deployment 2019-06-14 11:51:30 +02:00
GOG Galaxy SDK Team
3bd0b71ab3 version 0.33 2019-06-13 12:30:12 +02:00
Aliaksei Paulouski
909cc10762 Increment version 2019-06-13 12:26:32 +02:00
Rafal Makagon
9b4537c54f Fix anonimize params method 2019-06-13 12:25:57 +02:00
Rafal Makagon
2e90c66390 Remove not used errors 2019-06-13 12:21:34 +02:00
Aliaksei Paulouski
8647b06ca2 Add TooManyRequests error 2019-06-13 08:52:40 +02:00
Aliaksei Paulouski
f6522be74d SDK-2874: Persistent cache 2019-06-12 17:13:41 +02:00
GOG Galaxy SDK Team
192d655d51 version 0.32.1 2019-06-10 19:04:08 +02:00
Mieszko Banczerowski
5ca9254d2a GPI-438: Add link to docs in readme.md; fix make.py 2019-06-10 18:53:02 +02:00
GOG Galaxy SDK Team
6c6dc42cd6 version 0.32.0 2019-06-07 15:08:47 +02:00
Mieszko Banczerowski
79808e49f7 Bump version after adding documentation 2019-06-07 14:02:23 +02:00
Mateusz Silaczewski
7099cf3195 Merge branch 'documentation' of https://gitlab.gog.com/galaxy-client/galaxy-plugin-api into documentation 2019-06-06 15:56:01 +02:00
GOG Galaxy SDK Team
f97b6c8971 version 0.31.3 2019-05-31 12:09:18 +02:00
Rafal Makagon
ed91fd582c Deploy setup py 2019-05-31 12:08:49 +02:00
GOG Galaxy SDK Team
0af7387342 version 0.31.2 2019-05-31 11:53:15 +02:00
Rafal Makagon
bd393a96f0 Increment version 2019-05-31 11:50:53 +02:00
Piotr Marzec
c53aab1abb Legal Notice added 2019-05-31 11:29:05 +02:00
GOG Galaxy SDK Team
60fab25a55 version 0.31.1 2019-05-29 13:09:13 +02:00
GOG Galaxy SDK Team
6f717a1e31 Initial commit 2019-05-29 13:08:20 +02:00
Romuald Juchnowicz-Bierbasz
80f40b1971 Increment version 2019-05-29 13:04:35 +02:00
Romuald Juchnowicz-Bierbasz
0da0296154 Release on gogcom 2019-05-28 17:59:11 +02:00
Romuald Juchnowicz-Bierbasz
9a115557b3 Fix username 2019-05-28 12:03:56 +02:00
Romuald Juchnowicz-Bierbasz
14c2d7d9e8 Update README 2019-05-28 11:28:32 +02:00
Romuald Juchnowicz-Bierbasz
4a7a759cea Update release scripts 2019-05-28 11:22:38 +02:00
Romuald Juchnowicz-Bierbasz
da8da24b01 Remove changelog 2019-05-28 11:19:48 +02:00
Rafal Makagon
ccbb13e685 Update README.md 2019-05-27 17:48:42 +02:00
Piotr Marzec
a3ca815975 cosmetic changes to readme.md 2019-05-27 17:41:28 +02:00
Romuald Juchnowicz-Bierbasz
f2d4127a31 Increment version 2019-05-24 14:04:17 +02:00
Romuald Juchnowicz-Bierbasz
07b6edce12 SDK-2840: Refactor import methods 2019-05-24 14:04:05 +02:00
Romuald Bierbasz
ef7f9ccca1 Fix pipeline 2019-05-21 14:27:02 +02:00
Romuald Juchnowicz-Bierbasz
3b296cbcc9 Fix pipeline 2019-05-21 14:18:45 +02:00
Romuald Juchnowicz-Bierbasz
f5361cd5ab Fix pipeline 2019-05-21 14:09:57 +02:00
Romuald Juchnowicz-Bierbasz
758909efba Fix release.py 2019-05-21 11:20:38 +02:00
Romuald Bierbasz
0bc8000f14 Fix release pipeline 2019-05-21 11:10:52 +02:00
Romuald Juchnowicz-Bierbasz
e62e7e0e6e Add pipeline for releasing to github 2019-05-20 15:52:35 +02:00
Romuald Juchnowicz-Bierbasz
be6c0eb03e Increment version 2019-05-20 15:38:24 +02:00
Aliaksei Paulouski
0ee56193de Fix aiohttp exception hierarchy 2019-05-20 11:20:27 +02:00
Romuald Juchnowicz-Bierbasz
6bc91a12fa Register new requests 2019-05-17 16:42:13 +02:00
Romuald Juchnowicz-Bierbasz
6d513d86bf Increment version 2019-05-17 16:42:13 +02:00
Romuald Juchnowicz-Bierbasz
bdd2225262 Add new interface methods for game time and achievements 2019-05-17 16:42:12 +02:00
Rafal Makagon
68fdc4d188 Printing own port 2019-05-17 15:32:24 +02:00
Romuald Juchnowicz-Bierbasz
f283c10a95 Anonymise params in pass_login_credentials 2019-05-15 19:49:50 +02:00
52 changed files with 2691 additions and 1250 deletions

7
.gitignore vendored
View File

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

View File

@@ -14,13 +14,19 @@ test_package:
deploy_package:
stage: deploy
variables:
TWINE_USERNAME: $PYPI_USERNAME
TWINE_PASSWORD: $PYPI_PASSWORD
script:
- pip install twine wheel
- rm -rf dist
- export VERSION=$(python setup.py --version)
- python setup.py sdist --formats=gztar upload -r gog-pypi
- python setup.py sdist --formats=gztar bdist_wheel
- twine upload dist/*
- curl -X POST --silent --show-error --fail
"https://gitlab.gog.com/api/v4/projects/${CI_PROJECT_ID}/repository/tags?tag_name=${VERSION}&ref=${CI_COMMIT_REF_NAME}&private_token=${PACKAGE_DEPLOYER_API_TOKEN}"
when: manual
only:
- master
except:
- tags
- tags

8
.travis.yml Normal file
View File

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

21
LICENSE Normal file
View File

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

82
PLATFORM_IDs.md Normal file
View File

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

133
README.md
View File

@@ -1,45 +1,128 @@
# Galaxy python plugin API
# GOG Galaxy Integrations Python API
## Usage
This Python library allows developers to easily build community integrations for various gaming platforms with GOG Galaxy 2.0.
Implement plugin:
- refer to our <a href='https://galaxy-integrations-python-api.readthedocs.io'>documentation</a>
## Features
Each integration in GOG Galaxy 2.0 comes as a separate Python script and is launched as a separate process that needs to communicate with the main instance of GOG Galaxy 2.0.
The provided features are:
- multistep authorization using a browser built into GOG Galaxy 2.0
- support for GOG Galaxy 2.0 features:
- importing owned and detecting installed games
- installing and launching games
- importing achievements and game time
- importing friends lists and statuses
- importing friends recommendations list
- receiving and sending chat messages
- cache storage
## Platform Id's
Each integration can implement only one platform. Each integration must declare which platform it's integrating.
[List of possible Platform IDs](PLATFORM_IDs.md)
## Basic usage
Each integration should inherit from the :class:`~galaxy.api.plugin.Plugin` class. Supported methods like :meth:`~galaxy.api.plugin.Plugin.get_owned_games` should be overwritten - they are called from the GOG Galaxy client at the appropriate times.
Each of those methods can raise exceptions inherited from the :exc:`~galaxy.api.jsonrpc.ApplicationError`.
Communication between an integration and the client is also possible with the use of notifications, for example: :meth:`~galaxy.api.plugin.Plugin.update_local_game_status`.
```python
import asyncio
from galaxy.api.plugin import Plugin
import sys
from galaxy.api.plugin import Plugin, create_and_run_plugin
from galaxy.api.consts import Platform
class PluginExample(Plugin):
def __init__(self, reader, writer, token):
super().__init__(
Platform.Generic, # Choose platform from available list
"0.1", # Version
reader,
writer,
token
)
# implement methods
async def authenticate(self, stored_credentials=None):
pass
def main():
create_and_run_plugin(PluginExample, sys.argv)
# run plugin event loop
if __name__ == "__main__":
asyncio.run(MockPlugin().run())
main()
```
Use [pyinstaller](https://www.pyinstaller.org/) to create plugin executbale.
## Deployment
## Development
The client has a built-in Python 3.7 interpreter, so integrations are delivered as Python modules.
In order to be found by GOG Galaxy 2.0 an integration folder should be placed in [lookup directory](#deploy-location). Beside all the Python files, the integration folder must contain [manifest.json](#deploy-manifest) and all third-party dependencies. See an [exemplary structure](#deploy-structure-example).
### Lookup directory
<a name="deploy-location"></a>
- Windows:
`%localappdata%\GOG.com\Galaxy\plugins\installed`
- macOS:
`~/Library/Application Support/GOG.com/Galaxy/plugins/installed`
### Manifest
<a name="deploy-manifest"></a>
Obligatory JSON file to be placed in an integration folder.
```json
{
"name": "Example plugin",
"platform": "generic",
"guid": "UNIQUE-GUID",
"version": "0.1",
"description": "Example plugin",
"author": "Name",
"email": "author@email.com",
"url": "https://github.com/user/galaxy-plugin-example",
"script": "plugin.py"
}
```
| property | description |
|---------------|---|
| `guid` | |
| `description` | |
| `url` | |
| `script` | path of the entry point module, relative to the integration folder |
### Dependencies
All third-party packages (packages not included in the Python 3.7 standard library) should be deployed along with plugin files. Use the following command structure:
```pip install DEP --target DIR --implementation cp --python-version 37```
For example, a plugin that uses *requests* could have the following structure:
<a name="deploy-structure-example"></a>
Install required packages:
```bash
pip install -r requirements.txt
installed
└── my_integration
   ├── galaxy
   │   └── api
   ├── requests
   │   └── ...
   ├── plugin.py
└── manifest.json
```
Run tests:
```bash
pytest
```
## Legal Notice
## Changelog
### 0.21
* Add `Epic` platform.
### 0.16
* Do not log sensitive data.
* Return `LocalGameState` as int (possible combination of flags).
### 0.15
* `shutdown()` is called on socket disconnection.
### 0.14
* Added required version parameter to Plugin constructor.
By integrating or attempting to integrate any applications or content with or into GOG Galaxy 2.0 you represent that such application or content is your original creation (other than any software made available by GOG) and/or that you have all necessary rights to grant such applicable rights to the relevant community integration to GOG and to GOG Galaxy 2.0 end users for the purpose of use of such community integration and that such community integration comply with any third party license and other requirements including compliance with applicable laws.

16
docs/make.py Normal file
View File

@@ -0,0 +1,16 @@
"""Builds documentation locally. Use for preview only"""
import pathlib
import subprocess
import webbrowser
source = pathlib.Path("docs", "source")
build = pathlib.Path("docs", "build")
master_doc = 'index.html'
subprocess.run(['sphinx-build', '-M', 'clean', str(source), str(build)])
subprocess.run(['sphinx-build', '-M', 'html', str(source), str(build)])
master_path = build / 'html' / master_doc
webbrowser.open(f'file://{master_path.resolve()}')

5
docs/requirements.txt Normal file
View File

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

75
docs/source/conf.py Normal file
View File

@@ -0,0 +1,75 @@
# Configuration file for the Sphinx documentation builder.
# Documentation:
# http://www.sphinx-doc.org/en/master/config
import os
import sys
import subprocess
# -- Path setup --------------------------------------------------------------
_ROOT = os.path.join('..', '..')
sys.path.append(os.path.abspath(os.path.join(_ROOT, 'src')))
# -- Project information -----------------------------------------------------
project = 'GOG Galaxy Integrations API'
copyright = '2019, GOG.com'
_author, _version = subprocess.check_output(
['python', os.path.join(_ROOT, 'setup.py'), '--author', '--version'],
universal_newlines=True).strip().split('\n')
author = _author
version = _version
release = _version
# -- General configuration ---------------------------------------------------
# Add any Sphinx extension module names here, as strings. They can be
# extensions coming with Sphinx (named 'sphinx.ext.*') or your custom
# ones.
extensions = [
'sphinx.ext.autodoc',
'sphinxcontrib.asyncio',
'sphinx_autodoc_typehints',
'm2r' # mdinclude directive for makrdown files
]
autodoc_member_order = 'bysource'
autodoc_inherit_docstrings = False
autodoc_mock_imports = ["aiohttp"]
set_type_checking_flag = True
# Add any paths that contain templates here, relative to this directory.
templates_path = ['_templates']
# List of patterns, relative to source directory, that match files and
# directories to ignore when looking for source files.
# This pattern also affects html_static_path and html_extra_path.
exclude_patterns = [] # type: ignore
# -- Options for HTML output -------------------------------------------------
# The theme to use for HTML and HTML Help pages. See the documentation for
# a list of builtin themes.
#
html_theme = "sphinx_rtd_theme"
html_theme_options = {
# 'canonical_url': '', # main page to be serach in google with trailing slash
'display_version': True,
'style_external_links': True,
# Toc options
'collapse_navigation': False,
'sticky_navigation': True,
'navigation_depth': 4,
}
# Add any paths that contain custom static files (such as style sheets) here,
# relative to this directory. They are copied after the builtin static files,
# so a file named "default.css" will overwrite the builtin "default.css".
html_static_path = ['_static']
master_doc = 'index'

View File

@@ -0,0 +1,40 @@
galaxy.api
=================
plugin
------------------------
.. automodule:: galaxy.api.plugin
:members:
:undoc-members:
:exclude-members: JSONEncoder, features, achievements_import_finished, game_times_import_finished, start_achievements_import, start_game_times_import, get_game_times, get_unlocked_achievements
types
-----------------------
.. automodule:: galaxy.api.types
:members:
:undoc-members:
consts
------------------------
.. automodule:: galaxy.api.consts
:members:
:undoc-members:
:show-inheritance:
:exclude-members: Feature
errors
------------------------
.. autoexception:: galaxy.api.jsonrpc.ApplicationError
:show-inheritance:
.. autoexception:: galaxy.api.jsonrpc.UnknownError
:show-inheritance:
.. automodule:: galaxy.api.errors
:members:
:undoc-members:
:show-inheritance:

View File

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

16
docs/source/index.rst Normal file
View File

@@ -0,0 +1,16 @@
GOG Galaxy Integrations Python API
=================================================
.. toctree::
:maxdepth: 2
:includehidden:
Overview <overview>
galaxy.api
galaxy.http
Platform ID's <platforms>
Index
-------------------
* :ref:`genindex`

15
docs/source/overview.rst Normal file
View File

@@ -0,0 +1,15 @@
.. mdinclude:: ../../README.md
:end-line: 4
.. excluding self-pointing documentation link
.. mdinclude:: ../../README.md
:start-line: 6
:end-line: 26
.. excluding Platforms Id's link
:ref:`platforms-link`
.. mdinclude:: ../../README.md
:start-line: 28

View File

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

2
mypy.ini Normal file
View File

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

View File

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

View File

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

View File

@@ -2,8 +2,8 @@ from setuptools import setup, find_packages
setup(
name="galaxy.plugin.api",
version="0.28.1",
description="Galaxy python plugin API",
version="0.48",
description="GOG Galaxy Integrations Python API",
author='Galaxy team',
author_email='galaxy@gog.com',
packages=find_packages("src"),

12
src/.readthedocs.yml Normal file
View File

@@ -0,0 +1,12 @@
version: 2
sphinx:
configuration: docs/source/conf.py
formats: all
python:
version: 3.7
install:
- requirements: requirements.txt
- requirements: docs/requirements.txt

View File

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

View File

@@ -1,6 +1,8 @@
from enum import Enum, Flag
class Platform(Enum):
"""Supported gaming platforms"""
Unknown = "unknown"
Gog = "gog"
Steam = "steam"
@@ -11,8 +13,79 @@ class Platform(Enum):
Uplay = "uplay"
Battlenet = "battlenet"
Epic = "epic"
Bethesda = "bethesda"
ParadoxPlaza = "paradox"
HumbleBundle = "humble"
Kartridge = "kartridge"
ItchIo = "itch"
NintendoSwitch = "nswitch"
NintendoWiiU = "nwiiu"
NintendoWii = "nwii"
NintendoGameCube = "ncube"
RiotGames = "riot"
Wargaming = "wargaming"
NintendoGameBoy = "ngameboy"
Atari = "atari"
Amiga = "amiga"
SuperNintendoEntertainmentSystem = "snes"
Beamdog = "beamdog"
Direct2Drive = "d2d"
Discord = "discord"
DotEmu = "dotemu"
GameHouse = "gamehouse"
GreenManGaming = "gmg"
WePlay = "weplay"
ZxSpectrum = "zx"
ColecoVision = "vision"
NintendoEntertainmentSystem = "nes"
SegaMasterSystem = "sms"
Commodore64 = "c64"
PcEngine = "pce"
SegaGenesis = "segag"
NeoGeo = "neo"
Sega32X = "sega32"
SegaCd = "segacd"
_3Do = "3do"
SegaSaturn = "saturn"
PlayStation = "psx"
PlayStation2 = "ps2"
Nintendo64 = "n64"
AtariJaguar = "jaguar"
SegaDreamcast = "dc"
Xbox = "xboxog"
Amazon = "amazon"
GamersGate = "gg"
Newegg = "egg"
BestBuy = "bb"
GameUk = "gameuk"
Fanatical = "fanatical"
PlayAsia = "playasia"
Stadia = "stadia"
Arc = "arc"
ElderScrollsOnline = "eso"
Glyph = "glyph"
AionLegionsOfWar = "aionl"
Aion = "aion"
BladeAndSoul = "blade"
GuildWars = "gw"
GuildWars2 = "gw2"
Lineage2 = "lin2"
FinalFantasy11 = "ffxi"
FinalFantasy14 = "ffxiv"
TotalWar = "totalwar"
WindowsStore = "winstore"
EliteDangerous = "elites"
StarCitizen = "star"
PlayStationPortable = "psp"
PlayStationVita = "psvita"
NintendoDs = "nds"
Nintendo3Ds = "3ds"
PathOfExile = "pathofexile"
class Feature(Enum):
"""Possible features that can be implemented by an integration.
It does not have to support all or any specific features from the list.
"""
Unknown = "Unknown"
ImportInstalledGames = "ImportInstalledGames"
ImportOwnedGames = "ImportOwnedGames"
@@ -25,20 +98,23 @@ class Feature(Enum):
ImportUsers = "ImportUsers"
VerifyGame = "VerifyGame"
ImportFriends = "ImportFriends"
ShutdownPlatformClient = "ShutdownPlatformClient"
LaunchPlatformClient = "LaunchPlatformClient"
class LicenseType(Enum):
"""Possible game license types, understandable for the GOG Galaxy client."""
Unknown = "Unknown"
SinglePurchase = "SinglePurchase"
FreeToPlay = "FreeToPlay"
OtherUserLicense = "OtherUserLicense"
class LocalGameState(Flag):
"""Possible states that a local game can be in.
For example a game which is both installed and currently running should have its state set as a "bitwise or" of Running and Installed flags:
``local_game_state=<LocalGameState.Running|Installed: 3>``
"""
None_ = 0
Installed = 1
Running = 2
class PresenceState(Enum):
Unknown = "Unknown"
Online = "online"
Offline = "offline"
Away = "away"

View File

@@ -1,6 +1,6 @@
from galaxy.api.jsonrpc import ApplicationError, UnknownError
UnknownError = UnknownError
assert UnknownError
class AuthenticationRequired(ApplicationError):
def __init__(self, data=None):
@@ -22,6 +22,10 @@ class UnknownBackendResponse(ApplicationError):
def __init__(self, data=None):
super().__init__(4, "Backend responded in uknown way", data)
class TooManyRequests(ApplicationError):
def __init__(self, data=None):
super().__init__(5, "Too many requests. Try again later", data)
class InvalidCredentials(ApplicationError):
def __init__(self, data=None):
super().__init__(100, "Invalid credentials", data)
@@ -50,18 +54,6 @@ class AccessDenied(ApplicationError):
def __init__(self, data=None):
super().__init__(106, "Access denied", data)
class ParentalControlBlock(ApplicationError):
def __init__(self, data=None):
super().__init__(107, "Parental control block", data)
class DeviceBlocked(ApplicationError):
def __init__(self, data=None):
super().__init__(108, "Device blocked", data)
class RegionBlocked(ApplicationError):
def __init__(self, data=None):
super().__init__(109, "Region blocked", data)
class FailedParsingManifest(ApplicationError):
def __init__(self, data=None):
super().__init__(200, "Failed parsing manifest", data)
@@ -77,3 +69,7 @@ class IncoherentLastMessage(ApplicationError):
class MessageNotFound(ApplicationError):
def __init__(self, data=None):
super().__init__(500, "Message not found", data)
class ImportInProgress(ApplicationError):
def __init__(self, data=None):
super().__init__(600, "Import already in progress", data)

View File

@@ -5,6 +5,9 @@ import logging
import inspect
import json
from galaxy.reader import StreamLineReader
from galaxy.task_manager import TaskManager
class JsonRpcError(Exception):
def __init__(self, code, message, data=None):
self.code = code
@@ -12,6 +15,9 @@ class JsonRpcError(Exception):
self.data = data
super().__init__()
def __eq__(self, other):
return self.code == other.code and self.message == other.message and self.data == other.data
class ParseError(JsonRpcError):
def __init__(self):
super().__init__(-32700, "Parse error")
@@ -47,56 +53,54 @@ class UnknownError(ApplicationError):
super().__init__(0, "Unknown error", data)
Request = namedtuple("Request", ["method", "params", "id"], defaults=[{}, None])
Method = namedtuple("Method", ["callback", "signature", "internal", "sensitive_params"])
Method = namedtuple("Method", ["callback", "signature", "immediate", "sensitive_params"])
def anonymise_sensitive_params(params, sensitive_params):
anomized_data = "****"
if not sensitive_params:
return params
if isinstance(sensitive_params, bool):
if sensitive_params:
return {k:anomized_data for k,v in params.items()}
if isinstance(sensitive_params, Iterable):
anomized_params = params.copy()
for key in anomized_params.keys():
if key in sensitive_params:
anomized_params[key] = anomized_data
return anomized_params
return {k: anomized_data if k in sensitive_params else v for k, v in params.items()}
return anomized_data
return params
class Server():
def __init__(self, reader, writer, encoder=json.JSONEncoder()):
self._active = True
self._reader = reader
self._reader = StreamLineReader(reader)
self._writer = writer
self._encoder = encoder
self._methods = {}
self._notifications = {}
self._eof_listeners = []
self._task_manager = TaskManager("jsonrpc server")
def register_method(self, name, callback, internal, sensitive_params=False):
def register_method(self, name, callback, immediate, sensitive_params=False):
"""
Register method
:param name:
:param callback:
:param internal: if True the callback will be processed immediately (synchronously)
:param sensitive_params: list of parameters that will by anonymized before logging; if False - no params
are considered sensitive, if True - all params are considered sensitive
:param sensitive_params: list of parameters that are anonymized before logging; \
if False - no params are considered sensitive, if True - all params are considered sensitive
"""
self._methods[name] = Method(callback, inspect.signature(callback), internal, sensitive_params)
self._methods[name] = Method(callback, inspect.signature(callback), immediate, sensitive_params)
def register_notification(self, name, callback, internal, sensitive_params=False):
def register_notification(self, name, callback, immediate, sensitive_params=False):
"""
Register notification
:param name:
:param callback:
:param internal: if True the callback will be processed immediately (synchronously)
:param sensitive_params: list of parameters that will by anonymized before logging; if False - no params
are considered sensitive, if True - all params are considered sensitive
:param sensitive_params: list of parameters that are anonymized before logging; \
if False - no params are considered sensitive, if True - all params are considered sensitive
"""
self._notifications[name] = Method(callback, inspect.signature(callback), internal, sensitive_params)
def register_eof(self, callback):
self._eof_listeners.append(callback)
self._notifications[name] = Method(callback, inspect.signature(callback), immediate, sensitive_params)
async def run(self):
while self._active:
@@ -111,15 +115,18 @@ class Server():
data = data.strip()
logging.debug("Received %d bytes of data", len(data))
self._handle_input(data)
await asyncio.sleep(0) # To not starve task queue
def stop(self):
def close(self):
logging.info("Closing JSON-RPC server - not more messages will be read")
self._active = False
async def wait_closed(self):
await self._task_manager.wait()
def _eof(self):
logging.info("Received EOF")
self.stop()
for listener in self._eof_listeners:
listener()
self.close()
def _handle_input(self, data):
try:
@@ -139,7 +146,7 @@ class Server():
logging.error("Received unknown notification: %s", request.method)
return
callback, signature, internal, sensitive_params = method
callback, signature, immediate, sensitive_params = method
self._log_request(request, sensitive_params)
try:
@@ -147,12 +154,11 @@ class Server():
except TypeError:
self._send_error(request.id, InvalidParams())
if internal:
# internal requests are handled immediately
if immediate:
callback(*bound_args.args, **bound_args.kwargs)
else:
try:
asyncio.create_task(callback(*bound_args.args, **bound_args.kwargs))
self._task_manager.create_task(callback(*bound_args.args, **bound_args.kwargs), request.method)
except Exception:
logging.exception("Unexpected exception raised in notification handler")
@@ -163,7 +169,7 @@ class Server():
self._send_error(request.id, MethodNotFound())
return
callback, signature, internal, sensitive_params = method
callback, signature, immediate, sensitive_params = method
self._log_request(request, sensitive_params)
try:
@@ -171,8 +177,7 @@ class Server():
except TypeError:
self._send_error(request.id, InvalidParams())
if internal:
# internal requests are handled immediately
if immediate:
response = callback(*bound_args.args, **bound_args.kwargs)
self._send_response(request.id, response)
else:
@@ -184,11 +189,13 @@ class Server():
self._send_error(request.id, MethodNotFound())
except JsonRpcError as error:
self._send_error(request.id, error)
except Exception as e: #pylint: disable=broad-except
except asyncio.CancelledError:
self._send_error(request.id, Aborted())
except Exception as e: #pylint: disable=broad-except
logging.exception("Unexpected exception raised in plugin handler")
self._send_error(request.id, UnknownError(str(e)))
asyncio.create_task(handle())
self._task_manager.create_task(handle(), request.method)
@staticmethod
def _parse_request(data):
@@ -209,7 +216,7 @@ class Server():
logging.debug("Sending data: %s", line)
data = (line + "\n").encode("utf-8")
self._writer.write(data)
asyncio.create_task(self._writer.drain())
self._task_manager.create_task(self._writer.drain(), "drain")
except TypeError as error:
logging.error(str(error))
@@ -249,14 +256,16 @@ class NotificationClient():
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 will by anonymized before logging; if False - no params
are considered sensitive, if True - all params are considered sensitive
: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",
@@ -266,13 +275,16 @@ class NotificationClient():
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)
asyncio.create_task(self._writer.drain())
self._task_manager.create_task(self._writer.drain(), "drain")
except TypeError as error:
logging.error("Failed to parse outgoing message: %s", str(error))

View File

@@ -1,33 +1,40 @@
import asyncio
import dataclasses
import json
import logging
import logging.handlers
import dataclasses
from enum import Enum
from collections import OrderedDict
import sys
from enum import Enum
from typing import Any, Dict, List, Optional, Set, Union
from galaxy.api.jsonrpc import Server, NotificationClient
from galaxy.api.consts import Feature
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
from galaxy.task_manager import TaskManager
class JSONEncoder(json.JSONEncoder):
def default(self, o): # pylint: disable=method-hidden
def default(self, o): # pylint: disable=method-hidden
if dataclasses.is_dataclass(o):
# filter None values
def dict_factory(elements):
return {k: v for k, v in elements if v is not None}
return dataclasses.asdict(o, dict_factory=dict_factory)
if isinstance(o, Enum):
return o.value
return super().default(o)
class Plugin():
class Plugin:
"""Use and override methods of this class to create a new platform integration."""
def __init__(self, platform, version, reader, writer, handshake_token):
logging.info("Creating plugin for platform %s, version %s", platform.value, version)
self._platform = platform
self._version = version
self._feature_methods = OrderedDict()
self._features: Set[Feature] = set()
self._active = True
self._reader, self._writer = reader, writer
@@ -37,153 +44,171 @@ class Plugin():
self._server = Server(self._reader, self._writer, encoder)
self._notification_client = NotificationClient(self._writer, encoder)
def eof_handler():
self._shutdown()
self._server.register_eof(eof_handler)
self._achievements_import_in_progress = False
self._game_times_import_in_progress = False
self._persistent_cache = dict()
self._internal_task_manager = TaskManager("plugin internal")
self._external_task_manager = TaskManager("plugin external")
# internal
self._register_method("shutdown", self._shutdown, internal=True)
self._register_method("get_capabilities", self._get_capabilities, internal=True)
self._register_method("ping", self._ping, internal=True)
self._register_method("get_capabilities", self._get_capabilities, internal=True, immediate=True)
self._register_method(
"initialize_cache",
self._initialize_cache,
internal=True,
immediate=True,
sensitive_params="data"
)
self._register_method("ping", self._ping, internal=True, immediate=True)
# implemented by developer
self._register_method("init_authentication", self.authenticate, sensitive_params=["stored_credentials"])
self._register_method("pass_login_credentials", self.pass_login_credentials)
self._register_method(
"init_authentication",
self.authenticate,
sensitive_params=["stored_credentials"]
)
self._register_method(
"pass_login_credentials",
self.pass_login_credentials,
sensitive_params=["cookies", "credentials"]
)
self._register_method(
"import_owned_games",
self.get_owned_games,
result_name="owned_games",
feature=Feature.ImportOwnedGames
)
self._register_method(
"import_unlocked_achievements",
self.get_unlocked_achievements,
result_name="unlocked_achievements",
feature=Feature.ImportAchievements
)
self._register_method(
"import_local_games",
self.get_local_games,
result_name="local_games",
feature=Feature.ImportInstalledGames
)
self._register_notification("launch_game", self.launch_game, feature=Feature.LaunchGame)
self._register_notification("install_game", self.install_game, feature=Feature.InstallGame)
self._register_notification(
"uninstall_game",
self.uninstall_game,
feature=Feature.UninstallGame
)
self._register_method(
"import_friends",
self.get_friends,
result_name="friend_info_list",
feature=Feature.ImportFriends
)
self._register_method(
"import_user_infos",
self.get_users,
result_name="user_info_list",
feature=Feature.ImportUsers
)
self._register_method(
"send_message",
self.send_message,
feature=Feature.Chat
)
self._register_method(
"mark_as_read",
self.mark_as_read,
feature=Feature.Chat
)
self._register_method(
"import_rooms",
self.get_rooms,
result_name="rooms",
feature=Feature.Chat
)
self._register_method(
"import_room_history_from_message",
self.get_room_history_from_message,
result_name="messages",
feature=Feature.Chat
)
self._register_method(
"import_room_history_from_timestamp",
self.get_room_history_from_timestamp,
result_name="messages",
feature=Feature.Chat
result_name="owned_games"
)
self._detect_feature(Feature.ImportOwnedGames, ["get_owned_games"])
self._register_method(
"import_game_times",
self.get_game_times,
result_name="game_times",
feature=Feature.ImportGameTime
)
self._register_method("start_achievements_import", self._start_achievements_import)
self._detect_feature(Feature.ImportAchievements, ["get_unlocked_achievements"])
self._register_method("import_local_games", self.get_local_games, result_name="local_games")
self._detect_feature(Feature.ImportInstalledGames, ["get_local_games"])
self._register_notification("launch_game", self.launch_game)
self._detect_feature(Feature.LaunchGame, ["launch_game"])
self._register_notification("install_game", self.install_game)
self._detect_feature(Feature.InstallGame, ["install_game"])
self._register_notification("uninstall_game", self.uninstall_game)
self._detect_feature(Feature.UninstallGame, ["uninstall_game"])
self._register_notification("shutdown_platform_client", self.shutdown_platform_client)
self._detect_feature(Feature.ShutdownPlatformClient, ["shutdown_platform_client"])
self._register_notification("launch_platform_client", self.launch_platform_client)
self._detect_feature(Feature.LaunchPlatformClient, ["launch_platform_client"])
self._register_method("import_friends", self.get_friends, result_name="friend_info_list")
self._detect_feature(Feature.ImportFriends, ["get_friends"])
self._register_method("start_game_times_import", self._start_game_times_import)
self._detect_feature(Feature.ImportGameTime, ["get_game_time"])
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, tb):
self.close()
await self.wait_closed()
@property
def features(self):
features = []
if self.__class__ != Plugin:
for feature, handlers in self._feature_methods.items():
if self._implements(handlers):
features.append(feature)
def features(self) -> List[Feature]:
return list(self._features)
return features
@property
def persistent_cache(self) -> Dict[str, str]:
"""The cache is only available after the :meth:`~.handshake_complete()` is called.
"""
return self._persistent_cache
def _implements(self, handlers):
for handler in handlers:
if handler.__name__ not in self.__class__.__dict__:
def _implements(self, methods: List[str]) -> bool:
for method in methods:
if method not in self.__class__.__dict__:
return False
return True
def _register_method(self, name, handler, result_name=None, internal=False, sensitive_params=False, feature=None):
if internal:
def _detect_feature(self, feature: Feature, methods: List[str]):
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 wrap_result(result):
if result_name:
result = {
result_name: result
}
return result
if immediate:
def method(*args, **kwargs):
result = handler(*args, **kwargs)
if result_name:
result = {
result_name: result
}
return result
return wrap_result(result)
self._server.register_method(name, method, True, sensitive_params)
else:
async def method(*args, **kwargs):
result = await handler(*args, **kwargs)
if result_name:
result = {
result_name: result
}
return result
if not internal:
handler_ = self._wrap_external_method(handler, name)
else:
handler_ = handler
result = await handler_(*args, **kwargs)
return wrap_result(result)
self._server.register_method(name, method, False, sensitive_params)
if feature is not None:
self._feature_methods.setdefault(feature, []).append(handler)
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)
def _register_notification(self, name, handler, internal=False, sensitive_params=False, feature=None):
self._server.register_notification(name, handler, internal, sensitive_params)
if feature is not None:
self._feature_methods.setdefault(feature, []).append(handler)
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 main coorutine"""
async def pass_control():
while self._active:
try:
self.tick()
except Exception:
logging.exception("Unexpected exception raised in plugin tick")
await asyncio.sleep(1)
"""Plugin's main coroutine."""
await self._server.run()
await self._external_task_manager.wait()
await asyncio.gather(pass_control(), self._server.run())
def close(self) -> None:
if not self._active:
return
def _shutdown(self):
logging.info("Shuting down")
self._server.stop()
logging.info("Closing plugin")
self._server.close()
self._external_task_manager.cancel()
self._internal_task_manager.create_task(self.shutdown(), "shutdown")
self._active = False
self.shutdown()
async def wait_closed(self) -> None:
await self._external_task_manager.wait()
await self._internal_task_manager.wait()
await self._server.wait_closed()
await self._notification_client.close()
def create_task(self, coro, description):
"""Wrapper around asyncio.create_task - takes care of canceling tasks on shutdown"""
return self._external_task_manager.create_task(coro, description)
async def _pass_control(self):
while self._active:
try:
self.tick()
except Exception:
logging.exception("Unexpected exception raised in plugin tick")
await asyncio.sleep(1)
async def _shutdown(self):
logging.info("Shutting down")
self.close()
await self._external_task_manager.wait()
await self._internal_task_manager.wait()
def _get_capabilities(self):
return {
@@ -192,130 +217,564 @@ class Plugin():
"token": self._handshake_token
}
def _initialize_cache(self, data: Dict):
self._persistent_cache = data
try:
self.handshake_complete()
except Exception:
logging.exception("Unhandled exception during `handshake_complete` step")
self._internal_task_manager.create_task(self._pass_control(), "tick")
@staticmethod
def _ping():
pass
# notifications
def store_credentials(self, credentials):
"""Notify client to store plugin credentials.
They will be pass to next authencicate calls.
def store_credentials(self, credentials: Dict[str, Any]) -> None:
"""Notify the client to store authentication credentials.
Credentials are passed on the next authenticate call.
:param credentials: credentials that client will store; they are stored locally on a user pc
Example use case of store_credentials:
.. code-block:: python
:linenos:
async def pass_login_credentials(self, step, credentials, cookies):
if self.got_everything(credentials,cookies):
user_data = await self.parse_credentials(credentials,cookies)
else:
next_params = self.get_next_params(credentials,cookies)
next_cookies = self.get_next_cookies(credentials,cookies)
return NextStep("web_session", next_params, cookies=next_cookies)
self.store_credentials(user_data['credentials'])
return Authentication(user_data['userId'], user_data['username'])
"""
# temporary solution for persistent_cache vs credentials issue
self.persistent_cache['credentials'] = credentials # type: ignore
self._notification_client.notify("store_credentials", credentials, sensitive_params=True)
def add_game(self, game):
params = {"owned_game" : game}
def add_game(self, game: Game) -> None:
"""Notify the client to add game to the list of owned games
of the currently authenticated user.
:param game: Game to add to the list of owned games
Example use case of add_game:
.. code-block:: python
:linenos:
async def check_for_new_games(self):
games = await self.get_owned_games()
for game in games:
if game not in self.owned_games_cache:
self.owned_games_cache.append(game)
self.add_game(game)
"""
params = {"owned_game": game}
self._notification_client.notify("owned_game_added", params)
def remove_game(self, game_id):
params = {"game_id" : game_id}
def remove_game(self, game_id: str) -> None:
"""Notify the client to remove game from the list of owned games
of the currently authenticated user.
:param game_id: the id of the game to remove from the list of owned games
Example use case of remove_game:
.. code-block:: python
:linenos:
async def check_for_removed_games(self):
games = await self.get_owned_games()
for game in self.owned_games_cache:
if game not in games:
self.owned_games_cache.remove(game)
self.remove_game(game.game_id)
"""
params = {"game_id": game_id}
self._notification_client.notify("owned_game_removed", params)
def update_game(self, game):
params = {"owned_game" : game}
def update_game(self, game: Game) -> None:
"""Notify the client to update the status of a game
owned by the currently authenticated user.
:param game: Game to update
"""
params = {"owned_game": game}
self._notification_client.notify("owned_game_updated", params)
def unlock_achievement(self, game_id, achievement):
def unlock_achievement(self, game_id: str, achievement: Achievement) -> None:
"""Notify the client to unlock an achievement for a specific game.
:param game_id: the id of the game for which to unlock an achievement.
:param achievement: achievement to unlock.
"""
params = {
"game_id": game_id,
"achievement": achievement
}
self._notification_client.notify("achievement_unlocked", params)
def update_local_game_status(self, local_game):
params = {"local_game" : local_game}
def _game_achievements_import_success(self, game_id: str, achievements: List[Achievement]) -> None:
params = {
"game_id": game_id,
"unlocked_achievements": achievements
}
self._notification_client.notify("game_achievements_import_success", params)
def _game_achievements_import_failure(self, game_id: str, error: ApplicationError) -> None:
params = {
"game_id": game_id,
"error": {
"code": error.code,
"message": error.message
}
}
self._notification_client.notify("game_achievements_import_failure", params)
def _achievements_import_finished(self) -> None:
self._notification_client.notify("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.
:param local_game: the LocalGame to update
Example use case triggered by the :meth:`.tick` method:
.. code-block:: python
:linenos:
:emphasize-lines: 5
async def _check_statuses(self):
for game in await self._get_local_games():
if game.status == self._cached_game_statuses.get(game.id):
continue
self.update_local_game_status(LocalGame(game.id, game.status))
self._cached_games_statuses[game.id] = game.status
await asyncio.sleep(5) # interval
def tick(self):
if self._check_statuses_task is None or self._check_statuses_task.done():
self._check_statuses_task = asyncio.create_task(self._check_statuses())
"""
params = {"local_game": local_game}
self._notification_client.notify("local_game_status_changed", params)
def add_friend(self, user):
params = {"friend_info" : user}
def add_friend(self, user: FriendInfo) -> 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
"""
params = {"friend_info": user}
self._notification_client.notify("friend_added", params)
def remove_friend(self, user_id):
params = {"user_id" : user_id}
def remove_friend(self, user_id: str) -> None:
"""Notify the client to remove a user from friends list of the currently authenticated user.
:param user_id: id of the user to remove from friends list
"""
params = {"user_id": user_id}
self._notification_client.notify("friend_removed", params)
def update_room(self, room_id, unread_message_count=None, new_messages=None):
params = {"room_id": room_id}
if unread_message_count is not None:
params["unread_message_count"] = unread_message_count
if new_messages is not None:
params["messages"] = new_messages
self._notification_client.notify("chat_room_updated", params)
def update_game_time(self, game_time: GameTime) -> None:
"""Notify the client to update game time for a game.
def update_game_time(self, game_time):
params = {"game_time" : game_time}
:param game_time: game time to update
"""
params = {"game_time": game_time}
self._notification_client.notify("game_time_updated", params)
def lost_authentication(self):
def _game_time_import_success(self, game_time: GameTime) -> None:
params = {"game_time": game_time}
self._notification_client.notify("game_time_import_success", params)
def _game_time_import_failure(self, game_id: str, error: ApplicationError) -> None:
params = {
"game_id": game_id,
"error": {
"code": error.code,
"message": error.message
}
}
self._notification_client.notify("game_time_import_failure", params)
def _game_times_import_finished(self) -> None:
self._notification_client.notify("game_times_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)
def push_cache(self) -> None:
"""Push local copy of the persistent cache to the GOG Galaxy Client replacing existing one.
"""
self._notification_client.notify(
"push_cache",
params={"data": self._persistent_cache},
sensitive_params="data"
)
# handlers
def tick(self):
"""This method is called periodicaly.
Override it to implement periodical tasks like refreshing cache.
This method should not be blocking - any longer actions should be
handled by asycio tasks.
def handshake_complete(self) -> None:
"""This method is called right after the handshake with the GOG Galaxy Client is complete and
before any other operations are called by the GOG Galaxy Client.
Persistent cache is available when this method is called.
Override it if you need to do additional plugin initializations.
This method is called internally."""
def tick(self) -> None:
"""This method is called periodically.
Override it to implement periodical non-blocking tasks.
This method is called internally.
Example of possible override of the method:
.. code-block:: python
:linenos:
def tick(self):
if not self.checking_for_new_games:
asyncio.create_task(self.check_for_new_games())
if not self.checking_for_removed_games:
asyncio.create_task(self.check_for_removed_games())
if not self.updating_game_statuses:
asyncio.create_task(self.update_game_statuses())
"""
def shutdown(self):
"""This method is called on plugin shutdown.
async def shutdown(self) -> None:
"""This method is called on integration shutdown.
Override it to implement tear down.
"""
This method is called by the GOG Galaxy Client."""
# methods
async def authenticate(self, stored_credentials=None):
"""Overide this method to handle plugin authentication.
The method should return galaxy.api.types.Authentication
or raise galaxy.api.types.LoginError on authentication failure.
async def authenticate(self, stored_credentials: Optional[Dict] = None) -> Union[NextStep, Authentication]:
"""Override this method to handle user authentication.
This method should either return :class:`~galaxy.api.types.Authentication` if the authentication is finished
or :class:`~galaxy.api.types.NextStep` if it requires going to another url.
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.
Example of possible override of the method:
.. code-block:: python
:linenos:
async def authenticate(self, stored_credentials=None):
if not stored_credentials:
return NextStep("web_session", PARAMS, cookies=COOKIES)
else:
try:
user_data = self._authenticate(stored_credentials)
except AccessDenied:
raise InvalidCredentials()
return Authentication(user_data['userId'], user_data['username'])
"""
raise NotImplementedError()
async def pass_login_credentials(self, step, credentials, cookies):
async def pass_login_credentials(self, step: str, credentials: Dict[str, str], cookies: List[Dict[str, str]]) \
-> Union[NextStep, Authentication]:
"""This method is called if we return galaxy.api.types.NextStep from authenticate or from pass_login_credentials.
This method's parameters provide the data extracted from the web page navigation that previous NextStep finished on.
This method should either return galaxy.api.types.Authentication if the authentication is finished
or galaxy.api.types.NextStep if it requires going to another cef url.
This method is called by the GOG Galaxy Client.
:param step: deprecated.
:param credentials: end_uri previous NextStep finished on.
:param cookies: cookies extracted from the end_uri site.
Example of possible override of the method:
.. code-block:: python
:linenos:
async def pass_login_credentials(self, step, credentials, cookies):
if self.got_everything(credentials,cookies):
user_data = await self.parse_credentials(credentials,cookies)
else:
next_params = self.get_next_params(credentials,cookies)
next_cookies = self.get_next_cookies(credentials,cookies)
return NextStep("web_session", next_params, cookies=next_cookies)
self.store_credentials(user_data['credentials'])
return Authentication(user_data['userId'], user_data['username'])
"""
raise NotImplementedError()
async def get_owned_games(self):
async def get_owned_games(self) -> List[Game]:
"""Override this method to return owned games for currently logged in user.
This method is called by the GOG Galaxy Client.
Example of possible override of the method:
.. code-block:: python
:linenos:
async def get_owned_games(self):
if not self.authenticated():
raise AuthenticationRequired()
games = self.retrieve_owned_games()
return games
"""
raise NotImplementedError()
async def get_unlocked_achievements(self, game_id):
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:
logging.exception("Unexpected exception raised in import_game_achievements")
self._game_achievements_import_failure(game_id, UnknownError())
async def import_games_achievements(game_ids_, context_):
try:
imports = [import_game_achievements(game_id, context_) for game_id in game_ids_]
await asyncio.gather(*imports)
finally:
self._achievements_import_finished()
self._achievements_import_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
async def prepare_achievements_context(self, game_ids: List[str]) -> Any:
"""Override this method to prepare context for get_unlocked_achievements.
This allows for optimizations like batch requests to platform API.
Default implementation returns None.
:param game_ids: the ids of the games for which achievements are imported
:return: context
"""
return None
async def get_unlocked_achievements(self, game_id: str, context: Any) -> List[Achievement]:
"""Override this method to return list of unlocked achievements
for the game identified by the provided game_id.
This method is called by import task initialized by GOG Galaxy Client.
:param game_id: the id of the game for which the achievements are returned
:param context: the value returned from :meth:`prepare_achievements_context`
:return: list of Achievement objects
"""
raise NotImplementedError()
async def get_local_games(self):
def achievements_import_complete(self):
"""Override this method to handle operations after achievements import is finished
(like updating cache).
"""
async def get_local_games(self) -> List[LocalGame]:
"""Override this method to return the list of
games present locally on the users pc.
This method is called by the GOG Galaxy Client.
Example of possible override of the method:
.. code-block:: python
:linenos:
async def get_local_games(self):
local_games = []
for game in self.games_present_on_user_pc:
local_game = LocalGame()
local_game.game_id = game.id
local_game.local_game_state = game.get_installation_status()
local_games.append(local_game)
return local_games
"""
raise NotImplementedError()
async def launch_game(self, game_id):
async def launch_game(self, game_id: str) -> None:
"""Override this method to launch the game
identified by the provided game_id.
This method is called by the GOG Galaxy Client.
:param str game_id: the id of the game to launch
Example of possible override of the method:
.. code-block:: python
:linenos:
async def launch_game(self, game_id):
await self.open_uri(f"start client://launchgame/{game_id}")
"""
raise NotImplementedError()
async def install_game(self, game_id):
async def install_game(self, game_id: str) -> None:
"""Override this method to install the game
identified by the provided game_id.
This method is called by the GOG Galaxy Client.
:param str game_id: the id of the game to install
Example of possible override of the method:
.. code-block:: python
:linenos:
async def install_game(self, game_id):
await self.open_uri(f"start client://installgame/{game_id}")
"""
raise NotImplementedError()
async def uninstall_game(self, game_id):
async def uninstall_game(self, game_id: str) -> None:
"""Override this method to uninstall the game
identified by the provided game_id.
This method is called by the GOG Galaxy Client.
:param str game_id: the id of the game to uninstall
Example of possible override of the method:
.. code-block:: python
:linenos:
async def uninstall_game(self, game_id):
await self.open_uri(f"start client://uninstallgame/{game_id}")
"""
raise NotImplementedError()
async def get_friends(self):
async def shutdown_platform_client(self) -> None:
"""Override this method to gracefully terminate platform client.
This method is called by the GOG Galaxy Client."""
raise NotImplementedError()
async def get_users(self, user_id_list):
async def launch_platform_client(self) -> None:
"""Override this method to launch platform client. Preferably minimized to tray.
This method is called by the GOG Galaxy Client."""
raise NotImplementedError()
async def send_message(self, room_id, message_text):
async def get_friends(self) -> List[FriendInfo]:
"""Override this method to return the friends list
of the currently authenticated user.
This method is called by the GOG Galaxy Client.
Example of possible override of the method:
.. code-block:: python
:linenos:
async def get_friends(self):
if not self._http_client.is_authenticated():
raise AuthenticationRequired()
friends = self.retrieve_friends()
return friends
"""
raise NotImplementedError()
async def mark_as_read(self, room_id, last_message_id):
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:
logging.exception("Unexpected exception raised in import_game_time")
self._game_time_import_failure(game_id, UnknownError())
async def import_game_times(game_ids_, context_):
try:
imports = [import_game_time(game_id, context_) for game_id in game_ids_]
await asyncio.gather(*imports)
finally:
self._game_times_import_finished()
self._game_times_import_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
async def prepare_game_times_context(self, game_ids: List[str]) -> Any:
"""Override this method to prepare context for get_game_time.
This allows for optimizations like batch requests to platform API.
Default implementation returns None.
:param game_ids: the ids of the games for which game time are imported
:return: context
"""
return None
async def get_game_time(self, game_id: str, context: Any) -> GameTime:
"""Override this method to return the game time for the game
identified by the provided game_id.
This method is called by import task initialized by GOG Galaxy Client.
:param game_id: the id of the game for which the game time is returned
:param context: the value returned from :meth:`prepare_game_times_context`
:return: GameTime object
"""
raise NotImplementedError()
async def get_rooms(self):
raise NotImplementedError()
def game_times_import_complete(self) -> None:
"""Override this method to handle operations after game times import is finished
(like updating cache).
"""
async def get_room_history_from_message(self, room_id, message_id):
raise NotImplementedError()
async def get_room_history_from_timestamp(self, room_id, from_timestamp):
raise NotImplementedError()
async def get_game_times(self):
raise NotImplementedError()
def create_and_run_plugin(plugin_class, argv):
"""Call this method as an entry point for the implemented integration.
:param plugin_class: your plugin class.
:param argv: command line arguments with which the script was started.
Example of possible use of the method:
.. code-block:: python
:linenos:
def main():
create_and_run_plugin(PlatformPlugin, sys.argv)
if __name__ == "__main__":
main()
"""
if len(argv) < 3:
logging.critical("Not enough parameters, required: token, port")
sys.exit(1)
@@ -338,10 +797,15 @@ def create_and_run_plugin(plugin_class, argv):
async def coroutine():
reader, writer = await asyncio.open_connection("127.0.0.1", port)
plugin = plugin_class(reader, writer, token)
await plugin.run()
extra_info = writer.get_extra_info("sockname")
logging.info("Using local address: %s:%u", *extra_info)
async with plugin_class(reader, writer, token) as plugin:
await plugin.run()
try:
if sys.platform == "win32":
asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
asyncio.run(coroutine())
except Exception:
logging.exception("Error while running plugin")

View File

@@ -1,15 +1,28 @@
from dataclasses import dataclass
from typing import List, Dict, Optional
from galaxy.api.consts import LicenseType, LocalGameState, PresenceState
from galaxy.api.consts import LicenseType, LocalGameState
@dataclass
class Authentication():
"""Return this from :meth:`.authenticate` or :meth:`.pass_login_credentials`
to inform the client that authentication has successfully finished.
:param user_id: id of the authenticated user
:param user_name: username of the authenticated user
"""
user_id: str
user_name: str
@dataclass
class Cookie():
"""Cookie
:param name: name of the cookie
:param value: value of the cookie
:param domain: optional domain of the cookie
:param path: optional path of the cookie
"""
name: str
value: str
domain: Optional[str] = None
@@ -17,6 +30,38 @@ class Cookie():
@dataclass
class NextStep():
"""Return this from :meth:`.authenticate` or :meth:`.pass_login_credentials` to open client built-in browser with given url.
For example:
.. code-block:: python
:linenos:
PARAMS = {
"window_title": "Login to platform",
"window_width": 800,
"window_height": 600,
"start_uri": URL,
"end_uri_regex": r"^https://platform_website\.com/.*"
}
JS = {r"^https://platform_website\.com/.*": [
r'''
location.reload();
'''
]}
COOKIES = [Cookie("Cookie1", "ok", ".platform.com"),
Cookie("Cookie2", "ok", ".platform.com")
]
async def authenticate(self, stored_credentials=None):
if not stored_credentials:
return NextStep("web_session", PARAMS, cookies=COOKIES, js=JS)
:param auth_params: configuration options: {"window_title": :class:`str`, "window_width": :class:`str`, "window_height": :class:`int`, "start_uri": :class:`int`, "end_uri_regex": :class:`str`}
:param cookies: browser initial set of cookies
:param js: a map of the url regex patterns into the list of *js* scripts that should be executed on every document at given step of internal browser authentication.
"""
next_step: str
auth_params: Dict[str, str]
cookies: Optional[List[Cookie]] = None
@@ -24,17 +69,35 @@ class NextStep():
@dataclass
class LicenseInfo():
"""Information about the license of related product.
:param license_type: type of license
:param owner: optional owner of the related product, defaults to currently authenticated user
"""
license_type: LicenseType
owner: Optional[str] = None
@dataclass
class Dlc():
"""Downloadable content object.
:param dlc_id: id of the dlc
:param dlc_title: title of the dlc
:param license_info: information about the license attached to the dlc
"""
dlc_id: str
dlc_title: str
license_info: LicenseInfo
@dataclass
class Game():
"""Game object.
:param game_id: unique identifier of the game, this will be passed as parameter for methods such as launch_game
:param game_title: title of the game
:param dlcs: list of dlcs available for the game
:param license_info: information about the license attached to the game
"""
game_id: str
game_title: str
dlcs: Optional[List[Dlc]]
@@ -42,6 +105,12 @@ class Game():
@dataclass
class Achievement():
"""Achievement, has to be initialized with either id or name.
:param unlock_time: unlock time of the achievement
:param achievement_id: optional id of the achievement
:param achievement_name: optional name of the achievement
"""
unlock_time: int
achievement_id: Optional[str] = None
achievement_name: Optional[str] = None
@@ -52,43 +121,33 @@ class Achievement():
@dataclass
class LocalGame():
"""Game locally present on the authenticated user's computer.
:param game_id: id of the game
:param local_game_state: state of the game
"""
game_id: str
local_game_state: LocalGameState
@dataclass
class Presence():
presence_state: PresenceState
game_id: Optional[str] = None
presence_status: Optional[str] = None
@dataclass
class UserInfo():
user_id: str
is_friend: bool
user_name: str
avatar_url: str
presence: Presence
@dataclass
class FriendInfo():
"""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 Room():
room_id: str
unread_message_count: int
last_message_id: str
@dataclass
class Message():
message_id: str
sender_id: str
sent_time: int
message_text: str
@dataclass
class GameTime():
"""Game time of a game, defines the total time spent in the game
and the last time the game was played.
:param game_id: id of the related game
:param time_played: the total time spent in the game in **minutes**
:param last_time_played: last time the game was played (**unix timestamp**)
"""
game_id: str
time_played: int
last_played_time: int
time_played: Optional[int]
last_played_time: Optional[int]

View File

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

88
src/galaxy/proc_tools.py Normal file
View File

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

28
src/galaxy/reader.py Normal file
View File

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

View File

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

View File

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

View File

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

View File

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

View File

@@ -6,61 +6,60 @@ import pytest
from galaxy.api.plugin import Plugin
from galaxy.api.consts import Platform
from galaxy.unittest.mock import AsyncMock, coroutine_mock
from galaxy.unittest.mock import async_return_value
@pytest.fixture()
def reader():
stream = MagicMock(name="stream_reader")
stream.readline = AsyncMock()
stream.read = MagicMock()
yield stream
@pytest.fixture()
def writer():
async def writer():
stream = MagicMock(name="stream_writer")
stream.write = MagicMock()
stream.drain = AsyncMock()
stream.drain.side_effect = lambda: async_return_value(None)
yield stream
@pytest.fixture()
def readline(reader):
yield reader.readline
def read(reader):
yield reader.read
@pytest.fixture()
def write(writer):
yield writer.write
@pytest.fixture()
def plugin(reader, writer):
async def plugin(reader, writer):
"""Return plugin instance with all feature methods mocked"""
async_methods = (
methods = (
"handshake_complete",
"authenticate",
"get_owned_games",
"prepare_achievements_context",
"get_unlocked_achievements",
"achievements_import_complete",
"get_local_games",
"launch_game",
"launch_platform_client",
"install_game",
"uninstall_game",
"get_friends",
"get_users",
"send_message",
"mark_as_read",
"get_rooms",
"get_room_history_from_message",
"get_room_history_from_timestamp",
"get_game_times"
)
methods = (
"get_game_time",
"prepare_game_times_context",
"game_times_import_complete",
"shutdown_platform_client",
"shutdown",
"tick"
)
with ExitStack() as stack:
for method in async_methods:
stack.enter_context(patch.object(Plugin, method, new_callable=coroutine_mock))
for method in methods:
stack.enter_context(patch.object(Plugin, method))
yield Plugin(Platform.Generic, "0.1", reader, writer, "token")
async with Plugin(Platform.Generic, "0.1", reader, writer, "token") as plugin:
plugin.shutdown.return_value = async_return_value(None)
yield plugin
@pytest.fixture(autouse=True)
def my_caplog(caplog):

View File

@@ -1,91 +1,206 @@
import asyncio
import json
import pytest
from pytest import raises
from galaxy.api.types import Achievement
from galaxy.api.errors import UnknownError
from galaxy.api.errors import BackendError
from galaxy.unittest.mock import async_return_value
from tests import create_message, get_messages
def test_initialization_no_unlock_time():
with raises(Exception):
Achievement(achievement_id="lvl30", achievement_name="Got level 30")
def test_initialization_no_id_nor_name():
with raises(AssertionError):
Achievement(unlock_time=1234567890)
def test_success(plugin, readline, write):
@pytest.mark.asyncio
async def test_get_unlocked_achievements_success(plugin, read, write):
plugin.prepare_achievements_context.return_value = async_return_value(5)
request = {
"jsonrpc": "2.0",
"id": "3",
"method": "import_unlocked_achievements",
"method": "start_achievements_import",
"params": {
"game_id": "14"
"game_ids": ["14"]
}
}
readline.side_effect = [json.dumps(request), ""]
plugin.get_unlocked_achievements.coro.return_value = [
read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
plugin.get_unlocked_achievements.return_value = async_return_value([
Achievement(achievement_id="lvl10", unlock_time=1548421241),
Achievement(achievement_name="Got level 20", unlock_time=1548422395),
Achievement(achievement_id="lvl30", achievement_name="Got level 30", unlock_time=1548495633)
]
asyncio.run(plugin.run())
plugin.get_unlocked_achievements.assert_called_with(game_id="14")
response = json.loads(write.call_args[0][0])
])
await plugin.run()
plugin.prepare_achievements_context.assert_called_with(["14"])
plugin.get_unlocked_achievements.assert_called_with("14", 5)
plugin.achievements_import_complete.asert_called_with()
assert response == {
"jsonrpc": "2.0",
"id": "3",
"result": {
"unlocked_achievements": [
{
"achievement_id": "lvl10",
"unlock_time": 1548421241
},
{
"achievement_name": "Got level 20",
"unlock_time": 1548422395
},
{
"achievement_id": "lvl30",
"achievement_name": "Got level 30",
"unlock_time": 1548495633
}
]
assert get_messages(write) == [
{
"jsonrpc": "2.0",
"id": "3",
"result": None
},
{
"jsonrpc": "2.0",
"method": "game_achievements_import_success",
"params": {
"game_id": "14",
"unlocked_achievements": [
{
"achievement_id": "lvl10",
"unlock_time": 1548421241
},
{
"achievement_name": "Got level 20",
"unlock_time": 1548422395
},
{
"achievement_id": "lvl30",
"achievement_name": "Got level 30",
"unlock_time": 1548495633
}
]
}
},
{
"jsonrpc": "2.0",
"method": "achievements_import_finished",
"params": None
}
}
]
def test_failure(plugin, readline, write):
@pytest.mark.asyncio
@pytest.mark.parametrize("exception,code,message", [
(BackendError, 4, "Backend error"),
(KeyError, 0, "Unknown error")
])
async def test_get_unlocked_achievements_error(exception, code, message, plugin, read, write):
plugin.prepare_achievements_context.return_value = async_return_value(None)
request = {
"jsonrpc": "2.0",
"id": "3",
"method": "import_unlocked_achievements",
"method": "start_achievements_import",
"params": {
"game_id": "14"
"game_ids": ["14"]
}
}
readline.side_effect = [json.dumps(request), ""]
plugin.get_unlocked_achievements.coro.side_effect = UnknownError()
asyncio.run(plugin.run())
read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
plugin.get_unlocked_achievements.side_effect = exception
await plugin.run()
plugin.get_unlocked_achievements.assert_called()
response = json.loads(write.call_args[0][0])
plugin.achievements_import_complete.asert_called_with()
assert response == {
assert get_messages(write) == [
{
"jsonrpc": "2.0",
"id": "3",
"result": None
},
{
"jsonrpc": "2.0",
"method": "game_achievements_import_failure",
"params": {
"game_id": "14",
"error": {
"code": code,
"message": message
}
}
},
{
"jsonrpc": "2.0",
"method": "achievements_import_finished",
"params": None
}
]
@pytest.mark.asyncio
async def test_prepare_get_unlocked_achievements_context_error(plugin, read, write):
plugin.prepare_achievements_context.side_effect = BackendError()
request = {
"jsonrpc": "2.0",
"id": "3",
"error": {
"code": 0,
"message": "Unknown error"
"method": "start_achievements_import",
"params": {
"game_ids": ["14"]
}
}
read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
def test_unlock_achievement(plugin, write):
await plugin.run()
assert get_messages(write) == [
{
"jsonrpc": "2.0",
"id": "3",
"error": {
"code": 4,
"message": "Backend error"
}
}
]
@pytest.mark.asyncio
async def test_import_in_progress(plugin, read, write):
plugin.prepare_achievements_context.return_value = async_return_value(None)
plugin.get_unlocked_achievements.return_value = async_return_value([])
requests = [
{
"jsonrpc": "2.0",
"id": "3",
"method": "start_achievements_import",
"params": {
"game_ids": ["14"]
}
},
{
"jsonrpc": "2.0",
"id": "4",
"method": "start_achievements_import",
"params": {
"game_ids": ["15"]
}
}
]
read.side_effect = [
async_return_value(create_message(requests[0])),
async_return_value(create_message(requests[1])),
async_return_value(b"")
]
await plugin.run()
messages = get_messages(write)
assert {
"jsonrpc": "2.0",
"id": "3",
"result": None
} in messages
assert {
"jsonrpc": "2.0",
"id": "4",
"error": {
"code": 600,
"message": "Import already in progress"
}
} in messages
@pytest.mark.asyncio
async def test_unlock_achievement(plugin, write):
achievement = Achievement(achievement_id="lvl20", unlock_time=1548422395)
async def couritine():
plugin.unlock_achievement("14", achievement)
asyncio.run(couritine())
plugin.unlock_achievement("14", achievement)
response = json.loads(write.call_args[0][0])
assert response == {

View File

@@ -1,37 +1,40 @@
import asyncio
import json
import pytest
from galaxy.api.types import Authentication
from galaxy.api.errors import (
UnknownError, InvalidCredentials, NetworkError, LoggedInElsewhere, ProtocolError,
BackendNotAvailable, BackendTimeout, BackendError, TemporaryBlocked, Banned, AccessDenied,
ParentalControlBlock, DeviceBlocked, RegionBlocked
BackendNotAvailable, BackendTimeout, BackendError, TemporaryBlocked, Banned, AccessDenied
)
from galaxy.unittest.mock import async_return_value
def test_success(plugin, readline, write):
from tests import create_message, get_messages
@pytest.mark.asyncio
async def test_success(plugin, read, write):
request = {
"jsonrpc": "2.0",
"id": "3",
"method": "init_authentication"
}
readline.side_effect = [json.dumps(request), ""]
plugin.authenticate.coro.return_value = Authentication("132", "Zenek")
asyncio.run(plugin.run())
read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
plugin.authenticate.return_value = async_return_value(Authentication("132", "Zenek"))
await plugin.run()
plugin.authenticate.assert_called_with()
response = json.loads(write.call_args[0][0])
assert response == {
"jsonrpc": "2.0",
"id": "3",
"result": {
"user_id": "132",
"user_name": "Zenek"
assert get_messages(write) == [
{
"jsonrpc": "2.0",
"id": "3",
"result": {
"user_id": "132",
"user_name": "Zenek"
}
}
}
]
@pytest.mark.asyncio
@pytest.mark.parametrize("error,code,message", [
pytest.param(UnknownError, 0, "Unknown error", id="unknown_error"),
pytest.param(BackendNotAvailable, 2, "Backend not available", id="backend_not_available"),
@@ -44,33 +47,33 @@ def test_success(plugin, readline, write):
pytest.param(TemporaryBlocked, 104, "Temporary blocked", id="temporary_blocked"),
pytest.param(Banned, 105, "Banned", id="banned"),
pytest.param(AccessDenied, 106, "Access denied", id="access_denied"),
pytest.param(ParentalControlBlock, 107, "Parental control block", id="parental_control_clock"),
pytest.param(DeviceBlocked, 108, "Device blocked", id="device_blocked"),
pytest.param(RegionBlocked, 109, "Region blocked", id="region_blocked")
])
def test_failure(plugin, readline, write, error, code, message):
async def test_failure(plugin, read, write, error, code, message):
request = {
"jsonrpc": "2.0",
"id": "3",
"method": "init_authentication"
}
readline.side_effect = [json.dumps(request), ""]
plugin.authenticate.coro.side_effect = error()
asyncio.run(plugin.run())
read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
plugin.authenticate.side_effect = error()
await plugin.run()
plugin.authenticate.assert_called_with()
response = json.loads(write.call_args[0][0])
assert response == {
"jsonrpc": "2.0",
"id": "3",
"error": {
"code": code,
"message": message
assert get_messages(write) == [
{
"jsonrpc": "2.0",
"id": "3",
"error": {
"code": code,
"message": message
}
}
}
]
def test_stored_credentials(plugin, readline, write):
@pytest.mark.asyncio
async def test_stored_credentials(plugin, read, write):
request = {
"jsonrpc": "2.0",
"id": "3",
@@ -81,39 +84,37 @@ def test_stored_credentials(plugin, readline, write):
}
}
}
readline.side_effect = [json.dumps(request), ""]
plugin.authenticate.coro.return_value = Authentication("132", "Zenek")
asyncio.run(plugin.run())
read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
plugin.authenticate.return_value = async_return_value(Authentication("132", "Zenek"))
await plugin.run()
plugin.authenticate.assert_called_with(stored_credentials={"token": "ABC"})
write.assert_called()
def test_store_credentials(plugin, write):
@pytest.mark.asyncio
async def test_store_credentials(plugin, write):
credentials = {
"token": "ABC"
}
plugin.store_credentials(credentials)
async def couritine():
plugin.store_credentials(credentials)
assert get_messages(write) == [
{
"jsonrpc": "2.0",
"method": "store_credentials",
"params": credentials
}
]
asyncio.run(couritine())
response = json.loads(write.call_args[0][0])
assert response == {
"jsonrpc": "2.0",
"method": "store_credentials",
"params": credentials
}
@pytest.mark.asyncio
async def test_lost_authentication(plugin, write):
plugin.lost_authentication()
def test_lost_authentication(plugin, readline, write):
async def couritine():
plugin.lost_authentication()
asyncio.run(couritine())
response = json.loads(write.call_args[0][0])
assert response == {
"jsonrpc": "2.0",
"method": "authentication_lost",
"params": None
}
assert get_messages(write) == [
{
"jsonrpc": "2.0",
"method": "authentication_lost",
"params": None
}
]

View File

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

View File

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

View File

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

View File

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

View File

@@ -1,83 +1,215 @@
import asyncio
import json
from unittest.mock import call
import pytest
from galaxy.api.types import GameTime
from galaxy.api.errors import UnknownError
from galaxy.api.errors import BackendError
from galaxy.unittest.mock import async_return_value
def test_success(plugin, readline, write):
from tests import create_message, get_messages
@pytest.mark.asyncio
async def test_get_game_time_success(plugin, read, write):
plugin.prepare_game_times_context.return_value = async_return_value("abc")
request = {
"jsonrpc": "2.0",
"id": "3",
"method": "import_game_times"
"method": "start_game_times_import",
"params": {
"game_ids": ["3", "5", "7"]
}
}
readline.side_effect = [json.dumps(request), ""]
plugin.get_game_times.coro.return_value = [
GameTime("3", 60, 1549550504),
GameTime("5", 10, 1549550502)
read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
plugin.get_game_time.side_effect = [
async_return_value(GameTime("3", 60, 1549550504)),
async_return_value(GameTime("5", 10, None)),
async_return_value(GameTime("7", None, 1549550502)),
]
asyncio.run(plugin.run())
plugin.get_game_times.assert_called_with()
response = json.loads(write.call_args[0][0])
await plugin.run()
plugin.get_game_time.assert_has_calls([
call("3", "abc"),
call("5", "abc"),
call("7", "abc"),
])
plugin.game_times_import_complete.assert_called_once_with()
assert response == {
assert get_messages(write) == [
{
"jsonrpc": "2.0",
"id": "3",
"result": None
},
{
"jsonrpc": "2.0",
"method": "game_time_import_success",
"params": {
"game_time": {
"game_id": "3",
"last_played_time": 1549550504,
"time_played": 60
}
}
},
{
"jsonrpc": "2.0",
"method": "game_time_import_success",
"params": {
"game_time": {
"game_id": "5",
"time_played": 10
}
}
},
{
"jsonrpc": "2.0",
"method": "game_time_import_success",
"params": {
"game_time": {
"game_id": "7",
"last_played_time": 1549550502
}
}
},
{
"jsonrpc": "2.0",
"method": "game_times_import_finished",
"params": None
}
]
@pytest.mark.asyncio
@pytest.mark.parametrize("exception,code,message", [
(BackendError, 4, "Backend error"),
(KeyError, 0, "Unknown error")
])
async def test_get_game_time_error(exception, code, message, plugin, read, write):
plugin.prepare_game_times_context.return_value = async_return_value(None)
request = {
"jsonrpc": "2.0",
"id": "3",
"result": {
"game_times": [
{
"method": "start_game_times_import",
"params": {
"game_ids": ["6"]
}
}
read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
plugin.get_game_time.side_effect = exception
await plugin.run()
plugin.get_game_time.assert_called()
plugin.game_times_import_complete.assert_called_once_with()
assert get_messages(write) == [
{
"jsonrpc": "2.0",
"id": "3",
"result": None
},
{
"jsonrpc": "2.0",
"method": "game_time_import_failure",
"params": {
"game_id": "6",
"error": {
"code": code,
"message": message
}
}
},
{
"jsonrpc": "2.0",
"method": "game_times_import_finished",
"params": None
}
]
@pytest.mark.asyncio
async def test_prepare_get_game_time_context_error(plugin, read, write):
plugin.prepare_game_times_context.side_effect = BackendError()
request = {
"jsonrpc": "2.0",
"id": "3",
"method": "start_game_times_import",
"params": {
"game_ids": ["6"]
}
}
read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
await plugin.run()
assert get_messages(write) == [
{
"jsonrpc": "2.0",
"id": "3",
"error": {
"code": 4,
"message": "Backend error"
}
}
]
@pytest.mark.asyncio
async def test_import_in_progress(plugin, read, write):
plugin.prepare_game_times_context.return_value = async_return_value(None)
requests = [
{
"jsonrpc": "2.0",
"id": "3",
"method": "start_game_times_import",
"params": {
"game_ids": ["6"]
}
},
{
"jsonrpc": "2.0",
"id": "4",
"method": "start_game_times_import",
"params": {
"game_ids": ["7"]
}
}
]
read.side_effect = [
async_return_value(create_message(requests[0])),
async_return_value(create_message(requests[1])),
async_return_value(b"")
]
await plugin.run()
messages = get_messages(write)
assert {
"jsonrpc": "2.0",
"id": "3",
"result": None
} in messages
assert {
"jsonrpc": "2.0",
"id": "4",
"error": {
"code": 600,
"message": "Import already in progress"
}
} in messages
@pytest.mark.asyncio
async def test_update_game(plugin, write):
game_time = GameTime("3", 60, 1549550504)
plugin.update_game_time(game_time)
assert get_messages(write) == [
{
"jsonrpc": "2.0",
"method": "game_time_updated",
"params": {
"game_time": {
"game_id": "3",
"time_played": 60,
"last_played_time": 1549550504
},
{
"game_id": "5",
"time_played": 10,
"last_played_time": 1549550502
}
]
}
}
def test_failure(plugin, readline, write):
request = {
"jsonrpc": "2.0",
"id": "3",
"method": "import_game_times"
}
readline.side_effect = [json.dumps(request), ""]
plugin.get_game_times.coro.side_effect = UnknownError()
asyncio.run(plugin.run())
plugin.get_game_times.assert_called_with()
response = json.loads(write.call_args[0][0])
assert response == {
"jsonrpc": "2.0",
"id": "3",
"error": {
"code": 0,
"message": "Unknown error",
}
}
def test_update_game(plugin, write):
game_time = GameTime("3", 60, 1549550504)
async def couritine():
plugin.update_game_time(game_time)
asyncio.run(couritine())
response = json.loads(write.call_args[0][0])
assert response == {
"jsonrpc": "2.0",
"method": "game_time_updated",
"params": {
"game_time": {
"game_id": "3",
"time_played": 60,
"last_played_time": 1549550504
}
}
}
]

39
tests/test_http.py Normal file
View File

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

View File

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

View File

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

View File

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

View File

@@ -0,0 +1,17 @@
import pytest
from galaxy.unittest.mock import async_return_value
from tests import create_message
@pytest.mark.asyncio
async def test_success(plugin, read):
request = {
"jsonrpc": "2.0",
"method": "launch_platform_client"
}
read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
plugin.launch_platform_client.return_value = async_return_value(None)
await plugin.run()
plugin.launch_platform_client.assert_called_with()

View File

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

View File

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

View File

@@ -0,0 +1,73 @@
import pytest
from galaxy.unittest.mock import async_return_value
from tests import create_message, get_messages
def assert_rpc_response(write, response_id, result=None):
assert get_messages(write) == [
{
"jsonrpc": "2.0",
"id": str(response_id),
"result": result
}
]
def assert_rpc_request(write, method, params=None):
assert get_messages(write) == [
{
"jsonrpc": "2.0",
"method": method,
"params": {"data": params}
}
]
@pytest.fixture
def cache_data():
return {
"persistent key": "persistent value",
"persistent object": {"answer to everything": 42}
}
@pytest.mark.asyncio
async def test_initialize_cache(plugin, read, write, cache_data):
request_id = 3
request = {
"jsonrpc": "2.0",
"id": str(request_id),
"method": "initialize_cache",
"params": {"data": cache_data}
}
read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
assert {} == plugin.persistent_cache
await plugin.run()
plugin.handshake_complete.assert_called_once_with()
assert cache_data == plugin.persistent_cache
assert_rpc_response(write, response_id=request_id)
@pytest.mark.asyncio
async def test_set_cache(plugin, write, cache_data):
assert {} == plugin.persistent_cache
plugin.persistent_cache.update(cache_data)
plugin.push_cache()
assert_rpc_request(write, "push_cache", cache_data)
assert cache_data == plugin.persistent_cache
@pytest.mark.asyncio
async def test_clear_cache(plugin, write, cache_data):
plugin._persistent_cache = cache_data
plugin.persistent_cache.clear()
plugin.push_cache()
assert_rpc_request(write, "push_cache", {})
assert {} == plugin.persistent_cache

View File

@@ -0,0 +1,17 @@
import pytest
from galaxy.unittest.mock import async_return_value
from tests import create_message
@pytest.mark.asyncio
async def test_success(plugin, read):
request = {
"jsonrpc": "2.0",
"method": "shutdown_platform_client"
}
read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
plugin.shutdown_platform_client.return_value = async_return_value(None)
await plugin.run()
plugin.shutdown_platform_client.assert_called_with()

View File

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

View File

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

View File

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