diff --git a/selfprivacy_api/migrations/__init__.py b/selfprivacy_api/migrations/__init__.py index 8209198..b051f04 100644 --- a/selfprivacy_api/migrations/__init__.py +++ b/selfprivacy_api/migrations/__init__.py @@ -8,7 +8,9 @@ at api.skippedMigrations in userdata.json and populating it with IDs of the migrations to skip. Adding DISABLE_ALL to that array disables the migrations module entirely. """ -from selfprivacy_api.migrations.check_for_failed_binds_migration import CheckForFailedBindsMigration +from selfprivacy_api.migrations.check_for_failed_binds_migration import ( + CheckForFailedBindsMigration, +) from selfprivacy_api.utils import ReadUserData from selfprivacy_api.migrations.fix_nixos_config_branch import FixNixosConfigBranch from selfprivacy_api.migrations.create_tokens_json import CreateTokensJson diff --git a/tests/test_graphql/test_api_services.py b/tests/test_graphql/test_api_services.py new file mode 100644 index 0000000..ac6d217 --- /dev/null +++ b/tests/test_graphql/test_api_services.py @@ -0,0 +1,626 @@ +# pylint: disable=redefined-outer-name +# pylint: disable=unused-argument +import pytest + +from tests.common import read_json + + +class NextcloudMockReturnTrue: + def __init__(self, args, **kwargs): + self.args = args + self.kwargs = kwargs + + def enable(): + pass + + def disable(): + pass + + def stop(): + pass + + def is_movable(): + return True + + def move_to_volume(what): + return None + + def start(): + pass + + def restart(): + pass + + returncode = 0 + + +class BlockDevices: + def get_block_device(location): + return True + + +class ProcessMock: + """Mock subprocess.Popen""" + + def __init__(self, args, **kwargs): + self.args = args + self.kwargs = kwargs + + def communicate(): # pylint: disable=no-method-argument + return (b"", None) + + returncode = 0 + + +@pytest.fixture +def mock_subprocess_popen(mocker): + mock = mocker.patch("subprocess.Popen", autospec=True, return_value=ProcessMock) + return mock + + +@pytest.fixture +def one_user(mocker, datadir): + mocker.patch("selfprivacy_api.utils.USERDATA_FILE", new=datadir / "one_user.json") + assert read_json(datadir / "one_user.json")["users"] == [ + { + "username": "user1", + "hashedPassword": "HASHED_PASSWORD_1", + "sshKeys": ["ssh-rsa KEY user1@pc"], + } + ] + return datadir + + +@pytest.fixture +def mock_service_to_graphql_service(mocker): + mock = mocker.patch( + "selfprivacy_api.graphql.mutations.services_mutations.service_to_graphql_service", + autospec=True, + return_value=None, + ) + return mock + + +@pytest.fixture +def mock_job_to_api_job(mocker): + mock = mocker.patch( + "selfprivacy_api.graphql.mutations.services_mutations.job_to_api_job", + autospec=True, + return_value=None, + ) + return mock + + +@pytest.fixture +def mock_block_devices_return_none(mocker): + mock = mocker.patch( + "selfprivacy_api.utils.block_devices.BlockDevices", + autospec=True, + return_value=None, + ) + return mock + + +@pytest.fixture +def mock_block_devices(mocker): + mock = mocker.patch( + "selfprivacy_api.graphql.mutations.services_mutations.BlockDevices", + autospec=True, + return_value=BlockDevices, + ) + return mock + + +@pytest.fixture +def mock_get_service_by_id_return_none(mocker): + mock = mocker.patch( + "selfprivacy_api.graphql.mutations.services_mutations.get_service_by_id", + autospec=True, + return_value=None, + ) + return mock + + +@pytest.fixture +def mock_get_service_by_id(mocker): + mock = mocker.patch( + "selfprivacy_api.graphql.mutations.services_mutations.get_service_by_id", + autospec=True, + return_value=NextcloudMockReturnTrue, + ) + return mock + + +#################################################################### + + +API_ENABLE_SERVICE_MUTATION = """ +mutation enableService($serviceId: String!) { + enableService(serviceId: $serviceId) { + success + message + code + } +} +""" + + +def test_graphql_enable_service_unauthorized_client( + client, mock_get_service_by_id_return_none, mock_subprocess_popen +): + response = client.post( + "/graphql", + json={ + "query": API_ENABLE_SERVICE_MUTATION, + "variables": {"serviceId": "nextcloud"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is None + + +def test_graphql_enable_not_found_service( + authorized_client, + mock_get_service_by_id_return_none, + mock_subprocess_popen, + one_user, + mock_service_to_graphql_service, +): + response = authorized_client.post( + "/graphql", + json={ + "query": API_ENABLE_SERVICE_MUTATION, + "variables": {"serviceId": "nextcloud"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is not None + + assert response.json()["data"]["enableService"]["code"] == 404 + assert response.json()["data"]["enableService"]["message"] is not None + assert response.json()["data"]["enableService"]["success"] is False + + +def test_graphql_enable_service( + authorized_client, + mock_get_service_by_id, + mock_subprocess_popen, + one_user, + mock_service_to_graphql_service, +): + response = authorized_client.post( + "/graphql", + json={ + "query": API_ENABLE_SERVICE_MUTATION, + "variables": {"serviceId": "nextcloud"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is not None + + assert response.json()["data"]["enableService"]["code"] == 200 + assert response.json()["data"]["enableService"]["message"] is not None + assert response.json()["data"]["enableService"]["success"] is True + + +API_DISABLE_SERVICE_MUTATION = """ +mutation disableService($serviceId: String!) { + disableService(serviceId: $serviceId) { + success + message + code + } +} +""" + + +def test_graphql_disable_service_unauthorized_client( + client, + mock_get_service_by_id_return_none, + mock_subprocess_popen, + one_user, + mock_service_to_graphql_service, +): + response = client.post( + "/graphql", + json={ + "query": API_DISABLE_SERVICE_MUTATION, + "variables": {"serviceId": "nextcloud"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is None + + +def test_graphql_disable_not_found_service( + authorized_client, + mock_get_service_by_id_return_none, + mock_subprocess_popen, + one_user, + mock_service_to_graphql_service, +): + response = authorized_client.post( + "/graphql", + json={ + "query": API_DISABLE_SERVICE_MUTATION, + "variables": {"serviceId": "nextcloud"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is not None + + assert response.json()["data"]["disableService"]["code"] == 404 + assert response.json()["data"]["disableService"]["message"] is not None + assert response.json()["data"]["disableService"]["success"] is False + + +def test_graphql_disable_services( + authorized_client, + mock_get_service_by_id, + mock_subprocess_popen, + one_user, + mock_service_to_graphql_service, +): + response = authorized_client.post( + "/graphql", + json={ + "query": API_DISABLE_SERVICE_MUTATION, + "variables": {"serviceId": "nextcloud"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is not None + + assert response.json()["data"]["disableService"]["code"] == 200 + assert response.json()["data"]["disableService"]["message"] is not None + assert response.json()["data"]["disableService"]["success"] is True + + +API_STOP_SERVICE_MUTATION = """ +mutation stopService($serviceId: String!) { + stopService(serviceId: $serviceId) { + success + message + code + } +} +""" + + +def test_graphql_stop_service_unauthorized_client( + client, + mock_get_service_by_id_return_none, + mock_service_to_graphql_service, + mock_subprocess_popen, + one_user, +): + response = client.post( + "/graphql", + json={ + "query": API_STOP_SERVICE_MUTATION, + "variables": {"serviceId": "nextcloud"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is None + + +def test_graphql_stop_not_found_service( + authorized_client, + mock_get_service_by_id_return_none, + mock_service_to_graphql_service, + mock_subprocess_popen, + one_user, +): + response = authorized_client.post( + "/graphql", + json={ + "query": API_STOP_SERVICE_MUTATION, + "variables": {"serviceId": "nextcloud"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is not None + + assert response.json()["data"]["stopService"]["code"] == 404 + assert response.json()["data"]["stopService"]["message"] is not None + assert response.json()["data"]["stopService"]["success"] is False + + +def test_graphql_stop_service( + authorized_client, + mock_get_service_by_id, + mock_service_to_graphql_service, + mock_subprocess_popen, + one_user, +): + response = authorized_client.post( + "/graphql", + json={ + "query": API_STOP_SERVICE_MUTATION, + "variables": {"serviceId": "nextcloud"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is not None + + assert response.json()["data"]["stopService"]["code"] == 200 + assert response.json()["data"]["stopService"]["message"] is not None + assert response.json()["data"]["stopService"]["success"] is True + + +API_START_SERVICE_MUTATION = """ +mutation startService($serviceId: String!) { + startService(serviceId: $serviceId) { + success + message + code + } +} +""" + + +def test_graphql_start_service_unauthorized_client( + client, + mock_get_service_by_id_return_none, + mock_service_to_graphql_service, + mock_subprocess_popen, + one_user, +): + response = client.post( + "/graphql", + json={ + "query": API_START_SERVICE_MUTATION, + "variables": {"serviceId": "nextcloud"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is None + + +def test_graphql_start_not_found_service( + authorized_client, + mock_get_service_by_id_return_none, + mock_service_to_graphql_service, + mock_subprocess_popen, + one_user, +): + response = authorized_client.post( + "/graphql", + json={ + "query": API_START_SERVICE_MUTATION, + "variables": {"serviceId": "nextcloud"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is not None + + assert response.json()["data"]["startService"]["code"] == 404 + assert response.json()["data"]["startService"]["message"] is not None + assert response.json()["data"]["startService"]["success"] is False + + +def test_graphql_start_service( + authorized_client, + mock_get_service_by_id, + mock_service_to_graphql_service, + mock_subprocess_popen, + one_user, +): + response = authorized_client.post( + "/graphql", + json={ + "query": API_START_SERVICE_MUTATION, + "variables": {"serviceId": "nextcloud"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is not None + + assert response.json()["data"]["startService"]["code"] == 200 + assert response.json()["data"]["startService"]["message"] is not None + assert response.json()["data"]["startService"]["success"] is True + + +API_RESTART_SERVICE_MUTATION = """ +mutation restartService($serviceId: String!) { + restartService(serviceId: $serviceId) { + success + message + code + } +} +""" + + +def test_graphql_restart_service_unauthorized_client( + client, + mock_get_service_by_id_return_none, + mock_service_to_graphql_service, + mock_subprocess_popen, + one_user, +): + response = client.post( + "/graphql", + json={ + "query": API_RESTART_SERVICE_MUTATION, + "variables": {"serviceId": "nextcloud"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is None + + +def test_graphql_restart_not_found_service( + authorized_client, + mock_get_service_by_id_return_none, + mock_service_to_graphql_service, + mock_subprocess_popen, + one_user, +): + response = authorized_client.post( + "/graphql", + json={ + "query": API_RESTART_SERVICE_MUTATION, + "variables": {"serviceId": "nextcloud"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is not None + + assert response.json()["data"]["restartService"]["code"] == 404 + assert response.json()["data"]["restartService"]["message"] is not None + assert response.json()["data"]["restartService"]["success"] is False + + +def test_graphql_restart_service( + authorized_client, + mock_get_service_by_id, + mock_service_to_graphql_service, + mock_subprocess_popen, + one_user, +): + response = authorized_client.post( + "/graphql", + json={ + "query": API_RESTART_SERVICE_MUTATION, + "variables": {"serviceId": "nextcloud"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is not None + + assert response.json()["data"]["restartService"]["code"] == 200 + assert response.json()["data"]["restartService"]["message"] is not None + assert response.json()["data"]["restartService"]["success"] is True + + +API_MOVE_SERVICE_MUTATION = """ +mutation moveService($input: MoveServiceInput!) { + moveService(input: $input) { + success + message + code + } +} +""" + + +def test_graphql_move_service_unauthorized_client( + client, + mock_get_service_by_id_return_none, + mock_service_to_graphql_service, + mock_subprocess_popen, + one_user, +): + response = client.post( + "/graphql", + json={ + "query": API_MOVE_SERVICE_MUTATION, + "variables": { + "input": {"serviceId": "nextcloud", "location": "sdx"}, + }, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is None + + +def test_graphql_move_not_found_service( + authorized_client, + mock_get_service_by_id_return_none, + mock_service_to_graphql_service, + mock_subprocess_popen, + one_user, +): + response = authorized_client.post( + "/graphql", + json={ + "query": API_MOVE_SERVICE_MUTATION, + "variables": { + "input": {"serviceId": "nextcloud", "location": "sdx"}, + }, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is not None + + assert response.json()["data"]["moveService"]["code"] == 404 + assert response.json()["data"]["moveService"]["message"] is not None + assert response.json()["data"]["moveService"]["success"] is False + + +def test_graphql_move_not_movable_service( + authorized_client, + mock_get_service_by_id_return_none, + mock_service_to_graphql_service, + mock_subprocess_popen, + one_user, +): + response = authorized_client.post( + "/graphql", + json={ + "query": API_MOVE_SERVICE_MUTATION, + "variables": { + "input": {"serviceId": "nextcloud", "location": "sdx"}, + }, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is not None + + assert response.json()["data"]["moveService"]["code"] == 404 + assert response.json()["data"]["moveService"]["message"] is not None + assert response.json()["data"]["moveService"]["success"] is False + + +def test_graphql_move_service_volume_not_found( + authorized_client, + mock_get_service_by_id_return_none, + mock_service_to_graphql_service, + mock_block_devices_return_none, + mock_subprocess_popen, + one_user, +): + response = authorized_client.post( + "/graphql", + json={ + "query": API_MOVE_SERVICE_MUTATION, + "variables": { + "input": {"serviceId": "nextcloud", "location": "sdx"}, + }, + }, + ) + + assert response.status_code == 200 + assert response.json().get("data") is not None + + assert response.json()["data"]["moveService"]["code"] == 404 + assert response.json()["data"]["moveService"]["message"] is not None + assert response.json()["data"]["moveService"]["success"] is False + + +def test_graphql_move_service( + authorized_client, + mock_get_service_by_id, + mock_service_to_graphql_service, + mock_block_devices, + mock_subprocess_popen, + one_user, + mock_job_to_api_job, +): + response = authorized_client.post( + "/graphql", + json={ + "query": API_MOVE_SERVICE_MUTATION, + "variables": { + "input": {"serviceId": "nextcloud", "location": "sdx"}, + }, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is not None + + assert response.json()["data"]["moveService"]["code"] == 200 + assert response.json()["data"]["moveService"]["message"] is not None + assert response.json()["data"]["moveService"]["success"] is True diff --git a/tests/test_graphql/test_api_services/one_user.json b/tests/test_graphql/test_api_services/one_user.json new file mode 100644 index 0000000..5df2108 --- /dev/null +++ b/tests/test_graphql/test_api_services/one_user.json @@ -0,0 +1,61 @@ +{ + "backblaze": { + "accountId": "ID", + "accountKey": "KEY", + "bucket": "selfprivacy" + }, + "api": { + "token": "TEST_TOKEN", + "enableSwagger": false + }, + "bitwarden": { + "enable": false + }, + "cloudflare": { + "apiKey": "TOKEN" + }, + "databasePassword": "PASSWORD", + "domain": "test.tld", + "hashedMasterPassword": "HASHED_PASSWORD", + "hostname": "test-instance", + "nextcloud": { + "adminPassword": "ADMIN", + "databasePassword": "ADMIN", + "enable": true + }, + "resticPassword": "PASS", + "ssh": { + "enable": true, + "passwordAuthentication": true, + "rootKeys": [ + "ssh-ed25519 KEY test@pc" + ] + }, + "username": "tester", + "gitea": { + "enable": false + }, + "ocserv": { + "enable": true + }, + "pleroma": { + "enable": true + }, + "autoUpgrade": { + "enable": true, + "allowReboot": true + }, + "timezone": "Europe/Moscow", + "sshKeys": [ + "ssh-rsa KEY test@pc" + ], + "users": [ + { + "username": "user1", + "hashedPassword": "HASHED_PASSWORD_1", + "sshKeys": [ + "ssh-rsa KEY user1@pc" + ] + } + ] +} \ No newline at end of file diff --git a/tests/test_graphql/test_api_storage.py b/tests/test_graphql/test_api_storage.py new file mode 100644 index 0000000..f4acdaa --- /dev/null +++ b/tests/test_graphql/test_api_storage.py @@ -0,0 +1,268 @@ +import pytest + + +class BlockDeviceMockReturnTrue: + """Mock BlockDevices""" + + def __init__(self, *args, **kwargs): + self.args = args + self.kwargs = kwargs + + def mount(self): + return True + + def unmount(self): + return True + + def resize(self): + return True + + returncode = 0 + + +class BlockDevicesMockReturnTrue: + def get_block_device(name: str): # type: ignore + return BlockDeviceMockReturnTrue() + + def __new__(cls, *args, **kwargs): + pass + + def __init__(self): + pass + + +class BlockDevicesMockReturnNone: + def get_block_device(name: str): # type: ignore + return None + + def __new__(cls, *args, **kwargs): + pass + + def __init__(self): + pass + + +@pytest.fixture +def mock_block_devices_return_true(mocker): + mock = mocker.patch( + "selfprivacy_api.graphql.mutations.storage_mutations.BlockDevices", + # "selfprivacy_api.utils.block_devices.BlockDevices", + autospec=True, + return_value=BlockDevicesMockReturnTrue, + ) + return mock + + +@pytest.fixture +def mock_block_devices_return_none(mocker): + mock = mocker.patch( + "selfprivacy_api.utils.block_devices.BlockDevices", + autospec=True, + return_value=BlockDevicesMockReturnNone, + ) + return mock + + +@pytest.fixture +def mock_block_device_return_true(mocker): + mock = mocker.patch( + "selfprivacy_api.utils.block_devices.BlockDevice", + autospec=True, + return_value=BlockDeviceMockReturnTrue, + ) + return mock + + +API_RESIZE_VOLUME_MUTATION = """ +mutation resizeVolume($name: String!) { + resizeVolume(name: $name) { + success + message + code + } +} +""" + + +def test_graphql_resize_volume_unauthorized_client( + client, mock_block_devices_return_true +): + response = client.post( + "/graphql", + json={ + "query": API_RESIZE_VOLUME_MUTATION, + "variables": {"name": "sdx"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is None + + +def test_graphql_resize_volume_nonexistent_block_device( + authorized_client, mock_block_devices_return_none +): + response = authorized_client.post( + "/graphql", + json={ + "query": API_RESIZE_VOLUME_MUTATION, + "variables": {"name": "sdx"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is not None + + assert response.json()["data"]["resizeVolume"]["code"] == 404 + assert response.json()["data"]["resizeVolume"]["message"] is not None + assert response.json()["data"]["resizeVolume"]["success"] is False + + +def test_graphql_resize_volume(authorized_client, mock_block_devices_return_true): + response = authorized_client.post( + "/graphql", + json={ + "query": API_RESIZE_VOLUME_MUTATION, + "variables": {"name": "sdx"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is not None + + assert response.json()["data"]["resizeVolume"]["code"] == 200 + assert response.json()["data"]["resizeVolume"]["message"] is not None + assert response.json()["data"]["resizeVolume"]["success"] is True + + +API_MOUNT_VOLUME_MUTATION = """ +mutation mountVolume($name: String!) { + mountVolume(name: $name) { + success + message + code + } +} +""" + + +def test_graphql_mount_volume_unauthorized_client( + client, mock_block_device_return_true +): + response = client.post( + "/graphql", + json={ + "query": API_MOUNT_VOLUME_MUTATION, + "variables": {"name": "sdx"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is None + + +def test_graphql_mount_already_mounted_volume( + authorized_client, mock_block_devices_return_none +): + response = authorized_client.post( + "/graphql", + json={ + "query": API_MOUNT_VOLUME_MUTATION, + "variables": {"name": "sdx"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is not None + + assert response.json()["data"]["mountVolume"]["code"] == 404 + assert response.json()["data"]["mountVolume"]["message"] is not None + assert response.json()["data"]["mountVolume"]["success"] is False + + +def test_graphql_mount_not_found_volume( + authorized_client, mock_block_devices_return_none +): + response = authorized_client.post( + "/graphql", + json={ + "query": API_MOUNT_VOLUME_MUTATION, + "variables": {"name": "sdx"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is not None + + assert response.json()["data"]["mountVolume"]["code"] == 404 + assert response.json()["data"]["mountVolume"]["message"] is not None + assert response.json()["data"]["mountVolume"]["success"] is False + + +def test_graphql_mount_volume(authorized_client, mock_block_devices_return_true): + response = authorized_client.post( + "/graphql", + json={ + "query": API_MOUNT_VOLUME_MUTATION, + "variables": {"name": "sdx"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is not None + + assert response.json()["data"]["mountVolume"]["code"] == 200 + assert response.json()["data"]["mountVolume"]["message"] is not None + assert response.json()["data"]["mountVolume"]["success"] is True + + +API_UNMOUNT_VOLUME_MUTATION = """ +mutation unmountVolume($name: String!) { + unmountVolume(name: $name) { + success + message + code + } +} +""" + + +def test_graphql_unmount_volume_unauthorized_client( + client, mock_block_devices_return_true +): + response = client.post( + "/graphql", + json={ + "query": API_UNMOUNT_VOLUME_MUTATION, + "variables": {"name": "sdx"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is None + + +def test_graphql_unmount_not_found_volume( + authorized_client, mock_block_devices_return_none +): + response = authorized_client.post( + "/graphql", + json={ + "query": API_UNMOUNT_VOLUME_MUTATION, + "variables": {"name": "sdx"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is not None + + assert response.json()["data"]["unmountVolume"]["code"] == 404 + assert response.json()["data"]["unmountVolume"]["message"] is not None + assert response.json()["data"]["unmountVolume"]["success"] is False + + +def test_graphql_unmount_volume(authorized_client, mock_block_devices_return_true): + response = authorized_client.post( + "/graphql", + json={ + "query": API_UNMOUNT_VOLUME_MUTATION, + "variables": {"name": "sdx"}, + }, + ) + assert response.status_code == 200 + assert response.json().get("data") is not None + + assert response.json()["data"]["unmountVolume"]["code"] == 200 + assert response.json()["data"]["unmountVolume"]["message"] is not None + assert response.json()["data"]["unmountVolume"]["success"] is True