From aa3dc5fb05de7994c9a460ba7d519fbc70e21f11 Mon Sep 17 00:00:00 2001 From: Houkime <> Date: Fri, 23 Feb 2024 16:45:59 +0000 Subject: [PATCH] fix(backups): fix wrong typing in autobackups --- selfprivacy_api/backup/tasks.py | 20 ++++++++++++++++---- tests/test_autobackup.py | 26 ++++++++++++++++++++++++++ 2 files changed, 42 insertions(+), 4 deletions(-) diff --git a/selfprivacy_api/backup/tasks.py b/selfprivacy_api/backup/tasks.py index 465b1a8..c0f6a1d 100644 --- a/selfprivacy_api/backup/tasks.py +++ b/selfprivacy_api/backup/tasks.py @@ -72,14 +72,26 @@ def restore_snapshot( return True -@huey.periodic_task(validate_datetime=validate_datetime) -def automatic_backup(): +def do_autobackup(): """ - The worker periodic task that starts the automatic backup process. + Body of autobackup task, broken out to test it + For some reason, we cannot launch periodic huey tasks + inside tests """ time = datetime.utcnow().replace(tzinfo=timezone.utc) for service in Backups.services_to_back_up(time): - start_backup(service, BackupReason.AUTO) + handle = start_backup(service.get_id(), BackupReason.AUTO) + # To be on safe side, we do not do it in parallel + handle(blocking=True) + + +@huey.periodic_task(validate_datetime=validate_datetime) +def automatic_backup() -> bool: + """ + The worker periodic task that starts the automatic backup process. + """ + do_autobackup() + return True @huey.periodic_task(crontab(hour="*/" + str(SNAPSHOT_CACHE_TTL_HOURS))) diff --git a/tests/test_autobackup.py b/tests/test_autobackup.py index 63c625f..410694b 100644 --- a/tests/test_autobackup.py +++ b/tests/test_autobackup.py @@ -14,9 +14,12 @@ from selfprivacy_api.graphql.common_types.backup import ( from selfprivacy_api.backup import Backups, Snapshot from selfprivacy_api.backup.tasks import ( prune_autobackup_snapshots, + automatic_backup, + do_autobackup, ) from tests.test_backup import backups +from tests.test_graphql.test_services import only_dummy_service def backuppable_services() -> list[Service]: @@ -63,6 +66,29 @@ def test_set_autobackup_period(backups): assert Backups.autobackup_period_minutes() is None +def test_autobackup_taskbody(backups, only_dummy_service): + # We cannot test the timed task itself, but we reduced it + # to one line, and we test this line here + dummy_service = only_dummy_service + now = datetime.now(timezone.utc) + backup_period = 13 # minutes + + assert Backups.get_all_snapshots() == [] + + Backups.set_autobackup_period_minutes(backup_period) + assert Backups.is_time_to_backup_service(dummy_service, now) + assert Backups.is_time_to_backup(now) + assert dummy_service in Backups.services_to_back_up(now) + assert len(Backups.services_to_back_up(now)) == 1 + + do_autobackup() + + snapshots = Backups.get_all_snapshots() + assert len(snapshots) == 1 + assert snapshots[0].service_name == dummy_service.get_id() + assert snapshots[0].reason == BackupReason.AUTO + + def test_autobackup_timer_periods(backups, dummy_service): now = datetime.now(timezone.utc) backup_period = 13 # minutes