feature(backups):add a function to set provider from env
continuous-integration/drone/push Build is failing Details

backups-unlock
Houkime 2023-07-26 11:54:17 +00:00
parent e9bb6d9973
commit aa7cc71557
2 changed files with 63 additions and 1 deletions

View File

@ -2,6 +2,7 @@
This module contains the controller class for backups. This module contains the controller class for backups.
""" """
from datetime import datetime, timedelta from datetime import datetime, timedelta
import os
from os import statvfs from os import statvfs
from typing import List, Optional from typing import List, Optional
@ -43,6 +44,13 @@ DEFAULT_JSON_PROVIDER = {
"bucket": "", "bucket": "",
} }
BACKUP_PROVIDER_ENVS = {
"kind": "BACKUP_KIND",
"login": "BACKUP_LOGIN",
"key": "BACKUP_KEY",
"location": "BACKUP_LOCATION",
}
class NotDeadError(AssertionError): class NotDeadError(AssertionError):
""" """
@ -132,6 +140,24 @@ class Backups:
Storage.store_provider(none_provider) Storage.store_provider(none_provider)
return none_provider return none_provider
@staticmethod
def set_provider_from_envs():
for env in BACKUP_PROVIDER_ENVS.values():
if env not in os.environ.keys():
raise ValueError(
f"Cannot set backup provider from envs, there is no {env} set"
)
kind_str = os.environ[BACKUP_PROVIDER_ENVS["kind"]]
kind_enum = BackupProviderEnum[kind_str]
provider = Backups._construct_provider(
kind=kind_enum,
login=os.environ[BACKUP_PROVIDER_ENVS["login"]],
key=os.environ[BACKUP_PROVIDER_ENVS["key"]],
location=os.environ[BACKUP_PROVIDER_ENVS["location"]],
)
Storage.store_provider(provider)
@staticmethod @staticmethod
def _construct_provider( def _construct_provider(
kind: BackupProviderEnum, kind: BackupProviderEnum,

View File

@ -1,4 +1,5 @@
import pytest import pytest
import os
import os.path as path import os.path as path
from os import makedirs from os import makedirs
from os import remove from os import remove
@ -18,10 +19,11 @@ from selfprivacy_api.jobs import Jobs, JobStatus
from selfprivacy_api.models.backup.snapshot import Snapshot from selfprivacy_api.models.backup.snapshot import Snapshot
from selfprivacy_api.backup import Backups from selfprivacy_api.backup import Backups, BACKUP_PROVIDER_ENVS
import selfprivacy_api.backup.providers as providers import selfprivacy_api.backup.providers as providers
from selfprivacy_api.backup.providers import AbstractBackupProvider from selfprivacy_api.backup.providers import AbstractBackupProvider
from selfprivacy_api.backup.providers.backblaze import Backblaze from selfprivacy_api.backup.providers.backblaze import Backblaze
from selfprivacy_api.backup.providers.none import NoBackups
from selfprivacy_api.backup.util import sync from selfprivacy_api.backup.util import sync
from selfprivacy_api.backup.backuppers.restic_backupper import ResticBackupper from selfprivacy_api.backup.backuppers.restic_backupper import ResticBackupper
from selfprivacy_api.backup.jobs import add_backup_job, add_restore_job from selfprivacy_api.backup.jobs import add_backup_job, add_restore_job
@ -129,6 +131,40 @@ def test_config_load(generic_userdata):
assert provider.backupper.key == "KEY" assert provider.backupper.key == "KEY"
def test_reset_sets_to_none1():
Backups.reset()
provider = Backups.provider()
assert provider is not None
assert isinstance(provider, NoBackups)
def test_reset_sets_to_none2(backups):
# now with something set up first^^^
Backups.reset()
provider = Backups.provider()
assert provider is not None
assert isinstance(provider, NoBackups)
def test_setting_from_envs(tmpdir):
Backups.reset()
os.environ[BACKUP_PROVIDER_ENVS["kind"]] = "BACKBLAZE"
os.environ[BACKUP_PROVIDER_ENVS["login"]] = "ID"
os.environ[BACKUP_PROVIDER_ENVS["key"]] = "KEY"
os.environ[BACKUP_PROVIDER_ENVS["location"]] = "selfprivacy"
Backups.set_provider_from_envs()
provider = Backups.provider()
assert provider is not None
assert isinstance(provider, Backblaze)
assert provider.login == "ID"
assert provider.key == "KEY"
assert provider.location == "selfprivacy"
assert provider.backupper.account == "ID"
assert provider.backupper.key == "KEY"
def test_json_reset(generic_userdata): def test_json_reset(generic_userdata):
Backups.reset(reset_json=False) Backups.reset(reset_json=False)
provider = Backups.provider() provider = Backups.provider()