mirror of
https://github.com/gogcom/galaxy-integrations-python-api.git
synced 2025-12-30 18:38:13 -05:00
123 lines
3.8 KiB
Python
123 lines
3.8 KiB
Python
import pytest
|
|
|
|
from galaxy.api.types import Authentication
|
|
from galaxy.api.errors import (
|
|
UnknownError, InvalidCredentials, NetworkError, LoggedInElsewhere, ProtocolError,
|
|
BackendNotAvailable, BackendTimeout, BackendError, TemporaryBlocked, Banned, AccessDenied
|
|
)
|
|
from galaxy.unittest.mock import async_return_value, skip_loop
|
|
|
|
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"
|
|
}
|
|
read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
|
|
plugin.authenticate.return_value = async_return_value(Authentication("132", "Zenek"))
|
|
await plugin.run()
|
|
plugin.authenticate.assert_called_with()
|
|
|
|
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"),
|
|
pytest.param(BackendTimeout, 3, "Backend timed out", id="backend_timeout"),
|
|
pytest.param(BackendError, 4, "Backend error", id="backend_error"),
|
|
pytest.param(InvalidCredentials, 100, "Invalid credentials", id="invalid_credentials"),
|
|
pytest.param(NetworkError, 101, "Network error", id="network_error"),
|
|
pytest.param(LoggedInElsewhere, 102, "Logged in elsewhere", id="logged_elsewhere"),
|
|
pytest.param(ProtocolError, 103, "Protocol error", id="protocol_error"),
|
|
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"),
|
|
])
|
|
async def test_failure(plugin, read, write, error, code, message):
|
|
request = {
|
|
"jsonrpc": "2.0",
|
|
"id": "3",
|
|
"method": "init_authentication"
|
|
}
|
|
|
|
read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
|
|
plugin.authenticate.side_effect = error()
|
|
await plugin.run()
|
|
plugin.authenticate.assert_called_with()
|
|
|
|
assert get_messages(write) == [
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"id": "3",
|
|
"error": {
|
|
"code": code,
|
|
"message": message
|
|
}
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stored_credentials(plugin, read, write):
|
|
request = {
|
|
"jsonrpc": "2.0",
|
|
"id": "3",
|
|
"method": "init_authentication",
|
|
"params": {
|
|
"stored_credentials": {
|
|
"token": "ABC"
|
|
}
|
|
}
|
|
}
|
|
read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
|
|
plugin.authenticate.return_value = async_return_value(Authentication("132", "Zenek"))
|
|
await plugin.run()
|
|
plugin.authenticate.assert_called_with(stored_credentials={"token": "ABC"})
|
|
write.assert_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_store_credentials(plugin, write):
|
|
credentials = {
|
|
"token": "ABC"
|
|
}
|
|
plugin.store_credentials(credentials)
|
|
await skip_loop()
|
|
|
|
assert get_messages(write) == [
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"method": "store_credentials",
|
|
"params": credentials
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lost_authentication(plugin, write):
|
|
plugin.lost_authentication()
|
|
await skip_loop()
|
|
|
|
assert get_messages(write) == [
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"method": "authentication_lost",
|
|
"params": None
|
|
}
|
|
]
|