selfprivacy-rest-api/tests/test_graphql/test_services.py

import pytest
import shutil

from typing import Generator
from os import mkdir

from selfprivacy_api.utils.block_devices import BlockDevices

import selfprivacy_api.services as service_module
from selfprivacy_api.services import get_service_by_id
from selfprivacy_api.services.service import Service, ServiceStatus
from selfprivacy_api.services.test_service import DummyService

from tests.common import generate_service_query
from tests.test_graphql.common import assert_empty, assert_ok, get_data
from tests.test_block_device_utils import lsblk_singular_mock


LSBLK_BLOCKDEVICES_DICTS = [
    {
        "name": "sda1",
        "path": "/dev/sda1",
        "fsavail": "4614107136",
        "fssize": "19814920192",
        "fstype": "ext4",
        "fsused": "14345314304",
        "mountpoints": ["/nix/store", "/"],
        "label": None,
        "uuid": "ec80c004-baec-4a2c-851d-0e1807135511",
        "size": 20210236928,
        "model": None,
        "serial": None,
        "type": "part",
    },
    {
        "name": "sda2",
        "path": "/dev/sda2",
        "fsavail": "4614107136",
        "fssize": "19814920192",
        "fstype": "ext4",
        "fsused": "14345314304",
        "mountpoints": ["/home"],
        "label": None,
        "uuid": "deadbeef-baec-4a2c-851d-0e1807135511",
        "size": 20210236928,
        "model": None,
        "serial": None,
        "type": "part",
    },
]


@pytest.fixture()
def mock_lsblk_devices(mocker):
    mock = mocker.patch(
        "selfprivacy_api.utils.block_devices.BlockDevices.lsblk_device_dicts",
        autospec=True,
        return_value=LSBLK_BLOCKDEVICES_DICTS,
    )
    BlockDevices().update()
    assert BlockDevices().lsblk_device_dicts() == LSBLK_BLOCKDEVICES_DICTS
    devices = BlockDevices().get_block_devices()

    assert len(devices) == 2

    names = [device.name for device in devices]
    assert "sda1" in names
    assert "sda2" in names
    return mock


@pytest.fixture()
def dummy_service_with_binds(dummy_service, mock_lsblk_devices, volume_folders):
    binds = dummy_service.binds()
    for bind in binds:
        path = bind.binding_path
        shutil.move(bind.binding_path, bind.location_at_volume())
        mkdir(bind.binding_path)

        bind.ensure_ownership()
        bind.validate()

        bind.bind()
    return dummy_service


@pytest.fixture()
def only_dummy_service(dummy_service) -> Generator[DummyService, None, None]:
    # because queries to services that are not really there error out
    back_copy = service_module.services.copy()
    service_module.services.clear()
    service_module.services.append(dummy_service)
    yield dummy_service
    service_module.services.clear()
    service_module.services.extend(back_copy)


@pytest.fixture()
def mock_check_volume(mocker):
    mock = mocker.patch(
        "selfprivacy_api.services.service.check_volume",
        autospec=True,
        return_value=None,
    )
    return mock


API_START_MUTATION = """
mutation TestStartService($service_id: String!) {
    services {
        startService(serviceId: $service_id) {
            success
            message
            code
            service {
                id
                status
            }
        }
    }
}
"""

API_RESTART_MUTATION = """
mutation TestRestartService($service_id: String!) {
    services {
        restartService(serviceId: $service_id) {
            success
            message
            code
            service {
                id
                status
            }
        }
    }
}
"""

API_ENABLE_MUTATION = """
mutation TestStartService($service_id: String!) {
    services {
        enableService(serviceId: $service_id) {
            success
            message
            code
            service {
                id
                isEnabled
            }
        }
    }
}
"""
API_DISABLE_MUTATION = """
mutation TestStartService($service_id: String!) {
    services {
        disableService(serviceId: $service_id) {
            success
            message
            code
            service {
                id
                isEnabled
            }
        }
    }
}
"""

API_STOP_MUTATION = """
mutation TestStopService($service_id: String!) {
    services {
        stopService(serviceId: $service_id) {
            success
            message
            code
            service {
                id
                status
            }
        }
    }
}

"""
API_SERVICES_QUERY = """
allServices {
    id
    status
    isEnabled
}
"""

API_MOVE_MUTATION = """
mutation TestMoveService($input: MoveServiceInput!) {
    services {
        moveService(input: $input) {
            success
            message
            code
            job {
                uid
                status
            }
            service {
                id
                status
            }
        }
    }
}
"""


def assert_notfound(data):
    assert_errorcode(data, 404)


def assert_errorcode(data, errorcode):
    assert data["code"] == errorcode
    assert data["success"] is False
    assert data["message"] is not None


def api_enable(client, service: Service) -> dict:
    return api_enable_by_name(client, service.get_id())


def api_enable_by_name(client, service_id: str) -> dict:
    response = client.post(
        "/graphql",
        json={
            "query": API_ENABLE_MUTATION,
            "variables": {"service_id": service_id},
        },
    )
    return response


def api_disable(client, service: Service) -> dict:
    return api_disable_by_name(client, service.get_id())


def api_disable_by_name(client, service_id: str) -> dict:
    response = client.post(
        "/graphql",
        json={
            "query": API_DISABLE_MUTATION,
            "variables": {"service_id": service_id},
        },
    )
    return response


def api_start(client, service: Service) -> dict:
    return api_start_by_name(client, service.get_id())


def api_start_by_name(client, service_id: str) -> dict:
    response = client.post(
        "/graphql",
        json={
            "query": API_START_MUTATION,
            "variables": {"service_id": service_id},
        },
    )
    return response


def api_move(client, service: Service, location: str) -> dict:
    return api_move_by_name(client, service.get_id(), location)


def api_move_by_name(client, service_id: str, location: str) -> dict:
    response = client.post(
        "/graphql",
        json={
            "query": API_MOVE_MUTATION,
            "variables": {
                "input": {
                    "serviceId": service_id,
                    "location": location,
                }
            },
        },
    )
    return response


def api_restart(client, service: Service) -> dict:
    return api_restart_by_name(client, service.get_id())


def api_restart_by_name(client, service_id: str) -> dict:
    response = client.post(
        "/graphql",
        json={
            "query": API_RESTART_MUTATION,
            "variables": {"service_id": service_id},
        },
    )
    return response


def api_stop(client, service: Service) -> dict:
    return api_stop_by_name(client, service.get_id())


def api_stop_by_name(client, service_id: str) -> dict:
    response = client.post(
        "/graphql",
        json={
            "query": API_STOP_MUTATION,
            "variables": {"service_id": service_id},
        },
    )
    return response


def api_all_services(authorized_client):
    response = api_all_services_raw(authorized_client)
    data = get_data(response)
    result = data["services"]["allServices"]
    assert result is not None
    return result


def api_all_services_raw(client):
    return client.post(
        "/graphql",
        json={"query": generate_service_query([API_SERVICES_QUERY])},
    )


def api_service(authorized_client, service: Service):
    id = service.get_id()
    for _service in api_all_services(authorized_client):
        if _service["id"] == id:
            return _service


def test_get_services(authorized_client, only_dummy_service):
    services = api_all_services(authorized_client)
    assert len(services) == 1

    api_dummy_service = services[0]
    assert api_dummy_service["id"] == "testservice"
    assert api_dummy_service["status"] == ServiceStatus.ACTIVE.value
    assert api_dummy_service["isEnabled"] is True


def test_enable_return_value(authorized_client, only_dummy_service):
    dummy_service = only_dummy_service
    mutation_response = api_enable(authorized_client, dummy_service)
    data = get_data(mutation_response)["services"]["enableService"]
    assert_ok(data)
    service = data["service"]
    assert service["id"] == dummy_service.get_id()
    assert service["isEnabled"] == True


def test_disable_return_value(authorized_client, only_dummy_service):
    dummy_service = only_dummy_service
    mutation_response = api_disable(authorized_client, dummy_service)
    data = get_data(mutation_response)["services"]["disableService"]
    assert_ok(data)
    service = data["service"]
    assert service["id"] == dummy_service.get_id()
    assert service["isEnabled"] == False


def test_start_return_value(authorized_client, only_dummy_service):
    dummy_service = only_dummy_service
    mutation_response = api_start(authorized_client, dummy_service)
    data = get_data(mutation_response)["services"]["startService"]
    assert_ok(data)
    service = data["service"]
    assert service["id"] == dummy_service.get_id()
    assert service["status"] == ServiceStatus.ACTIVE.value


def test_restart(authorized_client, only_dummy_service):
    dummy_service = only_dummy_service
    dummy_service.set_delay(0.3)
    mutation_response = api_restart(authorized_client, dummy_service)
    data = get_data(mutation_response)["services"]["restartService"]
    assert_ok(data)
    service = data["service"]
    assert service["id"] == dummy_service.get_id()
    assert service["status"] == ServiceStatus.RELOADING.value


def test_stop_return_value(authorized_client, only_dummy_service):
    dummy_service = only_dummy_service
    mutation_response = api_stop(authorized_client, dummy_service)
    data = get_data(mutation_response)["services"]["stopService"]
    assert_ok(data)
    service = data["service"]
    assert service["id"] == dummy_service.get_id()
    assert service["status"] == ServiceStatus.INACTIVE.value


def test_allservices_unauthorized(client, only_dummy_service):
    dummy_service = only_dummy_service
    response = api_all_services_raw(client)

    assert response.status_code == 200
    assert response.json().get("data") is None


def test_start_unauthorized(client, only_dummy_service):
    dummy_service = only_dummy_service
    response = api_start(client, dummy_service)
    assert_empty(response)


def test_restart_unauthorized(client, only_dummy_service):
    dummy_service = only_dummy_service
    response = api_restart(client, dummy_service)
    assert_empty(response)


def test_stop_unauthorized(client, only_dummy_service):
    dummy_service = only_dummy_service
    response = api_stop(client, dummy_service)
    assert_empty(response)


def test_enable_unauthorized(client, only_dummy_service):
    dummy_service = only_dummy_service
    response = api_enable(client, dummy_service)
    assert_empty(response)


def test_disable_unauthorized(client, only_dummy_service):
    dummy_service = only_dummy_service
    response = api_disable(client, dummy_service)
    assert_empty(response)


def test_move_unauthorized(client, only_dummy_service):
    dummy_service = only_dummy_service
    response = api_move(client, dummy_service, "sda1")
    assert_empty(response)


def test_start_nonexistent(authorized_client, only_dummy_service):
    dummy_service = only_dummy_service
    mutation_response = api_start_by_name(authorized_client, "bogus_service")
    data = get_data(mutation_response)["services"]["startService"]
    assert_notfound(data)

    assert data["service"] is None


def test_restart_nonexistent(authorized_client, only_dummy_service):
    dummy_service = only_dummy_service
    mutation_response = api_restart_by_name(authorized_client, "bogus_service")
    data = get_data(mutation_response)["services"]["restartService"]
    assert_notfound(data)

    assert data["service"] is None


def test_stop_nonexistent(authorized_client, only_dummy_service):
    dummy_service = only_dummy_service
    mutation_response = api_stop_by_name(authorized_client, "bogus_service")
    data = get_data(mutation_response)["services"]["stopService"]
    assert_notfound(data)

    assert data["service"] is None


def test_enable_nonexistent(authorized_client, only_dummy_service):
    dummy_service = only_dummy_service
    mutation_response = api_enable_by_name(authorized_client, "bogus_service")
    data = get_data(mutation_response)["services"]["enableService"]
    assert_notfound(data)

    assert data["service"] is None


def test_disable_nonexistent(authorized_client, only_dummy_service):
    dummy_service = only_dummy_service
    mutation_response = api_disable_by_name(authorized_client, "bogus_service")
    data = get_data(mutation_response)["services"]["disableService"]
    assert_notfound(data)

    assert data["service"] is None


def test_stop_start(authorized_client, only_dummy_service):
    dummy_service = only_dummy_service

    api_dummy_service = api_all_services(authorized_client)[0]
    assert api_dummy_service["status"] == ServiceStatus.ACTIVE.value

    # attempting to start an already started service
    api_start(authorized_client, dummy_service)
    api_dummy_service = api_all_services(authorized_client)[0]
    assert api_dummy_service["status"] == ServiceStatus.ACTIVE.value

    api_stop(authorized_client, dummy_service)
    api_dummy_service = api_all_services(authorized_client)[0]
    assert api_dummy_service["status"] == ServiceStatus.INACTIVE.value

    # attempting to stop an already stopped service
    api_stop(authorized_client, dummy_service)
    api_dummy_service = api_all_services(authorized_client)[0]
    assert api_dummy_service["status"] == ServiceStatus.INACTIVE.value

    api_start(authorized_client, dummy_service)
    api_dummy_service = api_all_services(authorized_client)[0]
    assert api_dummy_service["status"] == ServiceStatus.ACTIVE.value


def test_disable_enable(authorized_client, only_dummy_service):
    dummy_service = only_dummy_service

    api_dummy_service = api_all_services(authorized_client)[0]
    assert api_dummy_service["isEnabled"] is True

    # attempting to enable an already enableed service
    api_enable(authorized_client, dummy_service)
    api_dummy_service = api_all_services(authorized_client)[0]
    assert api_dummy_service["isEnabled"] is True
    assert api_dummy_service["status"] == ServiceStatus.ACTIVE.value

    api_disable(authorized_client, dummy_service)
    api_dummy_service = api_all_services(authorized_client)[0]
    assert api_dummy_service["isEnabled"] is False
    assert api_dummy_service["status"] == ServiceStatus.ACTIVE.value

    # attempting to disable an already disableped service
    api_disable(authorized_client, dummy_service)
    api_dummy_service = api_all_services(authorized_client)[0]
    assert api_dummy_service["isEnabled"] is False
    assert api_dummy_service["status"] == ServiceStatus.ACTIVE.value

    api_enable(authorized_client, dummy_service)
    api_dummy_service = api_all_services(authorized_client)[0]
    assert api_dummy_service["isEnabled"] is True
    assert api_dummy_service["status"] == ServiceStatus.ACTIVE.value


def test_move_immovable(authorized_client, only_dummy_service):
    dummy_service = only_dummy_service
    dummy_service.set_movable(False)
    root = BlockDevices().get_root_block_device()
    mutation_response = api_move(authorized_client, dummy_service, root.name)
    data = get_data(mutation_response)["services"]["moveService"]
    assert_errorcode(data, 400)
    try:
        assert "not movable" in data["message"]
    except AssertionError:
        raise ValueError("wrong type of error?: ", data["message"])

    # is there a meaning in returning the service in this?
    assert data["service"] is not None
    assert data["job"] is None


def test_move_no_such_service(authorized_client, only_dummy_service):
    mutation_response = api_move_by_name(authorized_client, "bogus_service", "sda1")
    data = get_data(mutation_response)["services"]["moveService"]
    assert_errorcode(data, 404)

    assert data["service"] is None
    assert data["job"] is None


def test_move_no_such_volume(authorized_client, only_dummy_service):
    dummy_service = only_dummy_service
    mutation_response = api_move(authorized_client, dummy_service, "bogus_volume")
    data = get_data(mutation_response)["services"]["moveService"]
    assert_notfound(data)

    assert data["service"] is None
    assert data["job"] is None


def test_move_same_volume(authorized_client, dummy_service):
    # dummy_service = only_dummy_service

    # we need a drive that actually exists
    root_volume = BlockDevices().get_root_block_device()
    dummy_service.set_simulated_moves(False)
    dummy_service.set_drive(root_volume.name)

    mutation_response = api_move(authorized_client, dummy_service, root_volume.name)
    data = get_data(mutation_response)["services"]["moveService"]
    assert_errorcode(data, 400)

    # is there a meaning in returning the service in this?
    assert data["service"] is not None
    # We do not create a job if task is not created
    assert data["job"] is None


def test_graphql_move_service_without_folders_on_old_volume(
    authorized_client,
    generic_userdata,
    mock_lsblk_devices,
    dummy_service: DummyService,
):
    target = "sda1"
    BlockDevices().update()
    assert BlockDevices().get_block_device(target) is not None

    dummy_service.set_simulated_moves(False)
    dummy_service.set_drive("sda2")
    mutation_response = api_move(authorized_client, dummy_service, target)

    data = get_data(mutation_response)["services"]["moveService"]
    assert_errorcode(data, 400)
    assert "sda2/test_service is not found" in data["message"]


def test_graphql_move_service(
    authorized_client,
    generic_userdata,
    mock_check_volume,
    dummy_service_with_binds,
):
    dummy_service = dummy_service_with_binds

    origin = "sda1"
    target = "sda2"
    assert BlockDevices().get_block_device(target) is not None
    assert BlockDevices().get_block_device(origin) is not None

    dummy_service.set_drive(origin)
    dummy_service.set_simulated_moves(False)

    mutation_response = api_move(authorized_client, dummy_service, target)

    data = get_data(mutation_response)["services"]["moveService"]
    assert_ok(data)


def test_mailservice_cannot_enable_disable(authorized_client):
    mailservice = get_service_by_id("simple-nixos-mailserver")

    mutation_response = api_enable(authorized_client, mailservice)
    data = get_data(mutation_response)["services"]["enableService"]
    assert_errorcode(data, 400)
    # TODO?: we cannot convert mailservice to graphql Service without /var/domain yet
    # assert data["service"] is not None

    mutation_response = api_disable(authorized_client, mailservice)
    data = get_data(mutation_response)["services"]["disableService"]
    assert_errorcode(data, 400)
    # assert data["service"] is not None