test(backup): try to back up!

This commit is contained in:
Houkime 2023-02-08 14:05:25 +00:00
parent 6fa72dbac1
commit 17df21964a
4 changed files with 51 additions and 33 deletions

View file

@ -1,5 +1,10 @@
from abc import ABC
from abc import ABC, abstractmethod
class AbstractBackuper(ABC):
def __init__(self):
pass
@abstractmethod
def start_backup(self, folder: str):
raise NotImplementedError

View file

@ -18,44 +18,54 @@ class ResticBackuper(AbstractBackuper):
def restic_repo(self, repository_name: str) -> str:
# https://restic.readthedocs.io/en/latest/030_preparing_a_new_repo.html#other-services-via-rclone
# https://forum.rclone.org/t/can-rclone-be-run-solely-with-command-line-options-no-config-no-env-vars/6314/5
return f"rclone::{self.type}:{self._repository_name}/sfbackup"
return f"rclone::{self.type}:{repository_name}/sfbackup"
def rclone_args(self):
return "rclone.args=serve restic --stdio" + self.backend_rclone_args()
def backend_rclone_args(self, account: str, key: str) -> str:
def backend_rclone_args(self) -> str:
acc_arg = ""
key_arg = ""
if account != "":
acc_arg = f"{self.login_flag} {account}"
if key != "":
key_arg = f"{self.key_flag} {key}"
if self.account != "":
acc_arg = f"{self.login_flag} {self.account}"
if self.key != "":
key_arg = f"{self.key_flag} {self.key}"
return f"{acc_arg} {key_arg}"
def restic_command(self, account: str, key: str, *args):
return [
def restic_command(self, repo_name: str, *args):
command = [
"restic",
"-o",
self.rclone_args(),
"-r",
self.restic_repo(account, key),
].extend(args)
self.restic_repo(repo_name),
]
if args != []:
command.extend(args)
return command
def start_backup(self, folder: str):
def start_backup(self, folder: str, repo_name: str):
"""
Start backup with restic
"""
backup_command = self.restic_command(
self.account,
self.key,
repo_name,
"backup",
folder,
)
with open("/var/backup.log", "w", encoding="utf-8") as log_file:
subprocess.Popen(
backup_command,
shell=False,
stdout=log_file,
stderr=subprocess.STDOUT,
)
subprocess.Popen(
backup_command,
shell=False,
stderr=subprocess.STDOUT,
)
# TODO: we might want to provide logging facilities
# that are reroutable for testing
# with open("/var/backup.log", "w", encoding="utf-8") as log_file:
# subprocess.Popen(
# backup_command,
# shell=False,
# stdout=log_file,
# stderr=subprocess.STDOUT,
# )

View file

@ -14,8 +14,8 @@ from selfprivacy_api.services.test_service.icon import BITWARDEN_ICON
class DummyService(Service):
"""A test service"""
def __init__(self, location):
self.loccation = location
def __init_subclass__(cls, location):
cls.location = location
@staticmethod
def get_id() -> str:
@ -106,13 +106,9 @@ class DummyService(Service):
storage_usage = 0
return storage_usage
@staticmethod
def get_location() -> str:
with ReadUserData() as user_data:
if user_data.get("useBinds", False):
return user_data.get("bitwarden", {}).get("location", "sda1")
else:
return "sda1"
@classmethod
def get_location(cls) -> str:
return cls.location
@staticmethod
def get_dns_records() -> typing.List[ServiceDnsRecord]:

View file

@ -11,6 +11,7 @@ from selfprivacy_api.graphql.queries.providers import BackupProvider
TESTFILE_BODY = "testytest!"
REPO_NAME = "test_backup"
@pytest.fixture()
@ -18,11 +19,16 @@ def test_service(tmpdir):
testile_path = path.join(tmpdir, "testfile.txt")
with open(testile_path, "w") as file:
file.write(TESTFILE_BODY)
return DummyService(tmpdir)
# we need this to not change get_location() much
class TestDummyService (DummyService, location=tmpdir):
pass
return TestDummyService()
@pytest.fixture()
def memory_backup():
def memory_backup() -> AbstractBackupProvider:
ProviderClass = providers.get_provider(BackupProvider.MEMORY)
assert ProviderClass is not None
memory_provider = ProviderClass(login="", key="")
@ -40,3 +46,4 @@ def test_backup_service(test_service, memory_backup):
# temporarily incomplete
assert test_service is not None
assert memory_backup is not None
memory_backup.backuper.start_backup(test_service.get_location(), REPO_NAME)