mirror of
https://git.selfprivacy.org/SelfPrivacy/selfprivacy-rest-api.git
synced 2024-11-23 12:31:27 +00:00
c6919293b6
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.
373 lines
9.9 KiB
Python
373 lines
9.9 KiB
Python
from os import path
|
|
from tests.test_graphql.test_backup import dummy_service, backups, raw_dummy_service
|
|
from tests.common import generate_backup_query
|
|
|
|
|
|
from selfprivacy_api.graphql.common_types.service import service_to_graphql_service
|
|
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 {
|
|
provider
|
|
encryptionKey
|
|
isInitialized
|
|
autobackupPeriod
|
|
locationName
|
|
locationId
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
API_REMOVE_REPOSITORY_MUTATION = """
|
|
mutation TestRemoveRepo {
|
|
backup {
|
|
removeRepository {
|
|
success
|
|
message
|
|
code
|
|
configuration {
|
|
provider
|
|
encryptionKey
|
|
isInitialized
|
|
autobackupPeriod
|
|
locationName
|
|
locationId
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
API_INIT_MUTATION = """
|
|
mutation TestInitRepo($input: InitializeRepositoryInput!) {
|
|
backup {
|
|
initializeRepository(repository: $input) {
|
|
success
|
|
message
|
|
code
|
|
configuration {
|
|
provider
|
|
encryptionKey
|
|
isInitialized
|
|
autobackupPeriod
|
|
locationName
|
|
locationId
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
API_RESTORE_MUTATION = """
|
|
mutation TestRestoreService($snapshot_id: String!) {
|
|
backup {
|
|
restoreBackup(snapshotId: $snapshot_id) {
|
|
success
|
|
message
|
|
code
|
|
job {
|
|
uid
|
|
status
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
API_SNAPSHOTS_QUERY = """
|
|
allSnapshots {
|
|
id
|
|
service {
|
|
id
|
|
}
|
|
createdAt
|
|
}
|
|
"""
|
|
|
|
API_BACK_UP_MUTATION = """
|
|
mutation TestBackupService($service_id: String!) {
|
|
backup {
|
|
startBackup(serviceId: $service_id) {
|
|
success
|
|
message
|
|
code
|
|
job {
|
|
uid
|
|
status
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
|
|
def api_restore(authorized_client, snapshot_id):
|
|
response = authorized_client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": API_RESTORE_MUTATION,
|
|
"variables": {"snapshot_id": snapshot_id},
|
|
},
|
|
)
|
|
return response
|
|
|
|
|
|
def api_backup(authorized_client, service):
|
|
response = authorized_client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": API_BACK_UP_MUTATION,
|
|
"variables": {"service_id": service.get_id()},
|
|
},
|
|
)
|
|
return response
|
|
|
|
|
|
def api_set_period(authorized_client, period):
|
|
response = authorized_client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": API_SET_AUTOBACKUP_PERIOD_MUTATION,
|
|
"variables": {"period": period},
|
|
},
|
|
)
|
|
return response
|
|
|
|
|
|
def api_remove(authorized_client):
|
|
response = authorized_client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": API_REMOVE_REPOSITORY_MUTATION,
|
|
"variables": {},
|
|
},
|
|
)
|
|
return response
|
|
|
|
|
|
def api_reload_snapshots(authorized_client):
|
|
response = authorized_client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": API_RELOAD_SNAPSHOTS,
|
|
"variables": {},
|
|
},
|
|
)
|
|
return response
|
|
|
|
|
|
def api_init_without_key(
|
|
authorized_client, kind, login, password, location_name, location_id
|
|
):
|
|
response = authorized_client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": API_INIT_MUTATION,
|
|
"variables": {
|
|
"input": {
|
|
"provider": kind,
|
|
"locationId": location_id,
|
|
"locationName": location_name,
|
|
"login": login,
|
|
"password": password,
|
|
}
|
|
},
|
|
},
|
|
)
|
|
return response
|
|
|
|
|
|
def assert_ok(data):
|
|
assert data["code"] == 200
|
|
assert data["success"] is True
|
|
|
|
|
|
def get_data(response):
|
|
assert response.status_code == 200
|
|
response = response.json()
|
|
if (
|
|
"errors" in response.keys()
|
|
): # convenience for debugging, this will display error
|
|
assert response["errors"] == []
|
|
assert response["data"] is not None
|
|
data = response["data"]
|
|
return data
|
|
|
|
|
|
def api_snapshots(authorized_client):
|
|
response = authorized_client.post(
|
|
"/graphql",
|
|
json={"query": generate_backup_query([API_SNAPSHOTS_QUERY])},
|
|
)
|
|
data = get_data(response)
|
|
result = data["backup"]["allSnapshots"]
|
|
assert result is not None
|
|
return result
|
|
|
|
|
|
def test_dummy_service_convertible_to_gql(dummy_service):
|
|
gql_service = service_to_graphql_service(dummy_service)
|
|
assert gql_service is not None
|
|
|
|
|
|
def test_snapshots_empty(authorized_client, dummy_service):
|
|
snaps = api_snapshots(authorized_client)
|
|
assert snaps == []
|
|
|
|
|
|
def test_start_backup(authorized_client, dummy_service):
|
|
response = api_backup(authorized_client, dummy_service)
|
|
data = get_data(response)["backup"]["startBackup"]
|
|
assert data["success"] is True
|
|
job = data["job"]
|
|
|
|
assert Jobs.get_job(job["uid"]).status == JobStatus.FINISHED
|
|
snaps = api_snapshots(authorized_client)
|
|
assert len(snaps) == 1
|
|
snap = snaps[0]
|
|
|
|
assert snap["id"] is not None
|
|
assert snap["id"] != ""
|
|
assert snap["service"]["id"] == "testservice"
|
|
|
|
|
|
def test_restore(authorized_client, dummy_service):
|
|
api_backup(authorized_client, dummy_service)
|
|
snap = api_snapshots(authorized_client)[0]
|
|
assert snap["id"] is not None
|
|
|
|
response = api_restore(authorized_client, snap["id"])
|
|
data = get_data(response)["backup"]["restoreBackup"]
|
|
assert data["success"] is True
|
|
job = data["job"]
|
|
|
|
assert Jobs.get_job(job["uid"]).status == JobStatus.FINISHED
|
|
|
|
|
|
def test_reinit(authorized_client, dummy_service, tmpdir):
|
|
test_repo_path = path.join(tmpdir, "not_at_all_sus")
|
|
response = api_init_without_key(
|
|
authorized_client, "FILE", "", "", test_repo_path, ""
|
|
)
|
|
data = get_data(response)["backup"]["initializeRepository"]
|
|
assert_ok(data)
|
|
configuration = data["configuration"]
|
|
assert configuration["provider"] == "FILE"
|
|
assert configuration["locationId"] == ""
|
|
assert configuration["locationName"] == test_repo_path
|
|
assert len(configuration["encryptionKey"]) > 1
|
|
assert configuration["isInitialized"] is True
|
|
|
|
response = api_backup(authorized_client, dummy_service)
|
|
data = get_data(response)["backup"]["startBackup"]
|
|
assert data["success"] is True
|
|
job = data["job"]
|
|
|
|
assert Jobs.get_job(job["uid"]).status == JobStatus.FINISHED
|
|
|
|
|
|
def test_remove(authorized_client, generic_userdata):
|
|
response = api_remove(authorized_client)
|
|
data = get_data(response)["backup"]["removeRepository"]
|
|
assert_ok(data)
|
|
|
|
configuration = data["configuration"]
|
|
assert configuration["provider"] == "NONE"
|
|
assert configuration["locationId"] == ""
|
|
assert configuration["locationName"] == ""
|
|
# still generated every time it is missing
|
|
assert len(configuration["encryptionKey"]) > 1
|
|
assert configuration["isInitialized"] is False
|
|
|
|
|
|
def test_autobackup_period_nonzero(authorized_client):
|
|
new_period = 11
|
|
response = api_set_period(authorized_client, new_period)
|
|
data = get_data(response)["backup"]["setAutobackupPeriod"]
|
|
assert_ok(data)
|
|
|
|
configuration = data["configuration"]
|
|
assert configuration["autobackupPeriod"] == new_period
|
|
|
|
|
|
def test_autobackup_period_zero(authorized_client):
|
|
new_period = 0
|
|
# since it is none by default, we better first set it to something non-negative
|
|
response = api_set_period(authorized_client, 11)
|
|
# and now we nullify it
|
|
response = api_set_period(authorized_client, new_period)
|
|
data = get_data(response)["backup"]["setAutobackupPeriod"]
|
|
assert_ok(data)
|
|
|
|
configuration = data["configuration"]
|
|
assert configuration["autobackupPeriod"] == None
|
|
|
|
|
|
def test_autobackup_period_none(authorized_client):
|
|
# since it is none by default, we better first set it to something non-negative
|
|
response = api_set_period(authorized_client, 11)
|
|
# and now we nullify it
|
|
response = api_set_period(authorized_client, None)
|
|
data = get_data(response)["backup"]["setAutobackupPeriod"]
|
|
assert_ok(data)
|
|
|
|
configuration = data["configuration"]
|
|
assert configuration["autobackupPeriod"] == None
|
|
|
|
|
|
def test_autobackup_period_negative(authorized_client):
|
|
# since it is none by default, we better first set it to something non-negative
|
|
response = api_set_period(authorized_client, 11)
|
|
# and now we nullify it
|
|
response = api_set_period(authorized_client, -12)
|
|
data = get_data(response)["backup"]["setAutobackupPeriod"]
|
|
assert_ok(data)
|
|
|
|
configuration = data["configuration"]
|
|
assert configuration["autobackupPeriod"] == None
|
|
|
|
|
|
# We cannot really check the effect at this level, we leave it to backend tests
|
|
# But we still make it run in both empty and full scenarios and ask for snaps afterwards
|
|
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)["backup"]["forceSnapshotsReload"]
|
|
assert_ok(data)
|
|
|
|
snaps = api_snapshots(authorized_client)
|
|
assert snaps == []
|
|
|
|
|
|
def test_reload_snapshots(authorized_client, dummy_service):
|
|
response = api_backup(authorized_client, dummy_service)
|
|
data = get_data(response)["backup"]["startBackup"]
|
|
|
|
response = api_reload_snapshots(authorized_client)
|
|
data = get_data(response)["backup"]["forceSnapshotsReload"]
|
|
assert_ok(data)
|
|
|
|
snaps = api_snapshots(authorized_client)
|
|
assert len(snaps) == 1
|