Merge pull request #227 from rafsaf/improve-tests

Improve tests

backuper/backup_targets/file.py

@@ -17,9 +17,9 @@ def __init__(self, target_model: SingleFileTargetModel) -> None:
         self.target_model: SingleFileTargetModel = target_model
 
     def _backup(self) -> Path:
-        out_file = core.get_new_backup_path(
-            self.env_name, self.target_model.abs_path.name
-        )
+        escaped_filename = core.safe_text_version(self.target_model.abs_path.name)
+
+        out_file = core.get_new_backup_path(self.env_name, escaped_filename)
 
         shell_create_file_symlink = f"ln -s {self.target_model.abs_path} {out_file}"
         log.debug("start ln in subprocess: %s", shell_create_file_symlink)

backuper/backup_targets/folder.py

@@ -17,9 +17,9 @@ def __init__(self, target_model: DirectoryTargetModel) -> None:
         self.target_model: DirectoryTargetModel = target_model
 
     def _backup(self) -> Path:
-        out_file = core.get_new_backup_path(
-            self.env_name, self.target_model.abs_path.name
-        )
+        escaped_foldername = core.safe_text_version(self.target_model.abs_path.name)
+
+        out_file = core.get_new_backup_path(self.env_name, escaped_foldername)
 
         shell_create_dir_symlink = f"ln -s {self.target_model.abs_path} {out_file}"
         log.debug("start ln in subprocess: %s", shell_create_dir_symlink)

backuper/backup_targets/mariadb.py

@@ -24,6 +24,7 @@ class MariaDB(BaseBackupTarget):
     def __init__(self, target_model: MariaDBTargetModel) -> None:
         super().__init__(target_model)
         self.target_model: MariaDBTargetModel = target_model
+        self.db_name = shlex.quote(self.target_model.db)
         self.option_file: Path = self._init_option_file()
         self.db_version: str = self._mariadb_connection()
 
@@ -65,9 +66,8 @@ def _mariadb_connection(self) -> str:
             raise
         log.debug("start mariadb connection")
         try:
-            db = shlex.quote(self.target_model.db)
             result = core.run_subprocess(
-                f"mariadb --defaults-file={self.option_file} {db} "
+                f"mariadb --defaults-file={self.option_file} {self.db_name} "
                 f"--execute='SELECT version();'",
             )
         except core.CoreSubprocessError as conn_err:
@@ -93,12 +93,14 @@ def _mariadb_connection(self) -> str:
 
     def _backup(self) -> Path:
         escaped_dbname = core.safe_text_version(self.target_model.db)
-        name = f"{escaped_dbname}_{self.db_version}"
-        out_file = core.get_new_backup_path(self.env_name, name, sql=True)
-        db = shlex.quote(self.target_model.db)
+        escaped_version = core.safe_text_version(self.db_version)
+        name = f"{escaped_dbname}_{escaped_version}"
+
+        out_file = core.get_new_backup_path(self.env_name, name).with_suffix(".sql")
+
         shell_mariadb_dump_db = (
             f"mariadb-dump --defaults-file={self.option_file} "
-            f"--result-file={out_file} --verbose {db}"
+            f"--result-file={out_file} --verbose {self.db_name}"
         )
         log.debug("start mariadbdump in subprocess: %s", shell_mariadb_dump_db)
         core.run_subprocess(shell_mariadb_dump_db)

backuper/backup_targets/mysql.py

@@ -24,6 +24,7 @@ class MySQL(BaseBackupTarget):
     def __init__(self, target_model: MySQLTargetModel) -> None:
         super().__init__(target_model)
         self.target_model: MySQLTargetModel = target_model
+        self.db_name = shlex.quote(self.target_model.db)
         self.option_file: Path = self._init_option_file()
         self.db_version: str = self._mysql_connection()
 
@@ -65,9 +66,8 @@ def _mysql_connection(self) -> str:
             raise
         log.debug("start mysql connection")
         try:
-            db = shlex.quote(self.target_model.db)
             result = core.run_subprocess(
-                f"mariadb --defaults-file={self.option_file} {db} "
+                f"mariadb --defaults-file={self.option_file} {self.db_name} "
                 "--execute='SELECT version();'",
             )
         except core.CoreSubprocessError as err:
@@ -93,13 +93,14 @@ def _mysql_connection(self) -> str:
 
     def _backup(self) -> Path:
         escaped_dbname = core.safe_text_version(self.target_model.db)
-        name = f"{escaped_dbname}_{self.db_version}"
-        out_file = core.get_new_backup_path(self.env_name, name, sql=True)
+        escaped_version = core.safe_text_version(self.db_version)
+        name = f"{escaped_dbname}_{escaped_version}"
+
+        out_file = core.get_new_backup_path(self.env_name, name).with_suffix(".sql")
 
-        db = shlex.quote(self.target_model.db)
         shell_mysqldump_db = (
             f"mariadb-dump --defaults-file={self.option_file} "
-            f"--result-file={out_file} --verbose {db}"
+            f"--result-file={out_file} --verbose {self.db_name}"
         )
         log.debug("start mysqldump in subprocess: %s", shell_mysqldump_db)
         core.run_subprocess(shell_mysqldump_db)

backuper/backup_targets/postgresql.py

@@ -109,8 +109,11 @@ def _postgres_connection(self) -> str:
 
     def _backup(self) -> Path:
         escaped_dbname = core.safe_text_version(self.target_model.db)
-        name = f"{escaped_dbname}_{self.db_version}"
-        out_file = core.get_new_backup_path(self.env_name, name, sql=True)
+        escaped_version = core.safe_text_version(self.db_version)
+        name = f"{escaped_dbname}_{escaped_version}"
+
+        out_file = core.get_new_backup_path(self.env_name, name).with_suffix(".sql")
+
         shell_pg_dump_db = (
             f"pg_dump --clean --if-exists -v -O -d "
             f"{self.escaped_conn_uri} -f {out_file}"

backuper/core.py

@@ -61,17 +61,15 @@ def remove_path(path: Path) -> None:
             shutil.rmtree(path=path)
 
 
-def get_new_backup_path(env_name: str, name: str, sql: bool = False) -> Path:
+def get_new_backup_path(env_name: str, name: str) -> Path:
     base_dir_path = config.CONST_BACKUP_FOLDER_PATH / env_name
     base_dir_path.mkdir(mode=0o700, exist_ok=True, parents=True)
     new_file = (
         f"{env_name}_"
         f"{datetime.now(UTC).strftime('%Y%m%d_%H%M')}_"
         f"{name}_"
-        f"{secrets.token_urlsafe(3)}"
+        f"{secrets.token_urlsafe(6)}"
     )
-    if sql:
-        new_file += ".sql"
     return base_dir_path / new_file
 
 

backuper/models/backup_target_models.py

@@ -7,6 +7,7 @@
 from croniter import croniter
 from pydantic import (
     BaseModel,
+    ConfigDict,
     Field,
     SecretStr,
     field_validator,
@@ -25,6 +26,8 @@ class TargetModel(BaseModel):
         ge=0, le=36600, default=config.options.BACKUP_MIN_RETENTION_DAYS
     )
 
+    model_config = ConfigDict(frozen=True)
+
     @field_validator("cron_rule")
     def cron_rule_is_valid(cls, cron_rule: str) -> str:
         if not croniter.is_valid(cron_rule):

backuper/models/upload_provider_models.py

@@ -3,14 +3,16 @@
 
 import base64
 
-from pydantic import BaseModel, SecretStr, field_validator
+from pydantic import BaseModel, ConfigDict, SecretStr, field_validator
 
 from backuper import config
 
 
 class ProviderModel(BaseModel):
     name: str
 
+    model_config = ConfigDict(frozen=True)
+
 
 class DebugProviderModel(ProviderModel):
     name: str = config.UploadProviderEnum.LOCAL_FILES_DEBUG

tests/test_backup_target_file.py

@@ -0,0 +1,31 @@
+# Copyright: (c) 2024, Rafał Safin <rafal.safin@rafsaf.pl>
+# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+
+from freezegun import freeze_time
+
+from backuper import config
+from backuper.backup_targets.file import File
+
+from .conftest import CONST_TOKEN_URLSAFE, FILE_1
+
+
+@freeze_time("2024-03-14")
+def test_run_file_backup_output_file_has_proper_name() -> None:
+    file = File(target_model=FILE_1)
+    out_backup = file.make_backup()
+
+    escaped_file_name = FILE_1.abs_path.name.replace(".", "")
+    out_file = (
+        f"{file.env_name}/"
+        f"{file.env_name}_20240314_0000_{escaped_file_name}_{CONST_TOKEN_URLSAFE}"
+    )
+    out_path = config.CONST_BACKUP_FOLDER_PATH / out_file
+    assert out_backup == out_path
+
+
+def test_run_file_backup_output_file_has_exact_same_content() -> None:
+    file = File(target_model=FILE_1)
+    out_backup = file.make_backup()
+
+    assert out_backup.read_text() == FILE_1.abs_path.read_text()

tests/test_backup_target_folder.py

@@ -0,0 +1,34 @@
+# Copyright: (c) 2024, Rafał Safin <rafal.safin@rafsaf.pl>
+# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+
+from freezegun import freeze_time
+
+from backuper import config
+from backuper.backup_targets.folder import Folder
+
+from .conftest import CONST_TOKEN_URLSAFE, FOLDER_1
+
+
+@freeze_time("2024-03-14")
+def test_run_folder_backup_output_folder_has_proper_name() -> None:
+    folder = Folder(target_model=FOLDER_1)
+    out_backup = folder.make_backup()
+
+    folder_name = FOLDER_1.abs_path.name
+    out_file = (
+        f"{folder.env_name}/"
+        f"{folder.env_name}_20240314_0000_{folder_name}_{CONST_TOKEN_URLSAFE}"
+    )
+    out_path = config.CONST_BACKUP_FOLDER_PATH / out_file
+    assert out_backup == out_path
+
+
+def test_run_folder_backup_output_file_in_folder_has_exact_same_content() -> None:
+    folder = Folder(target_model=FOLDER_1)
+    out_backup = folder.make_backup()
+
+    file_in_folder = FOLDER_1.abs_path / "file.txt"
+    file_in_out_folder = out_backup / "file.txt"
+
+    assert file_in_folder.read_text() == file_in_out_folder.read_text()

tests/test_backup_target_mariadb.py

@@ -1,10 +1,12 @@
 # Copyright: (c) 2024, Rafał Safin <rafal.safin@rafsaf.pl>
 # GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt)
 
-from unittest.mock import Mock
+
+import shlex
 
 import pytest
 from freezegun import freeze_time
+from pydantic import SecretStr
 
 from backuper import config, core
 from backuper.backup_targets.mariadb import MariaDB
@@ -24,31 +26,93 @@ def test_mariadb_connection_success(mariadb_target: MariaDBTargetModel) -> None:
 
 
 @pytest.mark.parametrize("mariadb_target", ALL_MARIADB_DBS_TARGETS)
-def test_mariadb_connection_fail(
-    mariadb_target: MariaDBTargetModel,
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
+def test_mariadb_connection_fail(mariadb_target: MariaDBTargetModel) -> None:
     with pytest.raises(core.CoreSubprocessError):
         # simulate not existing db port 9999 and connection err
-        monkeypatch.setattr(mariadb_target, "port", 9999)
-        MariaDB(target_model=mariadb_target)
+        target_model = mariadb_target.model_copy(update={"port": 9999})
+        MariaDB(target_model=target_model)
 
 
 @freeze_time("2022-12-11")
 @pytest.mark.parametrize("mariadb_target", ALL_MARIADB_DBS_TARGETS)
-def test_run_mariadb_dump(
-    mariadb_target: MariaDBTargetModel,
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    mock = Mock(return_value="fixed_dbname")
-    monkeypatch.setattr(core, "safe_text_version", mock)
-
+def test_run_mariadb_dump(mariadb_target: MariaDBTargetModel) -> None:
     db = MariaDB(target_model=mariadb_target)
-    out_backup = db._backup()
+    out_backup = db.make_backup()
 
+    escaped_name = "database_12"
+    escaped_version = db.db_version.replace(".", "")
     out_file = (
         f"{db.env_name}/"
-        f"{db.env_name}_20221211_0000_fixed_dbname_{db.db_version}_{CONST_TOKEN_URLSAFE}.sql"
+        f"{db.env_name}_20221211_0000_{escaped_name}_{escaped_version}_{CONST_TOKEN_URLSAFE}.sql"
     )
     out_path = config.CONST_BACKUP_FOLDER_PATH / out_file
     assert out_backup == out_path
+
+
+@pytest.mark.parametrize("mariadb_target", ALL_MARIADB_DBS_TARGETS)
+def test_end_to_end_successful_restore_after_backup(
+    mariadb_target: MariaDBTargetModel,
+) -> None:
+    root_target_model = mariadb_target.model_copy(
+        update={
+            "user": "root",
+            "password": SecretStr(f"root-{mariadb_target.password.get_secret_value()}"),
+        }
+    )
+
+    db = MariaDB(target_model=root_target_model)
+    core.run_subprocess(
+        f"mariadb --defaults-file={db.option_file} {db.db_name} --execute="
+        "'DROP DATABASE IF EXISTS test_db;'",
+    )
+    core.run_subprocess(
+        f"mariadb --defaults-file={db.option_file} {db.db_name} --execute="
+        "'CREATE DATABASE test_db;'",
+    )
+
+    test_db_target = root_target_model.model_copy(update={"db": "test_db"})
+    test_db = MariaDB(target_model=test_db_target)
+
+    table_query = (
+        "CREATE TABLE my_table "
+        "(id SERIAL PRIMARY KEY, "
+        "name VARCHAR (50) UNIQUE NOT NULL, "
+        "age INTEGER);"
+    )
+    core.run_subprocess(
+        f"mariadb --defaults-file={test_db.option_file} {test_db.db_name} "
+        f"--execute='{table_query}'",
+    )
+
+    insert_query = shlex.quote(
+        "INSERT INTO my_table (name, age) "
+        "VALUES ('Geralt z Rivii', 60),('rafsaf', 24);"
+    )
+
+    core.run_subprocess(
+        f"mariadb --defaults-file={test_db.option_file} {test_db.db_name} "
+        f"--execute={insert_query}",
+    )
+
+    test_db_backup = test_db.make_backup()
+
+    core.run_subprocess(
+        f"mariadb --defaults-file={db.option_file} {db.db_name} --execute="
+        "'DROP DATABASE test_db;'",
+    )
+    core.run_subprocess(
+        f"mariadb --defaults-file={db.option_file} {db.db_name} --execute="
+        "'CREATE DATABASE test_db;'",
+    )
+
+    core.run_subprocess(
+        f"mariadb --defaults-file={test_db.option_file} {test_db.db_name}"
+        f" < {test_db_backup}",
+    )
+
+    result = core.run_subprocess(
+        f"mariadb --defaults-file={test_db.option_file} {test_db.db_name}"
+        " --execute='select * from my_table order by id asc;'",
+    )
+
+    assert result == ("id\tname\tage\n" "1\tGeralt z Rivii\t60\n" "2\trafsaf\t24\n")