SelfPrivacy
/
selfprivacy-rest-api
7
2
Fork 1
Code Issues 15 Pull Requests 4 Projects Releases 1 Wiki Activity

test(services): add unauthorized move test

remove-rest
Houkime 2023-11-22 18:13:07 +00:00
parent 2987065231
commit 5214d5e462
6 changed files with 72 additions and 160 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

89
tests/test_graphql/api_common.py
View File

@ -1,89 +0,0 @@
from tests.common import generate_api_query
from tests.conftest import TOKENS_FILE_CONTENTS, DEVICE_WE_AUTH_TESTS_WITH
ORIGINAL_DEVICES = TOKENS_FILE_CONTENTS["tokens"]
def assert_ok(response, request):
data = assert_data(response)
assert data[request]["success"] is True
assert data[request]["message"] is not None
assert data[request]["code"] == 200
def assert_errorcode(response, request, code):
data = assert_data(response)
assert data[request]["success"] is False
assert data[request]["message"] is not None
assert data[request]["code"] == code
def assert_empty(response):
assert response.status_code == 200
assert response.json().get("data") is None
def assert_data(response):
assert response.status_code == 200
data = response.json().get("data")
assert data is not None
assert "api" in data.keys()
return data["api"]
API_DEVICES_QUERY = """
devices {
creationDate
isCaller
name
}
"""
def request_devices(client):
return client.post(
"/graphql",
json={"query": generate_api_query([API_DEVICES_QUERY])},
)
def graphql_get_devices(client):
response = request_devices(client)
data = assert_data(response)
devices = data["devices"]
assert devices is not None
return devices
def set_client_token(client, token):
client.headers.update({"Authorization": "Bearer " + token})
def assert_token_valid(client, token):
set_client_token(client, token)
assert graphql_get_devices(client) is not None
def assert_same(graphql_devices, abstract_devices):
"""Orderless comparison"""
assert len(graphql_devices) == len(abstract_devices)
for original_device in abstract_devices:
assert original_device["name"] in [device["name"] for device in graphql_devices]
for device in graphql_devices:
if device["name"] == original_device["name"]:
assert device["creationDate"] == original_device["date"].isoformat()
def assert_original(client):
devices = graphql_get_devices(client)
assert_original_devices(devices)
def assert_original_devices(devices):
assert_same(devices, ORIGINAL_DEVICES)
for device in devices:
if device["name"] == DEVICE_WE_AUTH_TESTS_WITH["name"]:
assert device["isCaller"] is True
else:
assert device["isCaller"] is False

26
tests/test_graphql/common.py
View File

@ -4,18 +4,20 @@ from tests.conftest import TOKENS_FILE_CONTENTS, DEVICE_WE_AUTH_TESTS_WITH
ORIGINAL_DEVICES = TOKENS_FILE_CONTENTS["tokens"]
def assert_ok(response, request):
data = assert_data(response)
data[request]["success"] is True
data[request]["message"] is not None
data[request]["code"] == 200
def assert_ok(output: dict) -> None:
if output["success"] is False:
# convenience for debugging, this should display error
# if message is empty, consider adding helpful messages
raise ValueError(output["code"], output["message"])
assert output["success"] is True
assert output["message"] is not None
assert output["code"] == 200
def assert_errorcode(response, request, code):
data = assert_data(response)
data[request]["success"] is False
data[request]["message"] is not None
data[request]["code"] == code
def assert_errorcode(output: dict, code) -> None:
assert output["success"] is False
assert output["message"] is not None
assert output["code"] == code
def assert_empty(response):
@ -23,7 +25,7 @@ def assert_empty(response):
assert response.json().get("data") is None
def assert_data(response):
def get_data(response):
assert response.status_code == 200
response = response.json()
@ -54,7 +56,7 @@ def request_devices(client):
def graphql_get_devices(client):
response = request_devices(client)
data = assert_data(response)
data = get_data(response)
devices = data["api"]["devices"]
assert devices is not None
return devices

2
tests/test_graphql/test_api_backup.py
View File

@ -280,7 +280,7 @@ def get_data(response):
if (
"errors" in response.keys()
): # convenience for debugging, this will display error
assert response["errors"] == []
raise ValueError(response["errors"])
assert response["data"] is not None
data = response["data"]
return data

27
tests/test_graphql/test_api_devices.py
View File

@ -8,8 +8,8 @@ from tests.common import (
generate_api_query,
)
from tests.conftest import DEVICE_WE_AUTH_TESTS_WITH, TOKENS_FILE_CONTENTS
from tests.test_graphql.api_common import (
assert_data,
from tests.test_graphql.common import (
get_data,
assert_empty,
assert_ok,
assert_errorcode,
@ -36,7 +36,7 @@ def graphql_get_new_device_key(authorized_client) -> str:
"/graphql",
json={"query": NEW_DEVICE_KEY_MUTATION},
)
assert_ok(response, "getNewDeviceApiKey")
assert_ok(get_data(response)["api"]["getNewDeviceApiKey"])
key = response.json()["data"]["api"]["getNewDeviceApiKey"]["key"]
assert key.split(" ").__len__() == 12
@ -60,9 +60,10 @@ def graphql_try_auth_new_device(client, mnemonic_key, device_name):
def graphql_authorize_new_device(client, mnemonic_key, device_name) -> str:
response = graphql_try_auth_new_device(client, mnemonic_key, "new_device")
assert_ok(response, "authorizeWithNewDeviceApiKey")
assert_ok(get_data(response)["api"]["authorizeWithNewDeviceApiKey"])
token = response.json()["data"]["api"]["authorizeWithNewDeviceApiKey"]["token"]
assert_token_valid(client, token)
return token
def test_graphql_tokens_info(authorized_client, tokens_file):
@ -114,7 +115,7 @@ def test_graphql_delete_token(authorized_client, tokens_file):
},
},
)
assert_ok(response, "deleteDeviceApiToken")
assert_ok(get_data(response)["api"]["deleteDeviceApiToken"])
devices = graphql_get_devices(authorized_client)
assert_same(devices, test_devices)
@ -130,7 +131,7 @@ def test_graphql_delete_self_token(authorized_client, tokens_file):
},
},
)
assert_errorcode(response, "deleteDeviceApiToken", 400)
assert_errorcode(get_data(response)["api"]["deleteDeviceApiToken"], 400)
assert_original(authorized_client)
@ -147,7 +148,7 @@ def test_graphql_delete_nonexistent_token(
},
},
)
assert_errorcode(response, "deleteDeviceApiToken", 404)
assert_errorcode(get_data(response)["api"]["deleteDeviceApiToken"], 404)
assert_original(authorized_client)
@ -180,7 +181,7 @@ def test_graphql_refresh_token(authorized_client, client, tokens_file):
"/graphql",
json={"query": REFRESH_TOKEN_MUTATION},
)
assert_ok(response, "refreshDeviceApiToken")
assert_ok(get_data(response)["api"]["refreshDeviceApiToken"])
new_token = response.json()["data"]["api"]["refreshDeviceApiToken"]["token"]
assert_token_valid(client, new_token)
@ -250,10 +251,10 @@ def test_graphql_get_and_delete_new_device_key(client, authorized_client, tokens
"/graphql",
json={"query": INVALIDATE_NEW_DEVICE_KEY_MUTATION},
)
assert_ok(response, "invalidateNewDeviceApiKey")
assert_ok(get_data(response)["api"]["invalidateNewDeviceApiKey"])
response = graphql_try_auth_new_device(client, mnemonic_key, "new_device")
assert_errorcode(response, "authorizeWithNewDeviceApiKey", 404)
assert_errorcode(get_data(response)["api"]["authorizeWithNewDeviceApiKey"], 404)
AUTHORIZE_WITH_NEW_DEVICE_KEY_MUTATION = """
@ -285,7 +286,7 @@ def test_graphql_authorize_new_device_with_invalid_key(
client, authorized_client, tokens_file
):
response = graphql_try_auth_new_device(client, "invalid_token", "new_device")
assert_errorcode(response, "authorizeWithNewDeviceApiKey", 404)
assert_errorcode(get_data(response)["api"]["authorizeWithNewDeviceApiKey"], 404)
assert_original(authorized_client)
@ -297,7 +298,7 @@ def test_graphql_get_and_authorize_used_key(client, authorized_client, tokens_fi
devices = graphql_get_devices(authorized_client)
response = graphql_try_auth_new_device(client, mnemonic_key, "new_device2")
assert_errorcode(response, "authorizeWithNewDeviceApiKey", 404)
assert_errorcode(get_data(response)["api"]["authorizeWithNewDeviceApiKey"], 404)
assert graphql_get_devices(authorized_client) == devices
@ -309,7 +310,7 @@ def test_graphql_get_and_authorize_key_after_12_minutes(
mock = mocker.patch(DEVICE_KEY_VALIDATION_DATETIME, NearFuture)
response = graphql_try_auth_new_device(client, mnemonic_key, "new_device")
assert_errorcode(response, "authorizeWithNewDeviceApiKey", 404)
assert_errorcode(get_data(response)["api"]["authorizeWithNewDeviceApiKey"], 404)
def test_graphql_authorize_without_token(

46
tests/test_graphql/test_api_recovery.py
View File

@ -18,9 +18,9 @@ from tests.common import five_minutes_into_future_naive_utc as five_minutes_into
from tests.common import five_minutes_into_future as five_minutes_into_future_tz
from tests.common import five_minutes_into_past_naive_utc as five_minutes_into_past
from tests.test_graphql.api_common import (
from tests.test_graphql.common import (
assert_empty,
assert_data,
get_data,
assert_ok,
assert_errorcode,
assert_token_valid,
@ -49,9 +49,9 @@ def request_recovery_status(client):
def graphql_recovery_status(client):
response = request_recovery_status(client)
data = assert_data(response)
data = get_data(response)
status = data["recoveryKey"]
status = data["api"]["recoveryKey"]
assert status is not None
return status
@ -74,8 +74,10 @@ def request_make_new_recovery_key(client, expires_at=None, uses=None):
def graphql_make_new_recovery_key(client, expires_at=None, uses=None):
response = request_make_new_recovery_key(client, expires_at, uses)
assert_ok(response, "getNewRecoveryApiKey")
key = response.json()["data"]["api"]["getNewRecoveryApiKey"]["key"]
output = get_data(response)["api"]["getNewRecoveryApiKey"]
assert_ok(output)
key = output["key"]
assert key is not None
assert key.split(" ").__len__() == 18
return key
@ -98,8 +100,10 @@ def request_recovery_auth(client, key, device_name):
def graphql_use_recovery_key(client, key, device_name):
response = request_recovery_auth(client, key, device_name)
assert_ok(response, "useRecoveryApiKey")
token = response.json()["data"]["api"]["useRecoveryApiKey"]["token"]
output = get_data(response)["api"]["useRecoveryApiKey"]
assert_ok(output)
token = output["token"]
assert token is not None
assert_token_valid(client, token)
set_client_token(client, token)
@ -198,8 +202,10 @@ def test_graphql_use_recovery_key_after_expiration(
mock = mocker.patch(RECOVERY_KEY_VALIDATION_DATETIME, NearFuture)
response = request_recovery_auth(client, key, "new_test_token3")
assert_errorcode(response, "useRecoveryApiKey", 404)
assert response.json()["data"]["api"]["useRecoveryApiKey"]["token"] is None
output = get_data(response)["api"]["useRecoveryApiKey"]
assert_errorcode(output, 404)
assert output["token"] is None
assert_original(authorized_client)
status = graphql_recovery_status(authorized_client)
@ -222,8 +228,10 @@ def test_graphql_generate_recovery_key_with_expiration_in_the_past(
authorized_client, expires_at=expiration_date
)
assert_errorcode(response, "getNewRecoveryApiKey", 400)
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["key"] is None
output = get_data(response)["api"]["getNewRecoveryApiKey"]
assert_errorcode(output, 400)
assert output["key"] is None
assert graphql_recovery_status(authorized_client)["exists"] is False
@ -280,7 +288,8 @@ def test_graphql_generate_recovery_key_with_limited_uses(
assert status["usesLeft"] == 0
response = request_recovery_auth(client, mnemonic_key, "new_test_token3")
assert_errorcode(response, "useRecoveryApiKey", 404)
output = get_data(response)["api"]["useRecoveryApiKey"]
assert_errorcode(output, 404)
def test_graphql_generate_recovery_key_with_negative_uses(
@ -288,13 +297,16 @@ def test_graphql_generate_recovery_key_with_negative_uses(
):
response = request_make_new_recovery_key(authorized_client, uses=-1)
assert_errorcode(response, "getNewRecoveryApiKey", 400)
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["key"] is None
output = get_data(response)["api"]["getNewRecoveryApiKey"]
assert_errorcode(output, 400)
assert output["key"] is None
assert graphql_recovery_status(authorized_client)["exists"] is False
def test_graphql_generate_recovery_key_with_zero_uses(authorized_client, tokens_file):
response = request_make_new_recovery_key(authorized_client, uses=0)
assert_errorcode(response, "getNewRecoveryApiKey", 400)
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["key"] is None
output = get_data(response)["api"]["getNewRecoveryApiKey"]
assert_errorcode(output, 400)
assert output["key"] is None
assert graphql_recovery_status(authorized_client)["exists"] is False

42
tests/test_graphql/test_services.py
View File

@ -10,7 +10,7 @@ from selfprivacy_api.services.test_service import DummyService
from tests.test_common import raw_dummy_service, dummy_service
from tests.common import generate_service_query
from tests.test_graphql.test_api_backup import assert_ok, get_data
from tests.test_graphql.common import assert_empty, assert_ok, get_data
@pytest.fixture()
@ -330,52 +330,38 @@ def test_allservices_unauthorized(client, only_dummy_service):
def test_start_unauthorized(client, only_dummy_service):
dummy_service = only_dummy_service
mutation_response = api_start(client, dummy_service)
assert mutation_response.status_code == 200
assert mutation_response.json().get("data") is None
response = api_start(client, dummy_service)
assert_empty(response)
def test_restart_unauthorized(client, only_dummy_service):
dummy_service = only_dummy_service
mutation_response = api_restart(client, dummy_service)
assert mutation_response.status_code == 200
assert mutation_response.json().get("data") is None
response = api_restart(client, dummy_service)
assert_empty(response)
def test_stop_unauthorized(client, only_dummy_service):
dummy_service = only_dummy_service
mutation_response = api_stop(client, dummy_service)
assert mutation_response.status_code == 200
assert mutation_response.json().get("data") is None
response = api_stop(client, dummy_service)
assert_empty(response)
def test_enable_unauthorized(client, only_dummy_service):
dummy_service = only_dummy_service
mutation_response = api_enable(client, dummy_service)
assert mutation_response.status_code == 200
assert mutation_response.json().get("data") is None
response = api_enable(client, dummy_service)
assert_empty(response)
def test_disable_unauthorized(client, only_dummy_service):
dummy_service = only_dummy_service
mutation_response = api_disable(client, dummy_service)
assert mutation_response.status_code == 200
assert mutation_response.json().get("data") is None
response = api_disable(client, dummy_service)
assert_empty(response)
def test_move_nonexistent(authorized_client, only_dummy_service):
def test_move_unauthorized(client, only_dummy_service):
dummy_service = only_dummy_service
mutation_response = api_move_by_name(authorized_client, "bogus_service", "sda1")
data = get_data(mutation_response)["services"]["moveService"]
assert_notfound(data)
assert data["service"] is None
assert data["job"] is None
response = api_move(client, dummy_service, "sda1")
assert_empty(response)
def test_start_nonexistent(authorized_client, only_dummy_service):