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

refactor(api): Group mutations 

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.
pull/35/head
Inex Code 2023-06-21 06:46:56 +03:00
parent e3a87f1d98
commit c6919293b6
18 changed files with 1023 additions and 565 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
selfprivacy_api/backup/providers/local_file.py
View File

@ -4,4 +4,4 @@ from selfprivacy_api.backup.backuppers.restic_backupper import ResticBackuper
class LocalFileBackup(AbstractBackupProvider):
backuper = ResticBackuper("", "", ":local:")
name = "FILE"
name = "FILE"

14
selfprivacy_api/graphql/mutations/backup_mutations.py
View File

@ -6,7 +6,7 @@ from strawberry.types import Info
from selfprivacy_api.graphql import IsAuthenticated
from selfprivacy_api.graphql.mutations.mutation_interface import (
GenericMutationReturn,
GenericJobButationReturn,
GenericJobMutationReturn,
MutationReturnInterface,
)
from selfprivacy_api.graphql.queries.backup import BackupConfiguration
@ -83,12 +83,12 @@ class BackupMutations:
)
@strawberry.mutation(permission_classes=[IsAuthenticated])
def start_backup(self, service_id: str) -> GenericJobButationReturn:
def start_backup(self, service_id: str) -> GenericJobMutationReturn:
"""Start backup"""
service = get_service_by_id(service_id)
if service is None:
return GenericJobButationReturn(
return GenericJobMutationReturn(
success=False,
code=300,
message=f"nonexistent service: {service_id}",
@ -99,7 +99,7 @@ class BackupMutations:
start_backup(service)
job = job_to_api_job(job)
return GenericJobButationReturn(
return GenericJobMutationReturn(
success=True,
code=200,
message="Backup job queued",
@ -107,12 +107,12 @@ class BackupMutations:
)
@strawberry.mutation(permission_classes=[IsAuthenticated])
def restore_backup(self, snapshot_id: str) -> GenericJobButationReturn:
def restore_backup(self, snapshot_id: str) -> GenericJobMutationReturn:
"""Restore backup"""
snap = Backups.get_snapshot_by_id(snapshot_id)
service = get_service_by_id(snap.service_name)
if snap is None:
return GenericJobButationReturn(
return GenericJobMutationReturn(
success=False,
code=400,
message=f"No such snapshot: {snapshot_id}",
@ -122,7 +122,7 @@ class BackupMutations:
job = add_restore_job(snap)
restore_snapshot(snap)
return GenericJobButationReturn(
return GenericJobMutationReturn(
success=True,
code=200,
message="restore job created",

215
selfprivacy_api/graphql/mutations/deprecated_mutations.py Normal file
View File

@ -0,0 +1,215 @@
"""Deprecated mutations
There was made a mistake, where mutations were not grouped, and were instead
placed in the root of mutations schema. In this file, we import all the
mutations from and provide them to the root for backwards compatibility.
"""
import strawberry
from selfprivacy_api.graphql import IsAuthenticated
from selfprivacy_api.graphql.common_types.user import UserMutationReturn
from selfprivacy_api.graphql.mutations.api_mutations import (
ApiKeyMutationReturn,
ApiMutations,
DeviceApiTokenMutationReturn,
)
from selfprivacy_api.graphql.mutations.backup_mutations import BackupMutations
from selfprivacy_api.graphql.mutations.job_mutations import JobMutations
from selfprivacy_api.graphql.mutations.mutation_interface import (
GenericJobMutationReturn,
GenericMutationReturn,
)
from selfprivacy_api.graphql.mutations.services_mutations import (
ServiceMutationReturn,
ServicesMutations,
)
from selfprivacy_api.graphql.mutations.storage_mutations import StorageMutations
from selfprivacy_api.graphql.mutations.system_mutations import (
AutoUpgradeSettingsMutationReturn,
SystemMutations,
TimezoneMutationReturn,
)
from selfprivacy_api.graphql.mutations.backup_mutations import BackupMutations
from selfprivacy_api.graphql.mutations.users_mutations import UsersMutations
def deprecated_mutation(func, group, auth=True):
return strawberry.mutation(
resolver=func,
permission_classes=[IsAuthenticated] if auth else [],
deprecation_reason=f"Use `{group}.{func.__name__}` instead",
)
@strawberry.type
class DeprecatedApiMutations:
get_new_recovery_api_key: ApiKeyMutationReturn = deprecated_mutation(
ApiMutations.get_new_recovery_api_key,
"api",
)
use_recovery_api_key: DeviceApiTokenMutationReturn = deprecated_mutation(
ApiMutations.use_recovery_api_key,
"api",
auth=False,
)
refresh_device_api_token: DeviceApiTokenMutationReturn = deprecated_mutation(
ApiMutations.refresh_device_api_token,
"api",
)
delete_device_api_token: GenericMutationReturn = deprecated_mutation(
ApiMutations.delete_device_api_token,
"api",
)
get_new_device_api_key: ApiKeyMutationReturn = deprecated_mutation(
ApiMutations.get_new_device_api_key,
"api",
)
invalidate_new_device_api_key: GenericMutationReturn = deprecated_mutation(
ApiMutations.invalidate_new_device_api_key,
"api",
)
authorize_with_new_device_api_key: DeviceApiTokenMutationReturn = (
deprecated_mutation(
ApiMutations.authorize_with_new_device_api_key,
"api",
auth=False,
)
)
@strawberry.type
class DeprecatedSystemMutations:
change_timezone: TimezoneMutationReturn = deprecated_mutation(
SystemMutations.change_timezone,
"system",
)
change_auto_upgrade_settings: AutoUpgradeSettingsMutationReturn = (
deprecated_mutation(
SystemMutations.change_auto_upgrade_settings,
"system",
)
)
run_system_rebuild: GenericMutationReturn = deprecated_mutation(
SystemMutations.run_system_rebuild,
"system",
)
run_system_rollback: GenericMutationReturn = deprecated_mutation(
SystemMutations.run_system_rollback,
"system",
)
run_system_upgrade: GenericMutationReturn = deprecated_mutation(
SystemMutations.run_system_upgrade,
"system",
)
reboot_system: GenericMutationReturn = deprecated_mutation(
SystemMutations.reboot_system,
"system",
)
pull_repository_changes: GenericMutationReturn = deprecated_mutation(
SystemMutations.pull_repository_changes,
"system",
)
@strawberry.type
class DeprecatedUsersMutations:
create_user: UserMutationReturn = deprecated_mutation(
UsersMutations.create_user,
"users",
)
delete_user: GenericMutationReturn = deprecated_mutation(
UsersMutations.delete_user,
"users",
)
update_user: UserMutationReturn = deprecated_mutation(
UsersMutations.update_user,
"users",
)
add_ssh_key: UserMutationReturn = deprecated_mutation(
UsersMutations.add_ssh_key,
"users",
)
remove_ssh_key: UserMutationReturn = deprecated_mutation(
UsersMutations.remove_ssh_key,
"users",
)
@strawberry.type
class DeprecatedStorageMutations:
resize_volume: GenericMutationReturn = deprecated_mutation(
StorageMutations.resize_volume,
"storage",
)
mount_volume: GenericMutationReturn = deprecated_mutation(
StorageMutations.mount_volume,
"storage",
)
unmount_volume: GenericMutationReturn = deprecated_mutation(
StorageMutations.unmount_volume,
"storage",
)
migrate_to_binds: GenericJobMutationReturn = deprecated_mutation(
StorageMutations.migrate_to_binds,
"storage",
)
@strawberry.type
class DeprecatedServicesMutations:
enable_service: ServiceMutationReturn = deprecated_mutation(
ServicesMutations.enable_service,
"services",
)
disable_service: ServiceMutationReturn = deprecated_mutation(
ServicesMutations.disable_service,
"services",
)
stop_service: ServiceMutationReturn = deprecated_mutation(
ServicesMutations.stop_service,
"services",
)
start_service: ServiceMutationReturn = deprecated_mutation(
ServicesMutations.start_service,
"services",
)
restart_service: ServiceMutationReturn = deprecated_mutation(
ServicesMutations.restart_service,
"services",
)
move_service: ServiceMutationReturn = deprecated_mutation(
ServicesMutations.move_service,
"services",
)
@strawberry.type
class DeprecatedJobMutations:
remove_job: GenericMutationReturn = deprecated_mutation(
JobMutations.remove_job,
"jobs",
)

2
selfprivacy_api/graphql/mutations/mutation_interface.py
View File

@ -17,5 +17,5 @@ class GenericMutationReturn(MutationReturnInterface):
@strawberry.type
class GenericJobButationReturn(MutationReturnInterface):
class GenericJobMutationReturn(MutationReturnInterface):
job: typing.Optional[ApiJob] = None

4
selfprivacy_api/graphql/mutations/services_mutations.py
View File

@ -10,7 +10,7 @@ from selfprivacy_api.graphql.common_types.service import (
service_to_graphql_service,
)
from selfprivacy_api.graphql.mutations.mutation_interface import (
GenericJobButationReturn,
GenericJobMutationReturn,
GenericMutationReturn,
)
@ -34,7 +34,7 @@ class MoveServiceInput:
@strawberry.type
class ServiceJobMutationReturn(GenericJobButationReturn):
class ServiceJobMutationReturn(GenericJobMutationReturn):
"""Service job mutation return type."""
service: typing.Optional[Service] = None

102
selfprivacy_api/graphql/mutations/ssh_mutations.py
View File

@ -1,102 +0,0 @@
#!/usr/bin/env python3
"""Users management module"""
# pylint: disable=too-few-public-methods
import strawberry
from selfprivacy_api.actions.users import UserNotFound
from selfprivacy_api.graphql import IsAuthenticated
from selfprivacy_api.actions.ssh import (
InvalidPublicKey,
KeyAlreadyExists,
KeyNotFound,
create_ssh_key,
remove_ssh_key,
)
from selfprivacy_api.graphql.common_types.user import (
UserMutationReturn,
get_user_by_username,
)
@strawberry.input
class SshMutationInput:
"""Input type for ssh mutation"""
username: str
ssh_key: str
@strawberry.type
class SshMutations:
"""Mutations ssh"""
@strawberry.mutation(permission_classes=[IsAuthenticated])
def add_ssh_key(self, ssh_input: SshMutationInput) -> UserMutationReturn:
"""Add a new ssh key"""
try:
create_ssh_key(ssh_input.username, ssh_input.ssh_key)
except KeyAlreadyExists:
return UserMutationReturn(
success=False,
message="Key already exists",
code=409,
)
except InvalidPublicKey:
return UserMutationReturn(
success=False,
message="Invalid key type. Only ssh-ed25519 and ssh-rsa are supported",
code=400,
)
except UserNotFound:
return UserMutationReturn(
success=False,
message="User not found",
code=404,
)
except Exception as e:
return UserMutationReturn(
success=False,
message=str(e),
code=500,
)
return UserMutationReturn(
success=True,
message="New SSH key successfully written",
code=201,
user=get_user_by_username(ssh_input.username),
)
@strawberry.mutation(permission_classes=[IsAuthenticated])
def remove_ssh_key(self, ssh_input: SshMutationInput) -> UserMutationReturn:
"""Remove ssh key from user"""
try:
remove_ssh_key(ssh_input.username, ssh_input.ssh_key)
except KeyNotFound:
return UserMutationReturn(
success=False,
message="Key not found",
code=404,
)
except UserNotFound:
return UserMutationReturn(
success=False,
message="User not found",
code=404,
)
except Exception as e:
return UserMutationReturn(
success=False,
message=str(e),
code=500,
)
return UserMutationReturn(
success=True,
message="SSH key successfully removed",
code=200,
user=get_user_by_username(ssh_input.username),
)

8
selfprivacy_api/graphql/mutations/storage_mutations.py
View File

@ -4,7 +4,7 @@ from selfprivacy_api.graphql import IsAuthenticated
from selfprivacy_api.graphql.common_types.jobs import job_to_api_job
from selfprivacy_api.utils.block_devices import BlockDevices
from selfprivacy_api.graphql.mutations.mutation_interface import (
GenericJobButationReturn,
GenericJobMutationReturn,
GenericMutationReturn,
)
from selfprivacy_api.jobs.migrate_to_binds import (
@ -79,10 +79,10 @@ class StorageMutations:
)
@strawberry.mutation(permission_classes=[IsAuthenticated])
def migrate_to_binds(self, input: MigrateToBindsInput) -> GenericJobButationReturn:
def migrate_to_binds(self, input: MigrateToBindsInput) -> GenericJobMutationReturn:
"""Migrate to binds"""
if is_bind_migrated():
return GenericJobButationReturn(
return GenericJobMutationReturn(
success=False, code=409, message="Already migrated to binds"
)
job = start_bind_migration(
@ -94,7 +94,7 @@ class StorageMutations:
pleroma_block_device=input.pleroma_block_device,
)
)
return GenericJobButationReturn(
return GenericJobMutationReturn(
success=True,
code=200,
message="Migration to binds started, rebuild the system to apply changes",

88
selfprivacy_api/graphql/mutations/users_mutations.py
View File

@ -3,10 +3,18 @@
# pylint: disable=too-few-public-methods
import strawberry
from selfprivacy_api.graphql import IsAuthenticated
from selfprivacy_api.actions.users import UserNotFound
from selfprivacy_api.graphql.common_types.user import (
UserMutationReturn,
get_user_by_username,
)
from selfprivacy_api.actions.ssh import (
InvalidPublicKey,
KeyAlreadyExists,
KeyNotFound,
create_ssh_key,
remove_ssh_key,
)
from selfprivacy_api.graphql.mutations.mutation_interface import (
GenericMutationReturn,
)
@ -21,8 +29,16 @@ class UserMutationInput:
password: str
@strawberry.input
class SshMutationInput:
"""Input type for ssh mutation"""
username: str
ssh_key: str
@strawberry.type
class UserMutations:
class UsersMutations:
"""Mutations change user settings"""
@strawberry.mutation(permission_classes=[IsAuthenticated])
@ -115,3 +131,73 @@ class UserMutations:
code=200,
user=get_user_by_username(user.username),
)
@strawberry.mutation(permission_classes=[IsAuthenticated])
def add_ssh_key(self, ssh_input: SshMutationInput) -> UserMutationReturn:
"""Add a new ssh key"""
try:
create_ssh_key(ssh_input.username, ssh_input.ssh_key)
except KeyAlreadyExists:
return UserMutationReturn(
success=False,
message="Key already exists",
code=409,
)
except InvalidPublicKey:
return UserMutationReturn(
success=False,
message="Invalid key type. Only ssh-ed25519 and ssh-rsa are supported",
code=400,
)
except UserNotFound:
return UserMutationReturn(
success=False,
message="User not found",
code=404,
)
except Exception as e:
return UserMutationReturn(
success=False,
message=str(e),
code=500,
)
return UserMutationReturn(
success=True,
message="New SSH key successfully written",
code=201,
user=get_user_by_username(ssh_input.username),
)
@strawberry.mutation(permission_classes=[IsAuthenticated])
def remove_ssh_key(self, ssh_input: SshMutationInput) -> UserMutationReturn:
"""Remove ssh key from user"""
try:
remove_ssh_key(ssh_input.username, ssh_input.ssh_key)
except KeyNotFound:
return UserMutationReturn(
success=False,
message="Key not found",
code=404,
)
except UserNotFound:
return UserMutationReturn(
success=False,
message="User not found",
code=404,
)
except Exception as e:
return UserMutationReturn(
success=False,
message=str(e),
code=500,
)
return UserMutationReturn(
success=True,
message="SSH key successfully removed",
code=200,
user=get_user_by_username(ssh_input.username),
)

76
selfprivacy_api/graphql/schema.py
View File

@ -7,10 +7,17 @@ import strawberry
from selfprivacy_api.graphql import IsAuthenticated
from selfprivacy_api.graphql.mutations.api_mutations import ApiMutations
from selfprivacy_api.graphql.mutations.backup_mutations import BackupMutations
from selfprivacy_api.graphql.mutations.deprecated_mutations import (
DeprecatedApiMutations,
DeprecatedJobMutations,
DeprecatedServicesMutations,
DeprecatedStorageMutations,
DeprecatedSystemMutations,
DeprecatedUsersMutations,
)
from selfprivacy_api.graphql.mutations.job_mutations import JobMutations
from selfprivacy_api.graphql.mutations.mutation_interface import GenericMutationReturn
from selfprivacy_api.graphql.mutations.services_mutations import ServicesMutations
from selfprivacy_api.graphql.mutations.ssh_mutations import SshMutations
from selfprivacy_api.graphql.mutations.storage_mutations import StorageMutations
from selfprivacy_api.graphql.mutations.system_mutations import SystemMutations
from selfprivacy_api.graphql.mutations.backup_mutations import BackupMutations
@ -23,7 +30,7 @@ from selfprivacy_api.graphql.queries.storage import Storage
from selfprivacy_api.graphql.queries.system import System
from selfprivacy_api.graphql.queries.backup import Backup
from selfprivacy_api.graphql.mutations.users_mutations import UserMutations
from selfprivacy_api.graphql.mutations.users_mutations import UsersMutations
from selfprivacy_api.graphql.queries.users import Users
from selfprivacy_api.jobs.test import test_job
@ -32,16 +39,16 @@ from selfprivacy_api.jobs.test import test_job
class Query:
"""Root schema for queries"""
@strawberry.field(permission_classes=[IsAuthenticated])
def system(self) -> System:
"""System queries"""
return System()
@strawberry.field
def api(self) -> Api:
"""API access status"""
return Api()
@strawberry.field(permission_classes=[IsAuthenticated])
def system(self) -> System:
"""System queries"""
return System()
@strawberry.field(permission_classes=[IsAuthenticated])
def users(self) -> Users:
"""Users queries"""
@ -70,17 +77,50 @@ class Query:
@strawberry.type
class Mutation(
ApiMutations,
SystemMutations,
UserMutations,
SshMutations,
StorageMutations,
ServicesMutations,
JobMutations,
BackupMutations,
DeprecatedApiMutations,
DeprecatedSystemMutations,
DeprecatedUsersMutations,
DeprecatedStorageMutations,
DeprecatedServicesMutations,
DeprecatedJobMutations,
):
"""Root schema for mutations"""
@strawberry.field
def api(self) -> ApiMutations:
"""API mutations"""
return ApiMutations()
@strawberry.field(permission_classes=[IsAuthenticated])
def system(self) -> SystemMutations:
"""System mutations"""
return SystemMutations()
@strawberry.field(permission_classes=[IsAuthenticated])
def users(self) -> UsersMutations:
"""Users mutations"""
return UsersMutations()
@strawberry.field(permission_classes=[IsAuthenticated])
def storage(self) -> StorageMutations:
"""Storage mutations"""
return StorageMutations()
@strawberry.field(permission_classes=[IsAuthenticated])
def services(self) -> ServicesMutations:
"""Services mutations"""
return ServicesMutations()
@strawberry.field(permission_classes=[IsAuthenticated])
def jobs(self) -> JobMutations:
"""Jobs mutations"""
return JobMutations()
@strawberry.field(permission_classes=[IsAuthenticated])
def backup(self) -> BackupMutations:
"""Backup mutations"""
return BackupMutations()
@strawberry.mutation(permission_classes=[IsAuthenticated])
def test_mutation(self) -> GenericMutationReturn:
"""Test mutation"""
@ -105,4 +145,8 @@ class Subscription:
await asyncio.sleep(0.5)
schema = strawberry.Schema(query=Query, mutation=Mutation, subscription=Subscription)
schema = strawberry.Schema(
query=Query,
mutation=Mutation,
subscription=Subscription,
)

46
tests/test_graphql/test_api_backup.py
View File

@ -8,21 +8,24 @@ from selfprivacy_api.jobs import Jobs, JobStatus
API_RELOAD_SNAPSHOTS = """
mutation TestSnapshotsReload {
backup {
forceSnapshotsReload {
success
message
code
}
}
}
"""
API_SET_AUTOBACKUP_PERIOD_MUTATION = """
mutation TestAutobackupPeriod($period: Int) {
backup {
setAutobackupPeriod(period: $period) {
success
message
code
configuration {
configuration {
provider
encryptionKey
isInitialized
@ -31,16 +34,18 @@ mutation TestAutobackupPeriod($period: Int) {
locationId
}
}
}
}
"""
API_REMOVE_REPOSITORY_MUTATION = """
mutation TestRemoveRepo {
backup {
removeRepository {
success
message
code
configuration {
configuration {
provider
encryptionKey
isInitialized
@ -49,16 +54,18 @@ mutation TestRemoveRepo {
locationId
}
}
}
}
"""
API_INIT_MUTATION = """
mutation TestInitRepo($input: InitializeRepositoryInput!) {
backup {
initializeRepository(repository: $input) {
success
message
code
configuration {
configuration {
provider
encryptionKey
isInitialized
@ -67,20 +74,23 @@ mutation TestInitRepo($input: InitializeRepositoryInput!) {
locationId
}
}
}
}
"""
API_RESTORE_MUTATION = """
mutation TestRestoreService($snapshot_id: String!) {
backup {
restoreBackup(snapshotId: $snapshot_id) {
success
message
code
job {
job {
uid
status
}
}
}
}
"""
@ -96,15 +106,17 @@ allSnapshots {
API_BACK_UP_MUTATION = """
mutation TestBackupService($service_id: String!) {
backup {
startBackup(serviceId: $service_id) {
success
message
code
job {
job {
uid
status
}
}
}
}
"""
@ -225,7 +237,7 @@ def test_snapshots_empty(authorized_client, dummy_service):
def test_start_backup(authorized_client, dummy_service):
response = api_backup(authorized_client, dummy_service)
data = get_data(response)["startBackup"]
data = get_data(response)["backup"]["startBackup"]
assert data["success"] is True
job = data["job"]
@ -245,7 +257,7 @@ def test_restore(authorized_client, dummy_service):
assert snap["id"] is not None
response = api_restore(authorized_client, snap["id"])
data = get_data(response)["restoreBackup"]
data = get_data(response)["backup"]["restoreBackup"]
assert data["success"] is True
job = data["job"]
@ -257,7 +269,7 @@ def test_reinit(authorized_client, dummy_service, tmpdir):
response = api_init_without_key(
authorized_client, "FILE", "", "", test_repo_path, ""
)
data = get_data(response)["initializeRepository"]
data = get_data(response)["backup"]["initializeRepository"]
assert_ok(data)
configuration = data["configuration"]
assert configuration["provider"] == "FILE"
@ -267,7 +279,7 @@ def test_reinit(authorized_client, dummy_service, tmpdir):
assert configuration["isInitialized"] is True
response = api_backup(authorized_client, dummy_service)
data = get_data(response)["startBackup"]
data = get_data(response)["backup"]["startBackup"]
assert data["success"] is True
job = data["job"]
@ -276,7 +288,7 @@ def test_reinit(authorized_client, dummy_service, tmpdir):
def test_remove(authorized_client, generic_userdata):
response = api_remove(authorized_client)
data = get_data(response)["removeRepository"]
data = get_data(response)["backup"]["removeRepository"]
assert_ok(data)
configuration = data["configuration"]
@ -291,7 +303,7 @@ def test_remove(authorized_client, generic_userdata):
def test_autobackup_period_nonzero(authorized_client):
new_period = 11
response = api_set_period(authorized_client, new_period)
data = get_data(response)["setAutobackupPeriod"]
data = get_data(response)["backup"]["setAutobackupPeriod"]
assert_ok(data)
configuration = data["configuration"]
@ -304,7 +316,7 @@ def test_autobackup_period_zero(authorized_client):
response = api_set_period(authorized_client, 11)
# and now we nullify it
response = api_set_period(authorized_client, new_period)
data = get_data(response)["setAutobackupPeriod"]
data = get_data(response)["backup"]["setAutobackupPeriod"]
assert_ok(data)
configuration = data["configuration"]
@ -316,7 +328,7 @@ def test_autobackup_period_none(authorized_client):
response = api_set_period(authorized_client, 11)
# and now we nullify it
response = api_set_period(authorized_client, None)
data = get_data(response)["setAutobackupPeriod"]
data = get_data(response)["backup"]["setAutobackupPeriod"]
assert_ok(data)
configuration = data["configuration"]
@ -328,7 +340,7 @@ def test_autobackup_period_negative(authorized_client):
response = api_set_period(authorized_client, 11)
# and now we nullify it
response = api_set_period(authorized_client, -12)
data = get_data(response)["setAutobackupPeriod"]
data = get_data(response)["backup"]["setAutobackupPeriod"]
assert_ok(data)
configuration = data["configuration"]
@ -341,7 +353,7 @@ def test_reload_snapshots_bare_bare_bare(authorized_client, dummy_service):
api_remove(authorized_client)
response = api_reload_snapshots(authorized_client)
data = get_data(response)["forceSnapshotsReload"]
data = get_data(response)["backup"]["forceSnapshotsReload"]
assert_ok(data)
snaps = api_snapshots(authorized_client)
@ -350,10 +362,10 @@ def test_reload_snapshots_bare_bare_bare(authorized_client, dummy_service):
def test_reload_snapshots(authorized_client, dummy_service):
response = api_backup(authorized_client, dummy_service)
data = get_data(response)["startBackup"]
data = get_data(response)["backup"]["startBackup"]
response = api_reload_snapshots(authorized_client)
data = get_data(response)["forceSnapshotsReload"]
data = get_data(response)["backup"]["forceSnapshotsReload"]
assert_ok(data)
snaps = api_snapshots(authorized_client)

259
tests/test_graphql/test_api_devices.py
View File

@ -75,10 +75,12 @@ def test_graphql_tokens_info_unauthorized(client, tokens_file):
DELETE_TOKEN_MUTATION = """
mutation DeleteToken($device: String!) {
deleteDeviceApiToken(device: $device) {
success
message
code
api {
deleteDeviceApiToken(device: $device) {
success
message
code
}
}
}
"""
@ -110,9 +112,9 @@ def test_graphql_delete_token(authorized_client, tokens_file):
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["deleteDeviceApiToken"]["success"] is True
assert response.json()["data"]["deleteDeviceApiToken"]["message"] is not None
assert response.json()["data"]["deleteDeviceApiToken"]["code"] == 200
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": [
{
@ -136,13 +138,16 @@ def test_graphql_delete_self_token(authorized_client, tokens_file):
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["deleteDeviceApiToken"]["success"] is False
assert response.json()["data"]["deleteDeviceApiToken"]["message"] is not None
assert response.json()["data"]["deleteDeviceApiToken"]["code"] == 400
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):
def test_graphql_delete_nonexistent_token(
authorized_client,
tokens_file,
):
response = authorized_client.post(
"/graphql",
json={
@ -154,19 +159,21 @@ def test_graphql_delete_nonexistent_token(authorized_client, tokens_file):
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["deleteDeviceApiToken"]["success"] is False
assert response.json()["data"]["deleteDeviceApiToken"]["message"] is not None
assert response.json()["data"]["deleteDeviceApiToken"]["code"] == 404
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 {
refreshDeviceApiToken {
success
message
code
token
api {
refreshDeviceApiToken {
success
message
code
token
}
}
}
"""
@ -181,19 +188,25 @@ def test_graphql_refresh_token_unauthorized(client, tokens_file):
assert response.json()["data"] is None
def test_graphql_refresh_token(authorized_client, tokens_file, token_repo):
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"]["refreshDeviceApiToken"]["success"] is True
assert response.json()["data"]["refreshDeviceApiToken"]["message"] is not None
assert response.json()["data"]["refreshDeviceApiToken"]["code"] == 200
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"]["refreshDeviceApiToken"]["token"],
token=response.json()["data"]["api"]["refreshDeviceApiToken"]["token"],
device_name="test_token",
created_at=datetime.datetime(2022, 1, 14, 8, 31, 10, 789314),
)
@ -201,17 +214,22 @@ def test_graphql_refresh_token(authorized_client, tokens_file, token_repo):
NEW_DEVICE_KEY_MUTATION = """
mutation NewDeviceKey {
getNewDeviceApiKey {
success
message
code
key
api {
getNewDeviceApiKey {
success
message
code
key
}
}
}
"""
def test_graphql_get_new_device_auth_key_unauthorized(client, tokens_file):
def test_graphql_get_new_device_auth_key_unauthorized(
client,
tokens_file,
):
response = client.post(
"/graphql",
json={"query": NEW_DEVICE_KEY_MUTATION},
@ -220,22 +238,26 @@ def test_graphql_get_new_device_auth_key_unauthorized(client, tokens_file):
assert response.json()["data"] is None
def test_graphql_get_new_device_auth_key(authorized_client, tokens_file):
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"]["getNewDeviceApiKey"]["success"] is True
assert response.json()["data"]["getNewDeviceApiKey"]["message"] is not None
assert response.json()["data"]["getNewDeviceApiKey"]["code"] == 200
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"]["getNewDeviceApiKey"]["key"].split(" ").__len__() == 12
response.json()["data"]["api"]["getNewDeviceApiKey"]["key"].split(" ").__len__()
== 12
)
token = (
Mnemonic(language="english")
.to_entropy(response.json()["data"]["getNewDeviceApiKey"]["key"])
.to_entropy(response.json()["data"]["api"]["getNewDeviceApiKey"]["key"])
.hex()
)
assert read_json(tokens_file)["new_device"]["token"] == token
@ -243,20 +265,25 @@ def test_graphql_get_new_device_auth_key(authorized_client, tokens_file):
INVALIDATE_NEW_DEVICE_KEY_MUTATION = """
mutation InvalidateNewDeviceKey {
invalidateNewDeviceApiKey {
success
message
code
api {
invalidateNewDeviceApiKey {
success
message
code
}
}
}
"""
def test_graphql_invalidate_new_device_token_unauthorized(client, tokens_file):
def test_graphql_invalidate_new_device_token_unauthorized(
client,
tokens_file,
):
response = client.post(
"/graphql",
json={
"query": DELETE_TOKEN_MUTATION,
"query": INVALIDATE_NEW_DEVICE_KEY_MUTATION,
"variables": {
"device": "test_token",
},
@ -266,22 +293,26 @@ def test_graphql_invalidate_new_device_token_unauthorized(client, tokens_file):
assert response.json()["data"] is None
def test_graphql_get_and_delete_new_device_key(authorized_client, tokens_file):
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"]["getNewDeviceApiKey"]["success"] is True
assert response.json()["data"]["getNewDeviceApiKey"]["message"] is not None
assert response.json()["data"]["getNewDeviceApiKey"]["code"] == 200
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"]["getNewDeviceApiKey"]["key"].split(" ").__len__() == 12
response.json()["data"]["api"]["getNewDeviceApiKey"]["key"].split(" ").__len__()
== 12
)
token = (
Mnemonic(language="english")
.to_entropy(response.json()["data"]["getNewDeviceApiKey"]["key"])
.to_entropy(response.json()["data"]["api"]["getNewDeviceApiKey"]["key"])
.hex()
)
assert read_json(tokens_file)["new_device"]["token"] == token
@ -291,35 +322,46 @@ def test_graphql_get_and_delete_new_device_key(authorized_client, tokens_file):
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["invalidateNewDeviceApiKey"]["success"] is True
assert response.json()["data"]["invalidateNewDeviceApiKey"]["message"] is not None
assert response.json()["data"]["invalidateNewDeviceApiKey"]["code"] == 200
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!) {
authorizeWithNewDeviceApiKey(input: $input) {
success
message
code
token
api {
authorizeWithNewDeviceApiKey(input: $input) {
success
message
code
token
}
}
}
"""
def test_graphql_get_and_authorize_new_device(client, authorized_client, tokens_file):
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"]["getNewDeviceApiKey"]["success"] is True
assert response.json()["data"]["getNewDeviceApiKey"]["message"] is not None
assert response.json()["data"]["getNewDeviceApiKey"]["code"] == 200
mnemonic_key = response.json()["data"]["getNewDeviceApiKey"]["key"]
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
@ -337,17 +379,24 @@ def test_graphql_get_and_authorize_new_device(client, authorized_client, tokens_
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["authorizeWithNewDeviceApiKey"]["success"] is True
assert (
response.json()["data"]["authorizeWithNewDeviceApiKey"]["message"] is not None
response.json()["data"]["api"]["authorizeWithNewDeviceApiKey"]["success"]
is True
)
assert response.json()["data"]["authorizeWithNewDeviceApiKey"]["code"] == 200
token = response.json()["data"]["authorizeWithNewDeviceApiKey"]["token"]
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):
def test_graphql_authorize_new_device_with_invalid_key(
client,
tokens_file,
):
response = client.post(
"/graphql",
json={
@ -362,25 +411,33 @@ def test_graphql_authorize_new_device_with_invalid_key(client, tokens_file):
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["authorizeWithNewDeviceApiKey"]["success"] is False
assert (
response.json()["data"]["authorizeWithNewDeviceApiKey"]["message"] is not None
response.json()["data"]["api"]["authorizeWithNewDeviceApiKey"]["success"]
is False
)
assert response.json()["data"]["authorizeWithNewDeviceApiKey"]["code"] == 404
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):
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"]["getNewDeviceApiKey"]["success"] is True
assert response.json()["data"]["getNewDeviceApiKey"]["message"] is not None
assert response.json()["data"]["getNewDeviceApiKey"]["code"] == 200
mnemonic_key = response.json()["data"]["getNewDeviceApiKey"]["key"]
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
@ -398,14 +455,18 @@ def test_graphql_get_and_authorize_used_key(client, authorized_client, tokens_fi
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["authorizeWithNewDeviceApiKey"]["success"] is True
assert (
response.json()["data"]["authorizeWithNewDeviceApiKey"]["message"] is not None
response.json()["data"]["api"]["authorizeWithNewDeviceApiKey"]["success"]
is True
)
assert response.json()["data"]["authorizeWithNewDeviceApiKey"]["code"] == 200
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"]["authorizeWithNewDeviceApiKey"]["token"]
== response.json()["data"]["api"]["authorizeWithNewDeviceApiKey"]["token"]
)
assert read_json(tokens_file)["tokens"][2]["name"] == "new_token"
@ -415,7 +476,7 @@ def test_graphql_get_and_authorize_used_key(client, authorized_client, tokens_fi
"query": AUTHORIZE_WITH_NEW_DEVICE_KEY_MUTATION,
"variables": {
"input": {
"key": mnemonic_key,
"key": NEW_DEVICE_KEY_MUTATION,
"deviceName": "test_token2",
}
},
@ -423,16 +484,22 @@ def test_graphql_get_and_authorize_used_key(client, authorized_client, tokens_fi
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["authorizeWithNewDeviceApiKey"]["success"] is False
assert (
response.json()["data"]["authorizeWithNewDeviceApiKey"]["message"] is not None
response.json()["data"]["api"]["authorizeWithNewDeviceApiKey"]["success"]
is False
)
assert response.json()["data"]["authorizeWithNewDeviceApiKey"]["code"] == 404
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
client,
authorized_client,
tokens_file,
):
response = authorized_client.post(
"/graphql",
@ -440,15 +507,16 @@ def test_graphql_get_and_authorize_key_after_12_minutes(
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["getNewDeviceApiKey"]["success"] is True
assert response.json()["data"]["getNewDeviceApiKey"]["message"] is not None
assert response.json()["data"]["getNewDeviceApiKey"]["code"] == 200
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"]["getNewDeviceApiKey"]["key"].split(" ").__len__() == 12
response.json()["data"]["api"]["getNewDeviceApiKey"]["key"].split(" ").__len__()
== 12
)
key = (
Mnemonic(language="english")
.to_entropy(response.json()["data"]["getNewDeviceApiKey"]["key"])
.to_entropy(response.json()["data"]["api"]["getNewDeviceApiKey"]["key"])
.hex()
)
assert read_json(tokens_file)["new_device"]["token"] == key
@ -473,14 +541,21 @@ def test_graphql_get_and_authorize_key_after_12_minutes(
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["authorizeWithNewDeviceApiKey"]["success"] is False
assert (
response.json()["data"]["authorizeWithNewDeviceApiKey"]["message"] is not None
response.json()["data"]["api"]["authorizeWithNewDeviceApiKey"]["success"]
is False
)
assert response.json()["data"]["authorizeWithNewDeviceApiKey"]["code"] == 404
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):
def test_graphql_authorize_without_token(
client,
tokens_file,
):
response = client.post(
"/graphql",
json={

158
tests/test_graphql/test_api_recovery.py
View File

@ -57,22 +57,26 @@ def test_graphql_recovery_key_status_when_none_exists(authorized_client, tokens_
API_RECOVERY_KEY_GENERATE_MUTATION = """
mutation TestGenerateRecoveryKey($limits: RecoveryKeyLimitsInput) {
getNewRecoveryApiKey(limits: $limits) {
success
message
code
key
api {
getNewRecoveryApiKey(limits: $limits) {
success
message
code
key
}
}
}
"""
API_RECOVERY_KEY_USE_MUTATION = """
mutation TestUseRecoveryKey($input: UseRecoveryKeyInput!) {
useRecoveryApiKey(input: $input) {
success
message
code
token
api {
useRecoveryApiKey(input: $input) {
success
message
code
token
}
}
}
"""
@ -87,18 +91,20 @@ def test_graphql_generate_recovery_key(client, authorized_client, tokens_file):
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["getNewRecoveryApiKey"]["success"] is True
assert response.json()["data"]["getNewRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["getNewRecoveryApiKey"]["code"] == 200
assert response.json()["data"]["getNewRecoveryApiKey"]["key"] is not None
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["success"] is True
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["code"] == 200
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["key"] is not None
assert (
response.json()["data"]["getNewRecoveryApiKey"]["key"].split(" ").__len__()
response.json()["data"]["api"]["getNewRecoveryApiKey"]["key"]
.split(" ")
.__len__()
== 18
)
assert read_json(tokens_file)["recovery_token"] is not None
time_generated = read_json(tokens_file)["recovery_token"]["date"]
assert time_generated is not None
key = response.json()["data"]["getNewRecoveryApiKey"]["key"]
key = response.json()["data"]["api"]["getNewRecoveryApiKey"]["key"]
assert (
datetime.datetime.strptime(time_generated, "%Y-%m-%dT%H:%M:%S.%f")
- datetime.timedelta(seconds=5)
@ -136,12 +142,12 @@ def test_graphql_generate_recovery_key(client, authorized_client, tokens_file):
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["useRecoveryApiKey"]["success"] is True
assert response.json()["data"]["useRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["useRecoveryApiKey"]["code"] == 200
assert response.json()["data"]["useRecoveryApiKey"]["token"] is not None
assert response.json()["data"]["api"]["useRecoveryApiKey"]["success"] is True
assert response.json()["data"]["api"]["useRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["api"]["useRecoveryApiKey"]["code"] == 200
assert response.json()["data"]["api"]["useRecoveryApiKey"]["token"] is not None
assert (
response.json()["data"]["useRecoveryApiKey"]["token"]
response.json()["data"]["api"]["useRecoveryApiKey"]["token"]
== read_json(tokens_file)["tokens"][2]["token"]
)
assert read_json(tokens_file)["tokens"][2]["name"] == "new_test_token"
@ -161,12 +167,12 @@ def test_graphql_generate_recovery_key(client, authorized_client, tokens_file):
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["useRecoveryApiKey"]["success"] is True
assert response.json()["data"]["useRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["useRecoveryApiKey"]["code"] == 200
assert response.json()["data"]["useRecoveryApiKey"]["token"] is not None
assert response.json()["data"]["api"]["useRecoveryApiKey"]["success"] is True
assert response.json()["data"]["api"]["useRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["api"]["useRecoveryApiKey"]["code"] == 200
assert response.json()["data"]["api"]["useRecoveryApiKey"]["token"] is not None
assert (
response.json()["data"]["useRecoveryApiKey"]["token"]
response.json()["data"]["api"]["useRecoveryApiKey"]["token"]
== read_json(tokens_file)["tokens"][3]["token"]
)
assert read_json(tokens_file)["tokens"][3]["name"] == "new_test_token2"
@ -190,17 +196,19 @@ def test_graphql_generate_recovery_key_with_expiration_date(
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["getNewRecoveryApiKey"]["success"] is True
assert response.json()["data"]["getNewRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["getNewRecoveryApiKey"]["code"] == 200
assert response.json()["data"]["getNewRecoveryApiKey"]["key"] is not None
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["success"] is True
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["code"] == 200
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["key"] is not None
assert (
response.json()["data"]["getNewRecoveryApiKey"]["key"].split(" ").__len__()
response.json()["data"]["api"]["getNewRecoveryApiKey"]["key"]
.split(" ")
.__len__()
== 18
)
assert read_json(tokens_file)["recovery_token"] is not None
key = response.json()["data"]["getNewRecoveryApiKey"]["key"]
key = response.json()["data"]["api"]["getNewRecoveryApiKey"]["key"]
assert read_json(tokens_file)["recovery_token"]["expiration"] == expiration_date_str
assert read_json(tokens_file)["recovery_token"]["token"] == mnemonic_to_hex(key)
@ -246,12 +254,12 @@ def test_graphql_generate_recovery_key_with_expiration_date(
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["useRecoveryApiKey"]["success"] is True
assert response.json()["data"]["useRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["useRecoveryApiKey"]["code"] == 200
assert response.json()["data"]["useRecoveryApiKey"]["token"] is not None
assert response.json()["data"]["api"]["useRecoveryApiKey"]["success"] is True
assert response.json()["data"]["api"]["useRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["api"]["useRecoveryApiKey"]["code"] == 200
assert response.json()["data"]["api"]["useRecoveryApiKey"]["token"] is not None
assert (
response.json()["data"]["useRecoveryApiKey"]["token"]
response.json()["data"]["api"]["useRecoveryApiKey"]["token"]
== read_json(tokens_file)["tokens"][2]["token"]
)
@ -270,12 +278,12 @@ def test_graphql_generate_recovery_key_with_expiration_date(
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["useRecoveryApiKey"]["success"] is True
assert response.json()["data"]["useRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["useRecoveryApiKey"]["code"] == 200
assert response.json()["data"]["useRecoveryApiKey"]["token"] is not None
assert response.json()["data"]["api"]["useRecoveryApiKey"]["success"] is True
assert response.json()["data"]["api"]["useRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["api"]["useRecoveryApiKey"]["code"] == 200
assert response.json()["data"]["api"]["useRecoveryApiKey"]["token"] is not None
assert (
response.json()["data"]["useRecoveryApiKey"]["token"]
response.json()["data"]["api"]["useRecoveryApiKey"]["token"]
== read_json(tokens_file)["tokens"][3]["token"]
)
@ -299,10 +307,10 @@ def test_graphql_generate_recovery_key_with_expiration_date(
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["useRecoveryApiKey"]["success"] is False
assert response.json()["data"]["useRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["useRecoveryApiKey"]["code"] == 404
assert response.json()["data"]["useRecoveryApiKey"]["token"] is None
assert response.json()["data"]["api"]["useRecoveryApiKey"]["success"] is False
assert response.json()["data"]["api"]["useRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["api"]["useRecoveryApiKey"]["code"] == 404
assert response.json()["data"]["api"]["useRecoveryApiKey"]["token"] is None
assert read_json(tokens_file)["tokens"] == new_data["tokens"]
@ -345,10 +353,10 @@ def test_graphql_generate_recovery_key_with_expiration_in_the_past(
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["getNewRecoveryApiKey"]["success"] is False
assert response.json()["data"]["getNewRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["getNewRecoveryApiKey"]["code"] == 400
assert response.json()["data"]["getNewRecoveryApiKey"]["key"] is None
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["success"] is False
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["code"] == 400
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["key"] is None
assert "recovery_token" not in read_json(tokens_file)
@ -393,12 +401,12 @@ def test_graphql_generate_recovery_key_with_limited_uses(
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["getNewRecoveryApiKey"]["success"] is True
assert response.json()["data"]["getNewRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["getNewRecoveryApiKey"]["code"] == 200
assert response.json()["data"]["getNewRecoveryApiKey"]["key"] is not None
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["success"] is True
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["code"] == 200
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["key"] is not None
mnemonic_key = response.json()["data"]["getNewRecoveryApiKey"]["key"]
mnemonic_key = response.json()["data"]["api"]["getNewRecoveryApiKey"]["key"]
key = mnemonic_to_hex(mnemonic_key)
assert read_json(tokens_file)["recovery_token"]["token"] == key
@ -433,10 +441,10 @@ def test_graphql_generate_recovery_key_with_limited_uses(
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["useRecoveryApiKey"]["success"] is True
assert response.json()["data"]["useRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["useRecoveryApiKey"]["code"] == 200
assert response.json()["data"]["useRecoveryApiKey"]["token"] is not None
assert response.json()["data"]["api"]["useRecoveryApiKey"]["success"] is True
assert response.json()["data"]["api"]["useRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["api"]["useRecoveryApiKey"]["code"] == 200
assert response.json()["data"]["api"]["useRecoveryApiKey"]["token"] is not None
# Try to get token status
response = authorized_client.post(
@ -467,10 +475,10 @@ def test_graphql_generate_recovery_key_with_limited_uses(
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["useRecoveryApiKey"]["success"] is True
assert response.json()["data"]["useRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["useRecoveryApiKey"]["code"] == 200
assert response.json()["data"]["useRecoveryApiKey"]["token"] is not None
assert response.json()["data"]["api"]["useRecoveryApiKey"]["success"] is True
assert response.json()["data"]["api"]["useRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["api"]["useRecoveryApiKey"]["code"] == 200
assert response.json()["data"]["api"]["useRecoveryApiKey"]["token"] is not None
# Try to get token status
response = authorized_client.post(
@ -501,10 +509,10 @@ def test_graphql_generate_recovery_key_with_limited_uses(
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["useRecoveryApiKey"]["success"] is False
assert response.json()["data"]["useRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["useRecoveryApiKey"]["code"] == 404
assert response.json()["data"]["useRecoveryApiKey"]["token"] is None
assert response.json()["data"]["api"]["useRecoveryApiKey"]["success"] is False
assert response.json()["data"]["api"]["useRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["api"]["useRecoveryApiKey"]["code"] == 404
assert response.json()["data"]["api"]["useRecoveryApiKey"]["token"] is None
def test_graphql_generate_recovery_key_with_negative_uses(
@ -524,10 +532,10 @@ def test_graphql_generate_recovery_key_with_negative_uses(
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["getNewRecoveryApiKey"]["success"] is False
assert response.json()["data"]["getNewRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["getNewRecoveryApiKey"]["code"] == 400
assert response.json()["data"]["getNewRecoveryApiKey"]["key"] is None
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["success"] is False
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["code"] == 400
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["key"] is None
def test_graphql_generate_recovery_key_with_zero_uses(authorized_client, tokens_file):
@ -545,7 +553,7 @@ def test_graphql_generate_recovery_key_with_zero_uses(authorized_client, tokens_
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["getNewRecoveryApiKey"]["success"] is False
assert response.json()["data"]["getNewRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["getNewRecoveryApiKey"]["code"] == 400
assert response.json()["data"]["getNewRecoveryApiKey"]["key"] is None
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["success"] is False
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["message"] is not None
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["code"] == 400
assert response.json()["data"]["api"]["getNewRecoveryApiKey"]["key"] is None

2
tests/test_graphql/test_localsecret.py
View File

@ -35,4 +35,4 @@ def test_local_secret_set(localsecret):
assert oldsecret != newsecret
LocalBackupSecret.set(newsecret)
assert LocalBackupSecret.get() == newsecret
assert LocalBackupSecret.get() == newsecret

122
tests/test_graphql/test_ssh.py
View File

@ -44,13 +44,15 @@ def some_users(mocker, datadir):
API_CREATE_SSH_KEY_MUTATION = """
mutation addSshKey($sshInput: SshMutationInput!) {
addSshKey(sshInput: $sshInput) {
success
message
code
user {
username
sshKeys
users {
addSshKey(sshInput: $sshInput) {
success
message
code
user {
username
sshKeys
}
}
}
}
@ -90,12 +92,12 @@ def test_graphql_add_ssh_key(authorized_client, some_users, mock_subprocess_pope
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["addSshKey"]["code"] == 201
assert response.json()["data"]["addSshKey"]["message"] is not None
assert response.json()["data"]["addSshKey"]["success"] is True
assert response.json()["data"]["users"]["addSshKey"]["code"] == 201
assert response.json()["data"]["users"]["addSshKey"]["message"] is not None
assert response.json()["data"]["users"]["addSshKey"]["success"] is True
assert response.json()["data"]["addSshKey"]["user"]["username"] == "user1"
assert response.json()["data"]["addSshKey"]["user"]["sshKeys"] == [
assert response.json()["data"]["users"]["addSshKey"]["user"]["username"] == "user1"
assert response.json()["data"]["users"]["addSshKey"]["user"]["sshKeys"] == [
"ssh-rsa KEY user1@pc",
"ssh-rsa KEY test_key@pc",
]
@ -117,12 +119,12 @@ def test_graphql_add_root_ssh_key(authorized_client, some_users, mock_subprocess
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["addSshKey"]["code"] == 201
assert response.json()["data"]["addSshKey"]["message"] is not None
assert response.json()["data"]["addSshKey"]["success"] is True
assert response.json()["data"]["users"]["addSshKey"]["code"] == 201
assert response.json()["data"]["users"]["addSshKey"]["message"] is not None
assert response.json()["data"]["users"]["addSshKey"]["success"] is True
assert response.json()["data"]["addSshKey"]["user"]["username"] == "root"
assert response.json()["data"]["addSshKey"]["user"]["sshKeys"] == [
assert response.json()["data"]["users"]["addSshKey"]["user"]["username"] == "root"
assert response.json()["data"]["users"]["addSshKey"]["user"]["sshKeys"] == [
"ssh-ed25519 KEY test@pc",
"ssh-rsa KEY test_key@pc",
]
@ -144,12 +146,12 @@ def test_graphql_add_main_ssh_key(authorized_client, some_users, mock_subprocess
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["addSshKey"]["code"] == 201
assert response.json()["data"]["addSshKey"]["message"] is not None
assert response.json()["data"]["addSshKey"]["success"] is True
assert response.json()["data"]["users"]["addSshKey"]["code"] == 201
assert response.json()["data"]["users"]["addSshKey"]["message"] is not None
assert response.json()["data"]["users"]["addSshKey"]["success"] is True
assert response.json()["data"]["addSshKey"]["user"]["username"] == "tester"
assert response.json()["data"]["addSshKey"]["user"]["sshKeys"] == [
assert response.json()["data"]["users"]["addSshKey"]["user"]["username"] == "tester"
assert response.json()["data"]["users"]["addSshKey"]["user"]["sshKeys"] == [
"ssh-rsa KEY test@pc",
"ssh-rsa KEY test_key@pc",
]
@ -171,9 +173,9 @@ def test_graphql_add_bad_ssh_key(authorized_client, some_users, mock_subprocess_
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["addSshKey"]["code"] == 400
assert response.json()["data"]["addSshKey"]["message"] is not None
assert response.json()["data"]["addSshKey"]["success"] is False
assert response.json()["data"]["users"]["addSshKey"]["code"] == 400
assert response.json()["data"]["users"]["addSshKey"]["message"] is not None
assert response.json()["data"]["users"]["addSshKey"]["success"] is False
def test_graphql_add_ssh_key_nonexistent_user(
@ -194,20 +196,22 @@ def test_graphql_add_ssh_key_nonexistent_user(
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["addSshKey"]["code"] == 404
assert response.json()["data"]["addSshKey"]["message"] is not None
assert response.json()["data"]["addSshKey"]["success"] is False
assert response.json()["data"]["users"]["addSshKey"]["code"] == 404
assert response.json()["data"]["users"]["addSshKey"]["message"] is not None
assert response.json()["data"]["users"]["addSshKey"]["success"] is False
API_REMOVE_SSH_KEY_MUTATION = """
mutation removeSshKey($sshInput: SshMutationInput!) {
removeSshKey(sshInput: $sshInput) {
success
message
code
user {
username
sshKeys
users {
removeSshKey(sshInput: $sshInput) {
success
message
code
user {
username
sshKeys
}
}
}
}
@ -247,12 +251,14 @@ def test_graphql_remove_ssh_key(authorized_client, some_users, mock_subprocess_p
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["removeSshKey"]["code"] == 200
assert response.json()["data"]["removeSshKey"]["message"] is not None
assert response.json()["data"]["removeSshKey"]["success"] is True
assert response.json()["data"]["users"]["removeSshKey"]["code"] == 200
assert response.json()["data"]["users"]["removeSshKey"]["message"] is not None
assert response.json()["data"]["users"]["removeSshKey"]["success"] is True
assert response.json()["data"]["removeSshKey"]["user"]["username"] == "user1"
assert response.json()["data"]["removeSshKey"]["user"]["sshKeys"] == []
assert (
response.json()["data"]["users"]["removeSshKey"]["user"]["username"] == "user1"
)
assert response.json()["data"]["users"]["removeSshKey"]["user"]["sshKeys"] == []
def test_graphql_remove_root_ssh_key(
@ -273,12 +279,14 @@ def test_graphql_remove_root_ssh_key(
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["removeSshKey"]["code"] == 200
assert response.json()["data"]["removeSshKey"]["message"] is not None
assert response.json()["data"]["removeSshKey"]["success"] is True
assert response.json()["data"]["users"]["removeSshKey"]["code"] == 200
assert response.json()["data"]["users"]["removeSshKey"]["message"] is not None
assert response.json()["data"]["users"]["removeSshKey"]["success"] is True
assert response.json()["data"]["removeSshKey"]["user"]["username"] == "root"
assert response.json()["data"]["removeSshKey"]["user"]["sshKeys"] == []
assert (
response.json()["data"]["users"]["removeSshKey"]["user"]["username"] == "root"
)
assert response.json()["data"]["users"]["removeSshKey"]["user"]["sshKeys"] == []
def test_graphql_remove_main_ssh_key(
@ -299,12 +307,14 @@ def test_graphql_remove_main_ssh_key(
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["removeSshKey"]["code"] == 200
assert response.json()["data"]["removeSshKey"]["message"] is not None
assert response.json()["data"]["removeSshKey"]["success"] is True
assert response.json()["data"]["users"]["removeSshKey"]["code"] == 200
assert response.json()["data"]["users"]["removeSshKey"]["message"] is not None
assert response.json()["data"]["users"]["removeSshKey"]["success"] is True
assert response.json()["data"]["removeSshKey"]["user"]["username"] == "tester"
assert response.json()["data"]["removeSshKey"]["user"]["sshKeys"] == []
assert (
response.json()["data"]["users"]["removeSshKey"]["user"]["username"] == "tester"
)
assert response.json()["data"]["users"]["removeSshKey"]["user"]["sshKeys"] == []
def test_graphql_remove_nonexistent_ssh_key(
@ -325,9 +335,9 @@ def test_graphql_remove_nonexistent_ssh_key(
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["removeSshKey"]["code"] == 404
assert response.json()["data"]["removeSshKey"]["message"] is not None
assert response.json()["data"]["removeSshKey"]["success"] is False
assert response.json()["data"]["users"]["removeSshKey"]["code"] == 404
assert response.json()["data"]["users"]["removeSshKey"]["message"] is not None
assert response.json()["data"]["users"]["removeSshKey"]["success"] is False
def test_graphql_remove_ssh_key_nonexistent_user(
@ -348,6 +358,6 @@ def test_graphql_remove_ssh_key_nonexistent_user(
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["removeSshKey"]["code"] == 404
assert response.json()["data"]["removeSshKey"]["message"] is not None
assert response.json()["data"]["removeSshKey"]["success"] is False
assert response.json()["data"]["users"]["removeSshKey"]["code"] == 404
assert response.json()["data"]["users"]["removeSshKey"]["message"] is not None
assert response.json()["data"]["users"]["removeSshKey"]["success"] is False

241
tests/test_graphql/test_system.py
View File

@ -382,11 +382,13 @@ def test_graphql_get_timezone_on_undefined(authorized_client, undefined_config):
API_CHANGE_TIMEZONE_MUTATION = """
mutation changeTimezone($timezone: String!) {
changeTimezone(timezone: $timezone) {
success
message
code
timezone
system {
changeTimezone(timezone: $timezone) {
success
message
code
timezone
}
}
}
"""
@ -420,10 +422,13 @@ def test_graphql_change_timezone(authorized_client, turned_on):
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["changeTimezone"]["success"] is True
assert response.json()["data"]["changeTimezone"]["message"] is not None
assert response.json()["data"]["changeTimezone"]["code"] == 200
assert response.json()["data"]["changeTimezone"]["timezone"] == "Europe/Helsinki"
assert response.json()["data"]["system"]["changeTimezone"]["success"] is True
assert response.json()["data"]["system"]["changeTimezone"]["message"] is not None
assert response.json()["data"]["system"]["changeTimezone"]["code"] == 200
assert (
response.json()["data"]["system"]["changeTimezone"]["timezone"]
== "Europe/Helsinki"
)
assert read_json(turned_on / "turned_on.json")["timezone"] == "Europe/Helsinki"
@ -440,10 +445,13 @@ def test_graphql_change_timezone_on_undefined(authorized_client, undefined_confi
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["changeTimezone"]["success"] is True
assert response.json()["data"]["changeTimezone"]["message"] is not None
assert response.json()["data"]["changeTimezone"]["code"] == 200
assert response.json()["data"]["changeTimezone"]["timezone"] == "Europe/Helsinki"
assert response.json()["data"]["system"]["changeTimezone"]["success"] is True
assert response.json()["data"]["system"]["changeTimezone"]["message"] is not None
assert response.json()["data"]["system"]["changeTimezone"]["code"] == 200
assert (
response.json()["data"]["system"]["changeTimezone"]["timezone"]
== "Europe/Helsinki"
)
assert (
read_json(undefined_config / "undefined.json")["timezone"] == "Europe/Helsinki"
)
@ -462,10 +470,10 @@ def test_graphql_change_timezone_without_timezone(authorized_client, turned_on):
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["changeTimezone"]["success"] is False
assert response.json()["data"]["changeTimezone"]["message"] is not None
assert response.json()["data"]["changeTimezone"]["code"] == 400
assert response.json()["data"]["changeTimezone"]["timezone"] is None
assert response.json()["data"]["system"]["changeTimezone"]["success"] is False
assert response.json()["data"]["system"]["changeTimezone"]["message"] is not None
assert response.json()["data"]["system"]["changeTimezone"]["code"] == 400
assert response.json()["data"]["system"]["changeTimezone"]["timezone"] is None
assert read_json(turned_on / "turned_on.json")["timezone"] == "Europe/Moscow"
@ -482,10 +490,10 @@ def test_graphql_change_timezone_with_invalid_timezone(authorized_client, turned
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["changeTimezone"]["success"] is False
assert response.json()["data"]["changeTimezone"]["message"] is not None
assert response.json()["data"]["changeTimezone"]["code"] == 400
assert response.json()["data"]["changeTimezone"]["timezone"] is None
assert response.json()["data"]["system"]["changeTimezone"]["success"] is False
assert response.json()["data"]["system"]["changeTimezone"]["message"] is not None
assert response.json()["data"]["system"]["changeTimezone"]["code"] == 400
assert response.json()["data"]["system"]["changeTimezone"]["timezone"] is None
assert read_json(turned_on / "turned_on.json")["timezone"] == "Europe/Moscow"
@ -589,12 +597,14 @@ def test_graphql_get_auto_upgrade_turned_off(authorized_client, turned_off):
API_CHANGE_AUTO_UPGRADE_SETTINGS = """
mutation changeServerSettings($settings: AutoUpgradeSettingsInput!) {
changeAutoUpgradeSettings(settings: $settings) {
success
message
code
enableAutoUpgrade
allowReboot
system {
changeAutoUpgradeSettings(settings: $settings) {
success
message
code
enableAutoUpgrade
allowReboot
}
}
}
"""
@ -634,14 +644,25 @@ def test_graphql_change_auto_upgrade(authorized_client, turned_on):
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["changeAutoUpgradeSettings"]["success"] is True
assert response.json()["data"]["changeAutoUpgradeSettings"]["message"] is not None
assert response.json()["data"]["changeAutoUpgradeSettings"]["code"] == 200
assert (
response.json()["data"]["changeAutoUpgradeSettings"]["enableAutoUpgrade"]
response.json()["data"]["system"]["changeAutoUpgradeSettings"]["success"]
is True
)
assert (
response.json()["data"]["system"]["changeAutoUpgradeSettings"]["message"]
is not None
)
assert response.json()["data"]["system"]["changeAutoUpgradeSettings"]["code"] == 200
assert (
response.json()["data"]["system"]["changeAutoUpgradeSettings"][
"enableAutoUpgrade"
]
is False
)
assert response.json()["data"]["changeAutoUpgradeSettings"]["allowReboot"] is True
assert (
response.json()["data"]["system"]["changeAutoUpgradeSettings"]["allowReboot"]
is True
)
assert read_json(turned_on / "turned_on.json")["autoUpgrade"]["enable"] is False
assert read_json(turned_on / "turned_on.json")["autoUpgrade"]["allowReboot"] is True
@ -662,14 +683,25 @@ def test_graphql_change_auto_upgrade_on_undefined(authorized_client, undefined_c
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["changeAutoUpgradeSettings"]["success"] is True
assert response.json()["data"]["changeAutoUpgradeSettings"]["message"] is not None
assert response.json()["data"]["changeAutoUpgradeSettings"]["code"] == 200
assert (
response.json()["data"]["changeAutoUpgradeSettings"]["enableAutoUpgrade"]
response.json()["data"]["system"]["changeAutoUpgradeSettings"]["success"]
is True
)
assert (
response.json()["data"]["system"]["changeAutoUpgradeSettings"]["message"]
is not None
)
assert response.json()["data"]["system"]["changeAutoUpgradeSettings"]["code"] == 200
assert (
response.json()["data"]["system"]["changeAutoUpgradeSettings"][
"enableAutoUpgrade"
]
is False
)
assert response.json()["data"]["changeAutoUpgradeSettings"]["allowReboot"] is True
assert (
response.json()["data"]["system"]["changeAutoUpgradeSettings"]["allowReboot"]
is True
)
assert (
read_json(undefined_config / "undefined.json")["autoUpgrade"]["enable"] is False
)
@ -695,14 +727,25 @@ def test_graphql_change_auto_upgrade_without_vlaues(authorized_client, no_values
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["changeAutoUpgradeSettings"]["success"] is True
assert response.json()["data"]["changeAutoUpgradeSettings"]["message"] is not None
assert response.json()["data"]["changeAutoUpgradeSettings"]["code"] == 200
assert (
response.json()["data"]["changeAutoUpgradeSettings"]["enableAutoUpgrade"]
response.json()["data"]["system"]["changeAutoUpgradeSettings"]["success"]
is True
)
assert (
response.json()["data"]["system"]["changeAutoUpgradeSettings"]["message"]
is not None
)
assert response.json()["data"]["system"]["changeAutoUpgradeSettings"]["code"] == 200
assert (
response.json()["data"]["system"]["changeAutoUpgradeSettings"][
"enableAutoUpgrade"
]
is True
)
assert (
response.json()["data"]["system"]["changeAutoUpgradeSettings"]["allowReboot"]
is True
)
assert response.json()["data"]["changeAutoUpgradeSettings"]["allowReboot"] is True
assert read_json(no_values / "no_values.json")["autoUpgrade"]["enable"] is True
assert read_json(no_values / "no_values.json")["autoUpgrade"]["allowReboot"] is True
@ -723,14 +766,25 @@ def test_graphql_change_auto_upgrade_turned_off(authorized_client, turned_off):
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["changeAutoUpgradeSettings"]["success"] is True
assert response.json()["data"]["changeAutoUpgradeSettings"]["message"] is not None
assert response.json()["data"]["changeAutoUpgradeSettings"]["code"] == 200
assert (
response.json()["data"]["changeAutoUpgradeSettings"]["enableAutoUpgrade"]
response.json()["data"]["system"]["changeAutoUpgradeSettings"]["success"]
is True
)
assert (
response.json()["data"]["system"]["changeAutoUpgradeSettings"]["message"]
is not None
)
assert response.json()["data"]["system"]["changeAutoUpgradeSettings"]["code"] == 200
assert (
response.json()["data"]["system"]["changeAutoUpgradeSettings"][
"enableAutoUpgrade"
]
is True
)
assert (
response.json()["data"]["system"]["changeAutoUpgradeSettings"]["allowReboot"]
is True
)
assert response.json()["data"]["changeAutoUpgradeSettings"]["allowReboot"] is True
assert read_json(turned_off / "turned_off.json")["autoUpgrade"]["enable"] is True
assert (
read_json(turned_off / "turned_off.json")["autoUpgrade"]["allowReboot"] is True
@ -752,14 +806,25 @@ def test_grphql_change_auto_upgrade_without_enable(authorized_client, turned_off
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["changeAutoUpgradeSettings"]["success"] is True
assert response.json()["data"]["changeAutoUpgradeSettings"]["message"] is not None
assert response.json()["data"]["changeAutoUpgradeSettings"]["code"] == 200
assert (
response.json()["data"]["changeAutoUpgradeSettings"]["enableAutoUpgrade"]
response.json()["data"]["system"]["changeAutoUpgradeSettings"]["success"]
is True
)
assert (
response.json()["data"]["system"]["changeAutoUpgradeSettings"]["message"]
is not None
)
assert response.json()["data"]["system"]["changeAutoUpgradeSettings"]["code"] == 200
assert (
response.json()["data"]["system"]["changeAutoUpgradeSettings"][
"enableAutoUpgrade"
]
is False
)
assert response.json()["data"]["changeAutoUpgradeSettings"]["allowReboot"] is True
assert (
response.json()["data"]["system"]["changeAutoUpgradeSettings"]["allowReboot"]
is True
)
assert read_json(turned_off / "turned_off.json")["autoUpgrade"]["enable"] is False
assert (
read_json(turned_off / "turned_off.json")["autoUpgrade"]["allowReboot"] is True
@ -783,14 +848,25 @@ def test_graphql_change_auto_upgrade_without_allow_reboot(
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["changeAutoUpgradeSettings"]["success"] is True
assert response.json()["data"]["changeAutoUpgradeSettings"]["message"] is not None
assert response.json()["data"]["changeAutoUpgradeSettings"]["code"] == 200
assert (
response.json()["data"]["changeAutoUpgradeSettings"]["enableAutoUpgrade"]
response.json()["data"]["system"]["changeAutoUpgradeSettings"]["success"]
is True
)
assert response.json()["data"]["changeAutoUpgradeSettings"]["allowReboot"] is False
assert (
response.json()["data"]["system"]["changeAutoUpgradeSettings"]["message"]
is not None
)
assert response.json()["data"]["system"]["changeAutoUpgradeSettings"]["code"] == 200
assert (
response.json()["data"]["system"]["changeAutoUpgradeSettings"][
"enableAutoUpgrade"
]
is True
)
assert (
response.json()["data"]["system"]["changeAutoUpgradeSettings"]["allowReboot"]
is False
)
assert read_json(turned_off / "turned_off.json")["autoUpgrade"]["enable"] is True
assert (
read_json(turned_off / "turned_off.json")["autoUpgrade"]["allowReboot"] is False
@ -810,14 +886,25 @@ def test_graphql_change_auto_upgrade_with_empty_input(authorized_client, turned_
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["changeAutoUpgradeSettings"]["success"] is True
assert response.json()["data"]["changeAutoUpgradeSettings"]["message"] is not None
assert response.json()["data"]["changeAutoUpgradeSettings"]["code"] == 200
assert (
response.json()["data"]["changeAutoUpgradeSettings"]["enableAutoUpgrade"]
response.json()["data"]["system"]["changeAutoUpgradeSettings"]["success"]
is True
)
assert (
response.json()["data"]["system"]["changeAutoUpgradeSettings"]["message"]
is not None
)
assert response.json()["data"]["system"]["changeAutoUpgradeSettings"]["code"] == 200
assert (
response.json()["data"]["system"]["changeAutoUpgradeSettings"][
"enableAutoUpgrade"
]
is False
)
assert (
response.json()["data"]["system"]["changeAutoUpgradeSettings"]["allowReboot"]
is False
)
assert response.json()["data"]["changeAutoUpgradeSettings"]["allowReboot"] is False
assert read_json(turned_off / "turned_off.json")["autoUpgrade"]["enable"] is False
assert (
read_json(turned_off / "turned_off.json")["autoUpgrade"]["allowReboot"] is False
@ -826,10 +913,12 @@ def test_graphql_change_auto_upgrade_with_empty_input(authorized_client, turned_
API_PULL_SYSTEM_CONFIGURATION_MUTATION = """
mutation testPullSystemConfiguration {
pullRepositoryChanges {
success
message
code
system {
pullRepositoryChanges {
success
message
code
}
}
}
"""
@ -861,9 +950,12 @@ def test_graphql_pull_system_configuration(
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["pullRepositoryChanges"]["success"] is True
assert response.json()["data"]["pullRepositoryChanges"]["message"] is not None
assert response.json()["data"]["pullRepositoryChanges"]["code"] == 200
assert response.json()["data"]["system"]["pullRepositoryChanges"]["success"] is True
assert (
response.json()["data"]["system"]["pullRepositoryChanges"]["message"]
is not None
)
assert response.json()["data"]["system"]["pullRepositoryChanges"]["code"] == 200
assert mock_subprocess_popen.call_count == 1
assert mock_subprocess_popen.call_args[0][0] == ["git", "pull"]
@ -886,9 +978,14 @@ def test_graphql_pull_system_broken_repo(
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["pullRepositoryChanges"]["success"] is False
assert response.json()["data"]["pullRepositoryChanges"]["message"] is not None
assert response.json()["data"]["pullRepositoryChanges"]["code"] == 500
assert (
response.json()["data"]["system"]["pullRepositoryChanges"]["success"] is False
)
assert (
response.json()["data"]["system"]["pullRepositoryChanges"]["message"]
is not None
)
assert response.json()["data"]["system"]["pullRepositoryChanges"]["code"] == 500
assert mock_broken_service.call_count == 1
assert mock_os_chdir.call_count == 2

64
tests/test_graphql/test_system_nixos_tasks.py
View File

@ -54,10 +54,12 @@ def mock_subprocess_check_output(mocker):
API_REBUILD_SYSTEM_MUTATION = """
mutation rebuildSystem {
runSystemRebuild {
success
message
code
system {
runSystemRebuild {
success
message
code
}
}
}
"""
@ -86,9 +88,9 @@ def test_graphql_system_rebuild(authorized_client, mock_subprocess_popen):
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["runSystemRebuild"]["success"] is True
assert response.json()["data"]["runSystemRebuild"]["message"] is not None
assert response.json()["data"]["runSystemRebuild"]["code"] == 200
assert response.json()["data"]["system"]["runSystemRebuild"]["success"] is True
assert response.json()["data"]["system"]["runSystemRebuild"]["message"] is not None
assert response.json()["data"]["system"]["runSystemRebuild"]["code"] == 200
assert mock_subprocess_popen.call_count == 1
assert mock_subprocess_popen.call_args[0][0] == [
"systemctl",
@ -99,10 +101,12 @@ def test_graphql_system_rebuild(authorized_client, mock_subprocess_popen):
API_UPGRADE_SYSTEM_MUTATION = """
mutation upgradeSystem {
runSystemUpgrade {
success
message
code
system {
runSystemUpgrade {
success
message
code
}
}
}
"""
@ -131,9 +135,9 @@ def test_graphql_system_upgrade(authorized_client, mock_subprocess_popen):
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["runSystemUpgrade"]["success"] is True
assert response.json()["data"]["runSystemUpgrade"]["message"] is not None
assert response.json()["data"]["runSystemUpgrade"]["code"] == 200
assert response.json()["data"]["system"]["runSystemUpgrade"]["success"] is True
assert response.json()["data"]["system"]["runSystemUpgrade"]["message"] is not None
assert response.json()["data"]["system"]["runSystemUpgrade"]["code"] == 200
assert mock_subprocess_popen.call_count == 1
assert mock_subprocess_popen.call_args[0][0] == [
"systemctl",
@ -144,10 +148,12 @@ def test_graphql_system_upgrade(authorized_client, mock_subprocess_popen):
API_ROLLBACK_SYSTEM_MUTATION = """
mutation rollbackSystem {
runSystemRollback {
success
message
code
system {
runSystemRollback {
success
message
code
}
}
}
"""
@ -176,9 +182,9 @@ def test_graphql_system_rollback(authorized_client, mock_subprocess_popen):
)
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["runSystemRollback"]["success"] is True
assert response.json()["data"]["runSystemRollback"]["message"] is not None
assert response.json()["data"]["runSystemRollback"]["code"] == 200
assert response.json()["data"]["system"]["runSystemRollback"]["success"] is True
assert response.json()["data"]["system"]["runSystemRollback"]["message"] is not None
assert response.json()["data"]["system"]["runSystemRollback"]["code"] == 200
assert mock_subprocess_popen.call_count == 1
assert mock_subprocess_popen.call_args[0][0] == [
"systemctl",
@ -189,10 +195,12 @@ def test_graphql_system_rollback(authorized_client, mock_subprocess_popen):
API_REBOOT_SYSTEM_MUTATION = """
mutation system {
rebootSystem {
success
message
code
system {
rebootSystem {
success
message
code
}
}
}
"""
@ -223,9 +231,9 @@ def test_graphql_reboot_system(authorized_client, mock_subprocess_popen):
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["rebootSystem"]["success"] is True
assert response.json()["data"]["rebootSystem"]["message"] is not None
assert response.json()["data"]["rebootSystem"]["code"] == 200
assert response.json()["data"]["system"]["rebootSystem"]["success"] is True
assert response.json()["data"]["system"]["rebootSystem"]["message"] is not None
assert response.json()["data"]["system"]["rebootSystem"]["code"] == 200
assert mock_subprocess_popen.call_count == 1
assert mock_subprocess_popen.call_args[0][0] == ["reboot"]

168
tests/test_graphql/test_users.py
View File

@ -295,13 +295,15 @@ def test_graphql_get_nonexistent_user(
API_CREATE_USERS_MUTATION = """
mutation createUser($user: UserMutationInput!) {
createUser(user: $user) {
success
message
code
user {
username
sshKeys
users {
createUser(user: $user) {
success
message
code
user {
username
sshKeys
}
}
}
}
@ -341,12 +343,12 @@ def test_graphql_add_user(authorized_client, one_user, mock_subprocess_popen):
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["createUser"]["message"] is not None
assert response.json()["data"]["createUser"]["code"] == 201
assert response.json()["data"]["createUser"]["success"] is True
assert response.json()["data"]["users"]["createUser"]["message"] is not None
assert response.json()["data"]["users"]["createUser"]["code"] == 201
assert response.json()["data"]["users"]["createUser"]["success"] is True
assert response.json()["data"]["createUser"]["user"]["username"] == "user2"
assert response.json()["data"]["createUser"]["user"]["sshKeys"] == []
assert response.json()["data"]["users"]["createUser"]["user"]["username"] == "user2"
assert response.json()["data"]["users"]["createUser"]["user"]["sshKeys"] == []
def test_graphql_add_undefined_settings(
@ -367,12 +369,12 @@ def test_graphql_add_undefined_settings(
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["createUser"]["message"] is not None
assert response.json()["data"]["createUser"]["code"] == 201
assert response.json()["data"]["createUser"]["success"] is True
assert response.json()["data"]["users"]["createUser"]["message"] is not None
assert response.json()["data"]["users"]["createUser"]["code"] == 201
assert response.json()["data"]["users"]["createUser"]["success"] is True
assert response.json()["data"]["createUser"]["user"]["username"] == "user2"
assert response.json()["data"]["createUser"]["user"]["sshKeys"] == []
assert response.json()["data"]["users"]["createUser"]["user"]["username"] == "user2"
assert response.json()["data"]["users"]["createUser"]["user"]["sshKeys"] == []
def test_graphql_add_without_password(
@ -393,11 +395,11 @@ def test_graphql_add_without_password(
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["createUser"]["message"] is not None
assert response.json()["data"]["createUser"]["code"] == 400
assert response.json()["data"]["createUser"]["success"] is False
assert response.json()["data"]["users"]["createUser"]["message"] is not None
assert response.json()["data"]["users"]["createUser"]["code"] == 400
assert response.json()["data"]["users"]["createUser"]["success"] is False
assert response.json()["data"]["createUser"]["user"] is None
assert response.json()["data"]["users"]["createUser"]["user"] is None
def test_graphql_add_without_both(authorized_client, one_user, mock_subprocess_popen):
@ -416,11 +418,11 @@ def test_graphql_add_without_both(authorized_client, one_user, mock_subprocess_p
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["createUser"]["message"] is not None
assert response.json()["data"]["createUser"]["code"] == 400
assert response.json()["data"]["createUser"]["success"] is False
assert response.json()["data"]["users"]["createUser"]["message"] is not None
assert response.json()["data"]["users"]["createUser"]["code"] == 400
assert response.json()["data"]["users"]["createUser"]["success"] is False
assert response.json()["data"]["createUser"]["user"] is None
assert response.json()["data"]["users"]["createUser"]["user"] is None
@pytest.mark.parametrize("username", invalid_usernames)
@ -442,11 +444,11 @@ def test_graphql_add_system_username(
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["createUser"]["message"] is not None
assert response.json()["data"]["createUser"]["code"] == 409
assert response.json()["data"]["createUser"]["success"] is False
assert response.json()["data"]["users"]["createUser"]["message"] is not None
assert response.json()["data"]["users"]["createUser"]["code"] == 409
assert response.json()["data"]["users"]["createUser"]["success"] is False
assert response.json()["data"]["createUser"]["user"] is None
assert response.json()["data"]["users"]["createUser"]["user"] is None
def test_graphql_add_existing_user(authorized_client, one_user, mock_subprocess_popen):
@ -465,13 +467,13 @@ def test_graphql_add_existing_user(authorized_client, one_user, mock_subprocess_
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["createUser"]["message"] is not None
assert response.json()["data"]["createUser"]["code"] == 409
assert response.json()["data"]["createUser"]["success"] is False
assert response.json()["data"]["users"]["createUser"]["message"] is not None
assert response.json()["data"]["users"]["createUser"]["code"] == 409
assert response.json()["data"]["users"]["createUser"]["success"] is False
assert response.json()["data"]["createUser"]["user"]["username"] == "user1"
assert response.json()["data"]["users"]["createUser"]["user"]["username"] == "user1"
assert (
response.json()["data"]["createUser"]["user"]["sshKeys"][0]
response.json()["data"]["users"]["createUser"]["user"]["sshKeys"][0]
== "ssh-rsa KEY user1@pc"
)
@ -492,13 +494,15 @@ def test_graphql_add_main_user(authorized_client, one_user, mock_subprocess_pope
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["createUser"]["message"] is not None
assert response.json()["data"]["createUser"]["code"] == 409
assert response.json()["data"]["createUser"]["success"] is False
assert response.json()["data"]["users"]["createUser"]["message"] is not None
assert response.json()["data"]["users"]["createUser"]["code"] == 409
assert response.json()["data"]["users"]["createUser"]["success"] is False
assert response.json()["data"]["createUser"]["user"]["username"] == "tester"
assert (
response.json()["data"]["createUser"]["user"]["sshKeys"][0]
response.json()["data"]["users"]["createUser"]["user"]["username"] == "tester"
)
assert (
response.json()["data"]["users"]["createUser"]["user"]["sshKeys"][0]
== "ssh-rsa KEY test@pc"
)
@ -518,11 +522,11 @@ def test_graphql_add_long_username(authorized_client, one_user, mock_subprocess_
)
assert response.json().get("data") is not None
assert response.json()["data"]["createUser"]["message"] is not None
assert response.json()["data"]["createUser"]["code"] == 400
assert response.json()["data"]["createUser"]["success"] is False
assert response.json()["data"]["users"]["createUser"]["message"] is not None
assert response.json()["data"]["users"]["createUser"]["code"] == 400
assert response.json()["data"]["users"]["createUser"]["success"] is False
assert response.json()["data"]["createUser"]["user"] is None
assert response.json()["data"]["users"]["createUser"]["user"] is None
@pytest.mark.parametrize("username", ["", "1", "фыр", "user1@", "^-^"])
@ -544,19 +548,21 @@ def test_graphql_add_invalid_username(
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["createUser"]["message"] is not None
assert response.json()["data"]["createUser"]["code"] == 400
assert response.json()["data"]["createUser"]["success"] is False
assert response.json()["data"]["users"]["createUser"]["message"] is not None
assert response.json()["data"]["users"]["createUser"]["code"] == 400
assert response.json()["data"]["users"]["createUser"]["success"] is False
assert response.json()["data"]["createUser"]["user"] is None
assert response.json()["data"]["users"]["createUser"]["user"] is None
API_DELETE_USER_MUTATION = """
mutation deleteUser($username: String!) {
deleteUser(username: $username) {
success
message
code
users {
deleteUser(username: $username) {
success
message
code
}
}
}
"""
@ -585,9 +591,9 @@ def test_graphql_delete_user(authorized_client, some_users, mock_subprocess_pope
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["deleteUser"]["code"] == 200
assert response.json()["data"]["deleteUser"]["message"] is not None
assert response.json()["data"]["deleteUser"]["success"] is True
assert response.json()["data"]["users"]["deleteUser"]["code"] == 200
assert response.json()["data"]["users"]["deleteUser"]["message"] is not None
assert response.json()["data"]["users"]["deleteUser"]["success"] is True
@pytest.mark.parametrize("username", ["", "def"])
@ -604,9 +610,9 @@ def test_graphql_delete_nonexistent_users(
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["deleteUser"]["code"] == 404
assert response.json()["data"]["deleteUser"]["message"] is not None
assert response.json()["data"]["deleteUser"]["success"] is False
assert response.json()["data"]["users"]["deleteUser"]["code"] == 404
assert response.json()["data"]["users"]["deleteUser"]["message"] is not None
assert response.json()["data"]["users"]["deleteUser"]["success"] is False
@pytest.mark.parametrize("username", invalid_usernames)
@ -624,11 +630,11 @@ def test_graphql_delete_system_users(
assert response.json().get("data") is not None
assert (
response.json()["data"]["deleteUser"]["code"] == 404
or response.json()["data"]["deleteUser"]["code"] == 400
response.json()["data"]["users"]["deleteUser"]["code"] == 404
or response.json()["data"]["users"]["deleteUser"]["code"] == 400
)
assert response.json()["data"]["deleteUser"]["message"] is not None
assert response.json()["data"]["deleteUser"]["success"] is False
assert response.json()["data"]["users"]["deleteUser"]["message"] is not None
assert response.json()["data"]["users"]["deleteUser"]["success"] is False
def test_graphql_delete_main_user(authorized_client, some_users, mock_subprocess_popen):
@ -642,20 +648,22 @@ def test_graphql_delete_main_user(authorized_client, some_users, mock_subprocess
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["deleteUser"]["code"] == 400
assert response.json()["data"]["deleteUser"]["message"] is not None
assert response.json()["data"]["deleteUser"]["success"] is False
assert response.json()["data"]["users"]["deleteUser"]["code"] == 400
assert response.json()["data"]["users"]["deleteUser"]["message"] is not None
assert response.json()["data"]["users"]["deleteUser"]["success"] is False
API_UPDATE_USER_MUTATION = """
mutation updateUser($user: UserMutationInput!) {
updateUser(user: $user) {
success
message
code
user {
username
sshKeys
users {
updateUser(user: $user) {
success
message
code
user {
username
sshKeys
}
}
}
}
@ -695,12 +703,12 @@ def test_graphql_update_user(authorized_client, some_users, mock_subprocess_pope
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["updateUser"]["code"] == 200
assert response.json()["data"]["updateUser"]["message"] is not None
assert response.json()["data"]["updateUser"]["success"] is True
assert response.json()["data"]["users"]["updateUser"]["code"] == 200
assert response.json()["data"]["users"]["updateUser"]["message"] is not None
assert response.json()["data"]["users"]["updateUser"]["success"] is True
assert response.json()["data"]["updateUser"]["user"]["username"] == "user1"
assert response.json()["data"]["updateUser"]["user"]["sshKeys"] == [
assert response.json()["data"]["users"]["updateUser"]["user"]["username"] == "user1"
assert response.json()["data"]["users"]["updateUser"]["user"]["sshKeys"] == [
"ssh-rsa KEY user1@pc"
]
assert mock_subprocess_popen.call_count == 1
@ -724,9 +732,9 @@ def test_graphql_update_nonexistent_user(
assert response.status_code == 200
assert response.json().get("data") is not None
assert response.json()["data"]["updateUser"]["code"] == 404
assert response.json()["data"]["updateUser"]["message"] is not None
assert response.json()["data"]["updateUser"]["success"] is False
assert response.json()["data"]["users"]["updateUser"]["code"] == 404
assert response.json()["data"]["users"]["updateUser"]["message"] is not None
assert response.json()["data"]["users"]["updateUser"]["success"] is False
assert response.json()["data"]["updateUser"]["user"] is None
assert response.json()["data"]["users"]["updateUser"]["user"] is None
assert mock_subprocess_popen.call_count == 1

17
tests/test_model_storage.py
View File

@ -10,6 +10,7 @@ from selfprivacy_api.utils.redis_pool import RedisPool
TEST_KEY = "model_storage"
redis = RedisPool().get_connection()
@pytest.fixture()
def clean_redis():
redis.delete(TEST_KEY)
@ -19,18 +20,14 @@ class DummyModel(BaseModel):
name: str
date: Optional[datetime]
def test_store_retrieve():
model = DummyModel(
name= "test",
date= datetime.now()
)
model = DummyModel(name="test", date=datetime.now())
store_model_as_hash(redis, TEST_KEY, model)
assert hash_as_model(redis, TEST_KEY, DummyModel) == model
assert hash_as_model(redis, TEST_KEY, DummyModel) == model
def test_store_retrieve_none():
model = DummyModel(
name= "test",
date= None
)
model = DummyModel(name="test", date=None)
store_model_as_hash(redis, TEST_KEY, model)
assert hash_as_model(redis, TEST_KEY, DummyModel) == model
assert hash_as_model(redis, TEST_KEY, DummyModel) == model