test(backup): migration test

This commit is contained in:
Houkime 2024-07-29 10:14:22 +00:00
parent 427fdbdb49
commit a4a70c07d3
2 changed files with 97 additions and 7 deletions

View file

@ -75,8 +75,8 @@ class BackupMutations:
secret = repository.local_secret secret = repository.local_secret
if secret is not None: if secret is not None:
LocalBackupSecret.set(secret) LocalBackupSecret.set(secret)
else:
Backups.init_repo() Backups.init_repo()
return GenericBackupConfigReturn( return GenericBackupConfigReturn(
success=True, success=True,
message="", message="",

View file

@ -11,6 +11,8 @@ from selfprivacy_api.graphql.common_types.backup import (
AutobackupQuotas, AutobackupQuotas,
) )
from selfprivacy_api.jobs import Jobs, JobStatus from selfprivacy_api.jobs import Jobs, JobStatus
from selfprivacy_api.backup.storage import Storage
from selfprivacy_api.backup.local_secret import LocalBackupSecret
API_RELOAD_SNAPSHOTS = """ API_RELOAD_SNAPSHOTS = """
mutation TestSnapshotsReload { mutation TestSnapshotsReload {
@ -152,6 +154,17 @@ allSnapshots {
} }
""" """
API_BACKUP_SETTINGS_QUERY = """
configuration {
provider
encryptionKey
isInitialized
autobackupPeriod
locationName
locationId
}
"""
API_BACK_UP_MUTATION = """ API_BACK_UP_MUTATION = """
mutation TestBackupService($service_id: String!) { mutation TestBackupService($service_id: String!) {
backup { backup {
@ -246,8 +259,14 @@ def api_reload_snapshots(authorized_client):
return response return response
def api_init_without_key( def api_init(
authorized_client, kind, login, password, location_name, location_id authorized_client,
kind,
login,
password,
location_name,
location_id,
local_secret=None,
): ):
response = authorized_client.post( response = authorized_client.post(
"/graphql", "/graphql",
@ -260,6 +279,7 @@ def api_init_without_key(
"locationName": location_name, "locationName": location_name,
"login": login, "login": login,
"password": password, "password": password,
"localSecret": local_secret,
} }
}, },
}, },
@ -299,6 +319,17 @@ def api_snapshots(authorized_client):
return result return result
def api_settings(authorized_client):
response = authorized_client.post(
"/graphql",
json={"query": generate_backup_query([API_BACKUP_SETTINGS_QUERY])},
)
data = get_data(response)
result = data["backup"]["configuration"]
assert result is not None
return result
def test_dummy_service_convertible_to_gql(dummy_service): def test_dummy_service_convertible_to_gql(dummy_service):
gql_service = service_to_graphql_service(dummy_service) gql_service = service_to_graphql_service(dummy_service)
assert gql_service is not None assert gql_service is not None
@ -354,9 +385,7 @@ def test_restore(authorized_client, dummy_service, backups):
def test_reinit(authorized_client, dummy_service, tmpdir, backups): def test_reinit(authorized_client, dummy_service, tmpdir, backups):
test_repo_path = path.join(tmpdir, "not_at_all_sus") test_repo_path = path.join(tmpdir, "not_at_all_sus")
response = api_init_without_key( response = api_init(authorized_client, "FILE", "", "", test_repo_path, "")
authorized_client, "FILE", "", "", test_repo_path, ""
)
data = get_data(response)["backup"]["initializeRepository"] data = get_data(response)["backup"]["initializeRepository"]
assert_ok(data) assert_ok(data)
configuration = data["configuration"] configuration = data["configuration"]
@ -374,6 +403,67 @@ def test_reinit(authorized_client, dummy_service, tmpdir, backups):
assert Jobs.get_job(job["uid"]).status == JobStatus.FINISHED assert Jobs.get_job(job["uid"]).status == JobStatus.FINISHED
def test_migrate(authorized_client, dummy_service, tmpdir, backups):
"""
Simulate the workflow of migrating to a new server
"""
# Using an alternative path to be sure that we do not
# match only by incident
test_repo_path = path.join(tmpdir, "not_at_all_sus")
response = api_init(authorized_client, "FILE", "", "", test_repo_path, "")
data = get_data(response)["backup"]["initializeRepository"]
assert_ok(data)
snaps = api_snapshots(authorized_client)
assert snaps == []
# Now, forget what we just did
del test_repo_path
del response
del data
del snaps
# I am a user at my old machine, I make a backup
response = api_backup(authorized_client, dummy_service)
data = get_data(response)["backup"]["startBackup"]
assert_ok(data)
# Then oh no, we need to migrate, we get our settings.
# Because we have forgot everything 2000 times already
# Was years, was years.
# I still remember login though
configuration = api_settings(authorized_client)
# Ok. Let's now go to another machine
# Another machine will not have any settings at all
Storage.reset()
LocalBackupSecret._full_reset()
# That's it, nothing left
new_configuration = api_settings(authorized_client)
assert new_configuration["isInitialized"] is False
# Reinit
response = api_init(
authorized_client,
configuration["provider"],
"",
"",
configuration["locationName"],
configuration["locationId"],
configuration["encryptionKey"],
)
data = get_data(response)["backup"]["initializeRepository"]
assert_ok(data)
new_configuration = api_settings(authorized_client)
assert new_configuration == configuration
api_reload_snapshots(authorized_client)
snaps = api_snapshots(authorized_client)
assert len(snaps) == 1
def test_remove(authorized_client, generic_userdata, backups): def test_remove(authorized_client, generic_userdata, backups):
response = api_remove(authorized_client) response = api_remove(authorized_client)
data = get_data(response)["backup"]["removeRepository"] data = get_data(response)["backup"]["removeRepository"]