fix(backups): fix wrong typing in autobackups
continuous-integration/drone/push Build is failing Details

pull/96/head
Houkime 2024-02-23 16:45:59 +00:00
parent 4757bedc4e
commit 52f8e283be
2 changed files with 42 additions and 4 deletions

View File

@ -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)))

View File

@ -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