SelfPrivacy/selfprivacy-rest-api
1
0
Fork 0
mirror of https://git.selfprivacy.org/SelfPrivacy/selfprivacy-rest-api.git synced 2024-11-18 16:39:13 +00:00
Code Issues Projects Releases Wiki Activity

feature(backup):remove code for finegrained autobackup control

Browse source
This commit is contained in:
Houkime 2023-07-19 15:35:24 +00:00
parent 466160dbf9
commit f4ac3d29a9
3 changed files with 29 additions and 92 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

58
selfprivacy_api/backup/__init__.py
View file

@ -398,27 +398,6 @@ class Backups:
# Autobackup
@staticmethod
def is_autobackup_enabled(service: Service) -> bool:
return Storage.is_autobackup_set(service.get_id())
@staticmethod
def enable_autobackup(service: Service) -> None:
Storage.set_autobackup(service)
@staticmethod
def disable_autobackup(service: Service) -> None:
"""also see disable_all_autobackup()"""
Storage.unset_autobackup(service)
@staticmethod
def disable_all_autobackup() -> None:
"""
Disables all automatic backing up,
but does not change per-service settings
"""
Storage.delete_backup_period()
@staticmethod
def autobackup_period_minutes() -> Optional[int]:
"""None means autobackup is disabled"""
@ -436,6 +415,14 @@ class Backups:
return
Storage.store_autobackup_period_minutes(minutes)
@staticmethod
def disable_all_autobackup() -> None:
"""
Disables all automatic backing up,
but does not change per-service settings
"""
Storage.delete_backup_period()
@staticmethod
def is_time_to_backup(time: datetime) -> bool:
"""
@ -443,19 +430,15 @@ class Backups:
of automatic backups
"""
return Backups._service_ids_to_back_up(time) != []
return Backups.services_to_back_up(time) != []
@staticmethod
def services_to_back_up(time: datetime) -> List[Service]:
result: list[Service] = []
for id in Backups._service_ids_to_back_up(time):
service = get_service_by_id(id)
if service is None:
raise ValueError(
"Cannot look up a service scheduled for backup!",
)
result.append(service)
return result
return [
service
for service in get_all_services()
if Backups.is_time_to_backup_service(service, time)
]
@staticmethod
def get_last_backed_up(service: Service) -> Optional[datetime]:
@ -463,11 +446,12 @@ class Backups:
return Storage.get_last_backup_time(service.get_id())
@staticmethod
def is_time_to_backup_service(service_id: str, time: datetime):
def is_time_to_backup_service(service: Service, time: datetime):
period = Backups.autobackup_period_minutes()
if period is None:
service_id = service.get_id()
if not service.can_be_backed_up():
return False
if not Storage.is_autobackup_set(service_id):
if period is None:
return False
last_backup = Storage.get_last_backup_time(service_id)
@ -479,12 +463,6 @@ class Backups:
return True
return False
@staticmethod
def _service_ids_to_back_up(time: datetime) -> List[str]:
# TODO: simplify in light that we do not use redis for this anymore
service_ids = [service.get_id() for service in get_all_services()]
return [id for id in service_ids if Backups.is_time_to_backup_service(id, time)]
# Helpers
@staticmethod

29
selfprivacy_api/backup/storage.py
View file

@ -20,7 +20,6 @@ from selfprivacy_api.backup.providers import get_kind
# a hack to store file path.
REDIS_SNAPSHOT_CACHE_EXPIRE_SECONDS = 24 * 60 * 60 # one day
REDIS_AUTOBACKUP_ENABLED_PREFIX = "backup:autobackup:services:"
REDIS_SNAPSHOTS_PREFIX = "backups:snapshots:"
REDIS_LAST_BACKUP_PREFIX = "backups:last-backed-up:"
REDIS_INITTED_CACHE_PREFIX = "backups:initted_services:"
@ -42,7 +41,6 @@ class Storage:
REDIS_INITTED_CACHE_PREFIX,
REDIS_SNAPSHOTS_PREFIX,
REDIS_LAST_BACKUP_PREFIX,
REDIS_AUTOBACKUP_ENABLED_PREFIX,
]
for prefix in prefixes_to_clean:
@ -54,12 +52,6 @@ class Storage:
for key in redis.keys(REDIS_SNAPSHOTS_PREFIX + "*"):
redis.delete(key)
@staticmethod
def services_with_autobackup() -> List[str]:
keys = redis.keys(REDIS_AUTOBACKUP_ENABLED_PREFIX + "*")
service_ids = [key.split(":")[-1] for key in keys]
return service_ids
@staticmethod
def __last_backup_key(service_id):
return REDIS_LAST_BACKUP_PREFIX + service_id
@ -115,27 +107,6 @@ class Storage:
result.append(snapshot)
return result
@staticmethod
def __autobackup_key(service_name: str) -> str:
return REDIS_AUTOBACKUP_ENABLED_PREFIX + service_name
@staticmethod
def set_autobackup(service: Service):
# shortcut this
redis.set(Storage.__autobackup_key(service.get_id()), 1)
@staticmethod
def unset_autobackup(service: Service):
"""also see disable_all_autobackup()"""
redis.delete(Storage.__autobackup_key(service.get_id()))
@staticmethod
def is_autobackup_set(service_name: str) -> bool:
service = get_service_by_id(service_name)
if service is None:
raise ValueError("nonexistent service: ", service_name)
return service.can_be_backed_up()
@staticmethod
def autobackup_period_minutes() -> Optional[int]:
"""None means autobackup is disabled"""

34
tests/test_graphql/test_backup.py
View file

@ -414,17 +414,6 @@ def test_restore_snapshot_task(
assert len(snaps) == 1
def test_autobackup_enable_service_storage(backups, dummy_service):
assert len(Storage.services_with_autobackup()) == 0
Backups.enable_autobackup(dummy_service)
assert len(Storage.services_with_autobackup()) == 1
assert Storage.services_with_autobackup()[0] == dummy_service.get_id()
Backups.disable_autobackup(dummy_service)
assert len(Storage.services_with_autobackup()) == 0
def test_set_autobackup_period(backups):
assert Backups.autobackup_period_minutes() is None
@ -449,7 +438,7 @@ def test_set_autobackup_period(backups):
def test_no_default_autobackup(backups, dummy_service):
now = datetime.now(timezone.utc)
assert not Backups.is_time_to_backup_service(dummy_service.get_id(), now)
assert not Backups.is_time_to_backup_service(dummy_service, now)
assert not Backups.is_time_to_backup(now)
@ -483,15 +472,15 @@ def test_autobackup_timer_periods(backups, dummy_service):
now = datetime.now(timezone.utc)
backup_period = 13 # minutes
assert not Backups.is_time_to_backup_service(dummy_service.get_id(), now)
assert not Backups.is_time_to_backup_service(dummy_service, now)
assert not Backups.is_time_to_backup(now)
Backups.set_autobackup_period_minutes(backup_period)
assert Backups.is_time_to_backup_service(dummy_service.get_id(), now)
assert Backups.is_time_to_backup_service(dummy_service, now)
assert Backups.is_time_to_backup(now)
Backups.set_autobackup_period_minutes(0)
assert not Backups.is_time_to_backup_service(dummy_service.get_id(), now)
assert not Backups.is_time_to_backup_service(dummy_service, now)
assert not Backups.is_time_to_backup(now)
@ -506,14 +495,14 @@ def test_autobackup_timer_enabling(backups, dummy_service):
) # there are other services too, not just our dummy
# not backuppable service is not backuppable even if period is set
assert not Backups.is_time_to_backup_service(dummy_service.get_id(), now)
assert not Backups.is_time_to_backup_service(dummy_service, now)
dummy_service.set_backuppable(True)
assert dummy_service.can_be_backed_up()
assert Backups.is_time_to_backup_service(dummy_service.get_id(), now)
assert Backups.is_time_to_backup_service(dummy_service, now)
Backups.disable_all_autobackup()
assert not Backups.is_time_to_backup_service(dummy_service.get_id(), now)
assert not Backups.is_time_to_backup_service(dummy_service, now)
assert not Backups.is_time_to_backup(now)
@ -521,21 +510,20 @@ def test_autobackup_timing(backups, dummy_service):
backup_period = 13 # minutes
now = datetime.now(timezone.utc)
Backups.enable_autobackup(dummy_service)
Backups.set_autobackup_period_minutes(backup_period)
assert Backups.is_time_to_backup_service(dummy_service.get_id(), now)
assert Backups.is_time_to_backup_service(dummy_service, now)
assert Backups.is_time_to_backup(now)
Backups.back_up(dummy_service)
now = datetime.now(timezone.utc)
assert not Backups.is_time_to_backup_service(dummy_service.get_id(), now)
assert not Backups.is_time_to_backup_service(dummy_service, now)
past = datetime.now(timezone.utc) - timedelta(minutes=1)
assert not Backups.is_time_to_backup_service(dummy_service.get_id(), past)
assert not Backups.is_time_to_backup_service(dummy_service, past)
future = datetime.now(timezone.utc) + timedelta(minutes=backup_period + 2)
assert Backups.is_time_to_backup_service(dummy_service.get_id(), future)
assert Backups.is_time_to_backup_service(dummy_service, future)
# Storage