mirror of
https://git.selfprivacy.org/SelfPrivacy/selfprivacy-rest-api.git
synced 2024-11-27 06:21:28 +00:00
c6919293b6
I've learned that there is no problem in grouping mutations like we do with queries. This is a big mistake from my side, now we have legacy not so conveniently placed endpoints. I've grouped all mutations, left the copies of old ones flattened in the root for backwards compatibility. We will migrate to mutation groups on client side, and backups now only use grouped mutations. Tests are updated.
572 lines
17 KiB
Python
572 lines
17 KiB
Python
# pylint: disable=redefined-outer-name
|
|
# pylint: disable=unused-argument
|
|
# pylint: disable=missing-function-docstring
|
|
import datetime
|
|
import pytest
|
|
from mnemonic import Mnemonic
|
|
|
|
from selfprivacy_api.repositories.tokens.json_tokens_repository import (
|
|
JsonTokensRepository,
|
|
)
|
|
from selfprivacy_api.models.tokens.token import Token
|
|
|
|
from tests.common import generate_api_query, read_json, write_json
|
|
|
|
TOKENS_FILE_CONTETS = {
|
|
"tokens": [
|
|
{
|
|
"token": "TEST_TOKEN",
|
|
"name": "test_token",
|
|
"date": "2022-01-14 08:31:10.789314",
|
|
},
|
|
{
|
|
"token": "TEST_TOKEN2",
|
|
"name": "test_token2",
|
|
"date": "2022-01-14 08:31:10.789314",
|
|
},
|
|
]
|
|
}
|
|
|
|
API_DEVICES_QUERY = """
|
|
devices {
|
|
creationDate
|
|
isCaller
|
|
name
|
|
}
|
|
"""
|
|
|
|
|
|
@pytest.fixture
|
|
def token_repo():
|
|
return JsonTokensRepository()
|
|
|
|
|
|
def test_graphql_tokens_info(authorized_client, tokens_file):
|
|
response = authorized_client.post(
|
|
"/graphql",
|
|
json={"query": generate_api_query([API_DEVICES_QUERY])},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is not None
|
|
assert response.json()["data"]["api"]["devices"] is not None
|
|
assert len(response.json()["data"]["api"]["devices"]) == 2
|
|
assert (
|
|
response.json()["data"]["api"]["devices"][0]["creationDate"]
|
|
== "2022-01-14T08:31:10.789314"
|
|
)
|
|
assert response.json()["data"]["api"]["devices"][0]["isCaller"] is True
|
|
assert response.json()["data"]["api"]["devices"][0]["name"] == "test_token"
|
|
assert (
|
|
response.json()["data"]["api"]["devices"][1]["creationDate"]
|
|
== "2022-01-14T08:31:10.789314"
|
|
)
|
|
assert response.json()["data"]["api"]["devices"][1]["isCaller"] is False
|
|
assert response.json()["data"]["api"]["devices"][1]["name"] == "test_token2"
|
|
|
|
|
|
def test_graphql_tokens_info_unauthorized(client, tokens_file):
|
|
response = client.post(
|
|
"/graphql",
|
|
json={"query": generate_api_query([API_DEVICES_QUERY])},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["data"] is None
|
|
|
|
|
|
DELETE_TOKEN_MUTATION = """
|
|
mutation DeleteToken($device: String!) {
|
|
api {
|
|
deleteDeviceApiToken(device: $device) {
|
|
success
|
|
message
|
|
code
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
|
|
def test_graphql_delete_token_unauthorized(client, tokens_file):
|
|
response = client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": DELETE_TOKEN_MUTATION,
|
|
"variables": {
|
|
"device": "test_token",
|
|
},
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["data"] is None
|
|
|
|
|
|
def test_graphql_delete_token(authorized_client, tokens_file):
|
|
response = authorized_client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": DELETE_TOKEN_MUTATION,
|
|
"variables": {
|
|
"device": "test_token2",
|
|
},
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is not None
|
|
assert response.json()["data"]["api"]["deleteDeviceApiToken"]["success"] is True
|
|
assert response.json()["data"]["api"]["deleteDeviceApiToken"]["message"] is not None
|
|
assert response.json()["data"]["api"]["deleteDeviceApiToken"]["code"] == 200
|
|
assert read_json(tokens_file) == {
|
|
"tokens": [
|
|
{
|
|
"token": "TEST_TOKEN",
|
|
"name": "test_token",
|
|
"date": "2022-01-14 08:31:10.789314",
|
|
}
|
|
]
|
|
}
|
|
|
|
|
|
def test_graphql_delete_self_token(authorized_client, tokens_file):
|
|
response = authorized_client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": DELETE_TOKEN_MUTATION,
|
|
"variables": {
|
|
"device": "test_token",
|
|
},
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is not None
|
|
assert response.json()["data"]["api"]["deleteDeviceApiToken"]["success"] is False
|
|
assert response.json()["data"]["api"]["deleteDeviceApiToken"]["message"] is not None
|
|
assert response.json()["data"]["api"]["deleteDeviceApiToken"]["code"] == 400
|
|
assert read_json(tokens_file) == TOKENS_FILE_CONTETS
|
|
|
|
|
|
def test_graphql_delete_nonexistent_token(
|
|
authorized_client,
|
|
tokens_file,
|
|
):
|
|
response = authorized_client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": DELETE_TOKEN_MUTATION,
|
|
"variables": {
|
|
"device": "test_token3",
|
|
},
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is not None
|
|
assert response.json()["data"]["api"]["deleteDeviceApiToken"]["success"] is False
|
|
assert response.json()["data"]["api"]["deleteDeviceApiToken"]["message"] is not None
|
|
assert response.json()["data"]["api"]["deleteDeviceApiToken"]["code"] == 404
|
|
assert read_json(tokens_file) == TOKENS_FILE_CONTETS
|
|
|
|
|
|
REFRESH_TOKEN_MUTATION = """
|
|
mutation RefreshToken {
|
|
api {
|
|
refreshDeviceApiToken {
|
|
success
|
|
message
|
|
code
|
|
token
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
|
|
def test_graphql_refresh_token_unauthorized(client, tokens_file):
|
|
response = client.post(
|
|
"/graphql",
|
|
json={"query": REFRESH_TOKEN_MUTATION},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["data"] is None
|
|
|
|
|
|
def test_graphql_refresh_token(
|
|
authorized_client,
|
|
tokens_file,
|
|
token_repo,
|
|
):
|
|
response = authorized_client.post(
|
|
"/graphql",
|
|
json={"query": REFRESH_TOKEN_MUTATION},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is not None
|
|
assert response.json()["data"]["api"]["refreshDeviceApiToken"]["success"] is True
|
|
assert (
|
|
response.json()["data"]["api"]["refreshDeviceApiToken"]["message"] is not None
|
|
)
|
|
assert response.json()["data"]["api"]["refreshDeviceApiToken"]["code"] == 200
|
|
token = token_repo.get_token_by_name("test_token")
|
|
assert token == Token(
|
|
token=response.json()["data"]["api"]["refreshDeviceApiToken"]["token"],
|
|
device_name="test_token",
|
|
created_at=datetime.datetime(2022, 1, 14, 8, 31, 10, 789314),
|
|
)
|
|
|
|
|
|
NEW_DEVICE_KEY_MUTATION = """
|
|
mutation NewDeviceKey {
|
|
api {
|
|
getNewDeviceApiKey {
|
|
success
|
|
message
|
|
code
|
|
key
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
|
|
def test_graphql_get_new_device_auth_key_unauthorized(
|
|
client,
|
|
tokens_file,
|
|
):
|
|
response = client.post(
|
|
"/graphql",
|
|
json={"query": NEW_DEVICE_KEY_MUTATION},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["data"] is None
|
|
|
|
|
|
def test_graphql_get_new_device_auth_key(
|
|
authorized_client,
|
|
tokens_file,
|
|
):
|
|
response = authorized_client.post(
|
|
"/graphql",
|
|
json={"query": NEW_DEVICE_KEY_MUTATION},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is not None
|
|
assert response.json()["data"]["api"]["getNewDeviceApiKey"]["success"] is True
|
|
assert response.json()["data"]["api"]["getNewDeviceApiKey"]["message"] is not None
|
|
assert response.json()["data"]["api"]["getNewDeviceApiKey"]["code"] == 200
|
|
assert (
|
|
response.json()["data"]["api"]["getNewDeviceApiKey"]["key"].split(" ").__len__()
|
|
== 12
|
|
)
|
|
token = (
|
|
Mnemonic(language="english")
|
|
.to_entropy(response.json()["data"]["api"]["getNewDeviceApiKey"]["key"])
|
|
.hex()
|
|
)
|
|
assert read_json(tokens_file)["new_device"]["token"] == token
|
|
|
|
|
|
INVALIDATE_NEW_DEVICE_KEY_MUTATION = """
|
|
mutation InvalidateNewDeviceKey {
|
|
api {
|
|
invalidateNewDeviceApiKey {
|
|
success
|
|
message
|
|
code
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
|
|
def test_graphql_invalidate_new_device_token_unauthorized(
|
|
client,
|
|
tokens_file,
|
|
):
|
|
response = client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": INVALIDATE_NEW_DEVICE_KEY_MUTATION,
|
|
"variables": {
|
|
"device": "test_token",
|
|
},
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["data"] is None
|
|
|
|
|
|
def test_graphql_get_and_delete_new_device_key(
|
|
authorized_client,
|
|
tokens_file,
|
|
):
|
|
response = authorized_client.post(
|
|
"/graphql",
|
|
json={"query": NEW_DEVICE_KEY_MUTATION},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is not None
|
|
assert response.json()["data"]["api"]["getNewDeviceApiKey"]["success"] is True
|
|
assert response.json()["data"]["api"]["getNewDeviceApiKey"]["message"] is not None
|
|
assert response.json()["data"]["api"]["getNewDeviceApiKey"]["code"] == 200
|
|
assert (
|
|
response.json()["data"]["api"]["getNewDeviceApiKey"]["key"].split(" ").__len__()
|
|
== 12
|
|
)
|
|
token = (
|
|
Mnemonic(language="english")
|
|
.to_entropy(response.json()["data"]["api"]["getNewDeviceApiKey"]["key"])
|
|
.hex()
|
|
)
|
|
assert read_json(tokens_file)["new_device"]["token"] == token
|
|
response = authorized_client.post(
|
|
"/graphql",
|
|
json={"query": INVALIDATE_NEW_DEVICE_KEY_MUTATION},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is not None
|
|
assert (
|
|
response.json()["data"]["api"]["invalidateNewDeviceApiKey"]["success"] is True
|
|
)
|
|
assert (
|
|
response.json()["data"]["api"]["invalidateNewDeviceApiKey"]["message"]
|
|
is not None
|
|
)
|
|
assert response.json()["data"]["api"]["invalidateNewDeviceApiKey"]["code"] == 200
|
|
assert read_json(tokens_file) == TOKENS_FILE_CONTETS
|
|
|
|
|
|
AUTHORIZE_WITH_NEW_DEVICE_KEY_MUTATION = """
|
|
mutation AuthorizeWithNewDeviceKey($input: UseNewDeviceKeyInput!) {
|
|
api {
|
|
authorizeWithNewDeviceApiKey(input: $input) {
|
|
success
|
|
message
|
|
code
|
|
token
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
|
|
def test_graphql_get_and_authorize_new_device(
|
|
client,
|
|
authorized_client,
|
|
tokens_file,
|
|
):
|
|
response = authorized_client.post(
|
|
"/graphql",
|
|
json={"query": NEW_DEVICE_KEY_MUTATION},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is not None
|
|
assert response.json()["data"]["api"]["getNewDeviceApiKey"]["success"] is True
|
|
assert response.json()["data"]["api"]["getNewDeviceApiKey"]["message"] is not None
|
|
assert response.json()["data"]["api"]["getNewDeviceApiKey"]["code"] == 200
|
|
mnemonic_key = response.json()["data"]["api"]["getNewDeviceApiKey"]["key"]
|
|
assert mnemonic_key.split(" ").__len__() == 12
|
|
key = Mnemonic(language="english").to_entropy(mnemonic_key).hex()
|
|
assert read_json(tokens_file)["new_device"]["token"] == key
|
|
response = client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": AUTHORIZE_WITH_NEW_DEVICE_KEY_MUTATION,
|
|
"variables": {
|
|
"input": {
|
|
"key": mnemonic_key,
|
|
"deviceName": "new_device",
|
|
}
|
|
},
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is not None
|
|
assert (
|
|
response.json()["data"]["api"]["authorizeWithNewDeviceApiKey"]["success"]
|
|
is True
|
|
)
|
|
assert (
|
|
response.json()["data"]["api"]["authorizeWithNewDeviceApiKey"]["message"]
|
|
is not None
|
|
)
|
|
assert response.json()["data"]["api"]["authorizeWithNewDeviceApiKey"]["code"] == 200
|
|
token = response.json()["data"]["api"]["authorizeWithNewDeviceApiKey"]["token"]
|
|
assert read_json(tokens_file)["tokens"][2]["token"] == token
|
|
assert read_json(tokens_file)["tokens"][2]["name"] == "new_device"
|
|
|
|
|
|
def test_graphql_authorize_new_device_with_invalid_key(
|
|
client,
|
|
tokens_file,
|
|
):
|
|
response = client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": AUTHORIZE_WITH_NEW_DEVICE_KEY_MUTATION,
|
|
"variables": {
|
|
"input": {
|
|
"key": "invalid_token",
|
|
"deviceName": "test_token",
|
|
}
|
|
},
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is not None
|
|
assert (
|
|
response.json()["data"]["api"]["authorizeWithNewDeviceApiKey"]["success"]
|
|
is False
|
|
)
|
|
assert (
|
|
response.json()["data"]["api"]["authorizeWithNewDeviceApiKey"]["message"]
|
|
is not None
|
|
)
|
|
assert response.json()["data"]["api"]["authorizeWithNewDeviceApiKey"]["code"] == 404
|
|
assert read_json(tokens_file) == TOKENS_FILE_CONTETS
|
|
|
|
|
|
def test_graphql_get_and_authorize_used_key(
|
|
client,
|
|
authorized_client,
|
|
tokens_file,
|
|
):
|
|
response = authorized_client.post(
|
|
"/graphql",
|
|
json={"query": NEW_DEVICE_KEY_MUTATION},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is not None
|
|
assert response.json()["data"]["api"]["getNewDeviceApiKey"]["success"] is True
|
|
assert response.json()["data"]["api"]["getNewDeviceApiKey"]["message"] is not None
|
|
assert response.json()["data"]["api"]["getNewDeviceApiKey"]["code"] == 200
|
|
mnemonic_key = response.json()["data"]["api"]["getNewDeviceApiKey"]["key"]
|
|
assert mnemonic_key.split(" ").__len__() == 12
|
|
key = Mnemonic(language="english").to_entropy(mnemonic_key).hex()
|
|
assert read_json(tokens_file)["new_device"]["token"] == key
|
|
response = client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": AUTHORIZE_WITH_NEW_DEVICE_KEY_MUTATION,
|
|
"variables": {
|
|
"input": {
|
|
"key": mnemonic_key,
|
|
"deviceName": "new_token",
|
|
}
|
|
},
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is not None
|
|
assert (
|
|
response.json()["data"]["api"]["authorizeWithNewDeviceApiKey"]["success"]
|
|
is True
|
|
)
|
|
assert (
|
|
response.json()["data"]["api"]["authorizeWithNewDeviceApiKey"]["message"]
|
|
is not None
|
|
)
|
|
assert response.json()["data"]["api"]["authorizeWithNewDeviceApiKey"]["code"] == 200
|
|
assert (
|
|
read_json(tokens_file)["tokens"][2]["token"]
|
|
== response.json()["data"]["api"]["authorizeWithNewDeviceApiKey"]["token"]
|
|
)
|
|
assert read_json(tokens_file)["tokens"][2]["name"] == "new_token"
|
|
|
|
response = client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": AUTHORIZE_WITH_NEW_DEVICE_KEY_MUTATION,
|
|
"variables": {
|
|
"input": {
|
|
"key": NEW_DEVICE_KEY_MUTATION,
|
|
"deviceName": "test_token2",
|
|
}
|
|
},
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is not None
|
|
assert (
|
|
response.json()["data"]["api"]["authorizeWithNewDeviceApiKey"]["success"]
|
|
is False
|
|
)
|
|
assert (
|
|
response.json()["data"]["api"]["authorizeWithNewDeviceApiKey"]["message"]
|
|
is not None
|
|
)
|
|
assert response.json()["data"]["api"]["authorizeWithNewDeviceApiKey"]["code"] == 404
|
|
assert read_json(tokens_file)["tokens"].__len__() == 3
|
|
|
|
|
|
def test_graphql_get_and_authorize_key_after_12_minutes(
|
|
client,
|
|
authorized_client,
|
|
tokens_file,
|
|
):
|
|
response = authorized_client.post(
|
|
"/graphql",
|
|
json={"query": NEW_DEVICE_KEY_MUTATION},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is not None
|
|
assert response.json()["data"]["api"]["getNewDeviceApiKey"]["success"] is True
|
|
assert response.json()["data"]["api"]["getNewDeviceApiKey"]["message"] is not None
|
|
assert response.json()["data"]["api"]["getNewDeviceApiKey"]["code"] == 200
|
|
assert (
|
|
response.json()["data"]["api"]["getNewDeviceApiKey"]["key"].split(" ").__len__()
|
|
== 12
|
|
)
|
|
key = (
|
|
Mnemonic(language="english")
|
|
.to_entropy(response.json()["data"]["api"]["getNewDeviceApiKey"]["key"])
|
|
.hex()
|
|
)
|
|
assert read_json(tokens_file)["new_device"]["token"] == key
|
|
|
|
file_data = read_json(tokens_file)
|
|
file_data["new_device"]["expiration"] = str(
|
|
datetime.datetime.now() - datetime.timedelta(minutes=13)
|
|
)
|
|
write_json(tokens_file, file_data)
|
|
|
|
response = client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": AUTHORIZE_WITH_NEW_DEVICE_KEY_MUTATION,
|
|
"variables": {
|
|
"input": {
|
|
"key": key,
|
|
"deviceName": "test_token",
|
|
}
|
|
},
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is not None
|
|
assert (
|
|
response.json()["data"]["api"]["authorizeWithNewDeviceApiKey"]["success"]
|
|
is False
|
|
)
|
|
assert (
|
|
response.json()["data"]["api"]["authorizeWithNewDeviceApiKey"]["message"]
|
|
is not None
|
|
)
|
|
assert response.json()["data"]["api"]["authorizeWithNewDeviceApiKey"]["code"] == 404
|
|
|
|
|
|
def test_graphql_authorize_without_token(
|
|
client,
|
|
tokens_file,
|
|
):
|
|
response = client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": AUTHORIZE_WITH_NEW_DEVICE_KEY_MUTATION,
|
|
"variables": {
|
|
"input": {
|
|
"deviceName": "test_token",
|
|
}
|
|
},
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is None
|