Increasing code coverage by adding extra unit tests (#444)

* First batch of tests

* Added API tests, continued on file based tests

* Got 50% test coverage - needs to be addressed further

mysql_autoxtrabackup/backup_backup/backup_builder.py

@@ -214,7 +214,7 @@ def inc_backup_command_builder(
         self,
         recent_full_bck: Optional[str],
         inc_backup_dir: Optional[str],
-        recent_inc_bck: Union[str, None] = None,
+        recent_inc_bck: Optional[str] = None,
     ) -> str:
         xtrabackup_inc_cmd_base = (
             "{} --defaults-file={} --user={} --password={}".format(

mysql_autoxtrabackup/backup_backup/backuper.py

@@ -94,8 +94,9 @@ def add_tag(
         return True
 
     @staticmethod
-    def show_tags(backup_dir: str) -> None:
-        if os.path.isfile("{}/backup_tags.txt".format(backup_dir)):
+    def show_tags(backup_dir: str, tag_file: Optional[str] = None) -> Optional[bool]:
+        tag_file = tag_file or "{}/backup_tags.txt".format(backup_dir)
+        if os.path.isfile(tag_file):
             with open("{}/backup_tags.txt".format(backup_dir), "r") as backup_tags:
                 from_file = backup_tags.read()
             column_names = "{0}\t{1}\t{2}\t{3}\t{4}\tTAG\n".format(
@@ -108,6 +109,7 @@ def show_tags(backup_dir: str) -> None:
             extra_str = "{}\n".format("-" * (len(column_names) + 21))
             print(column_names + extra_str + from_file)
             logger.info(column_names + extra_str + from_file)
+            return True
         else:
             logger.warning(
                 "Could not find backup_tags.txt inside given backup directory. Can't print tags."
@@ -116,28 +118,29 @@ def show_tags(backup_dir: str) -> None:
                 "WARNING: Could not find backup_tags.txt inside given backup directory. Can't print tags."
             )
 
-    def last_full_backup_date(self) -> bool:
+    def last_full_backup_date(self, path: Optional[str] = None, full_backup_interval: Optional[float] = None) -> bool:
         """
         Check if last full backup date retired or not.
         :return: True if last full backup date older than given interval, False if it is newer.
         """
         # Finding last full backup date from dir/folder name
-        max_dir = helpers.get_latest_dir_name(
-            str(self.builder_obj.backup_options.get("full_dir"))
-        )
+        full_dir = path or str(self.builder_obj.backup_options.get("full_dir"))
+        backup_interval = full_backup_interval or str(self.builder_obj.backup_options.get("full_backup_interval"))
+        max_dir = helpers.get_latest_dir_name(full_dir)
+
         dir_date = datetime.strptime(str(max_dir), "%Y-%m-%d_%H-%M-%S")
         now = datetime.now()
-        return float((now - dir_date).total_seconds()) >= float(
-            str(self.builder_obj.backup_options.get("full_backup_interval"))
-        )
+        return float((now - dir_date).total_seconds()) >= float(backup_interval)
 
     def clean_full_backup_dir(
-        self, remove_all: Union[bool, None] = None
-    ) -> Union[None, bool]:
+        self,
+        full_dir: Optional[str] = None,
+        remove_all: Optional[bool] = None,
+    ) -> Optional[bool]:
         # Deleting old full backup after taking new full backup.
         # Keeping the latest in order not to lose everything.
         logger.info("starting clean_full_backup_dir")
-        full_dir = str(self.builder_obj.backup_options.get("full_dir"))
+        full_dir = full_dir or str(self.builder_obj.backup_options.get("full_dir"))
         if not os.path.isdir(full_dir):
             return True
         if remove_all:
@@ -155,9 +158,9 @@ def clean_full_backup_dir(
                 logger.info("KEEPING {}".format(rm_dir))
         return True
 
-    def clean_inc_backup_dir(self) -> Union[None, bool]:
+    def clean_inc_backup_dir(self, inc_dir: Optional[str] = None) -> Optional[bool]:
         # Deleting incremental backups after taking new fresh full backup.
-        inc_dir = str(self.builder_obj.backup_options.get("inc_dir"))
+        inc_dir = inc_dir or str(self.builder_obj.backup_options.get("inc_dir"))
         if not os.path.isdir(inc_dir):
             return True
         for i in os.listdir(inc_dir):
@@ -214,6 +217,9 @@ def inc_backup(self) -> bool:
         recent_full_bck = helpers.get_latest_dir_name(
             str(self.builder_obj.backup_options.get("full_dir"))
         )
+        if not recent_full_bck:
+            raise RuntimeError("Failed to get Full backup path. Are you sure you have one?")
+
         # Get the recent incremental backup path
         recent_inc_bck = helpers.get_latest_dir_name(
             str(self.builder_obj.backup_options.get("inc_dir"))

mysql_autoxtrabackup/utils/helpers.py

@@ -51,13 +51,16 @@ def get_directory_size(path: str) -> int:
     return total_size
 
 
-def create_backup_directory(directory: str) -> str:
+def create_backup_directory(directory: str, forced_dir: Optional[str] = None) -> str:
     """
     Function for creating timestamped directory on given path
     :param directory: Directory path
+    :param forced_dir: Full Directory path forced to be created
     :return: Created new directory path
     """
     new_dir = os.path.join(directory, datetime.now().strftime("%Y-%m-%d_%H-%M-%S"))
+    if forced_dir:
+        new_dir = os.path.join(directory, forced_dir)
     try:
         # Creating directory
         os.makedirs(new_dir)

tests/conftest.py

@@ -1,9 +1,18 @@
 import pytest
 from mysql_autoxtrabackup.backup_backup.backuper import Backup
+from mysql_autoxtrabackup.api.main import app
+from fastapi.testclient import TestClient
+
 
 bck_obj = Backup()
+client = TestClient(app)
 
 
 @pytest.fixture()
 def return_bck_obj():
     return bck_obj
+
+
+@pytest.fixture()
+def fastapi_client():
+    return client

tests/test_api.py

@@ -0,0 +1,21 @@
+
+class TestAPI:
+
+    def test_take_backup(self, fastapi_client):
+        response = fastapi_client.post('/backup')
+        assert response.status_code == 201
+        assert response.json() == {"result":"Successfully finished the backup process"}
+
+    def test_prepare_backup(self, fastapi_client):
+        response = fastapi_client.post('/prepare')
+        assert response.status_code == 200
+        assert response.json() == {"result":"Successfully prepared all the backups"}
+
+    def test_list_backups(self, fastapi_client):
+        response = fastapi_client.get('/backups')
+        assert response.status_code == 200
+
+    def test_delete_backups(self, fastapi_client):
+        response = fastapi_client.delete('/delete')
+        assert response.status_code == 200
+        assert response.json() == {"result":"There is no backups or backups removed successfully"}

tests/test_backup.py

@@ -6,10 +6,6 @@
 
 @pytest.mark.usefixtures("return_bck_obj")
 class TestBackup:
-    def test_create_mysql_client_command(self, return_bck_obj):
-        result = '/usr/bin/mysql --defaults-file= -uroot --password=12345 --socket=/var/lib/mysql/mysql.sock -e "select 1"'
-        sql = "select 1"
-        assert return_bck_obj.create_mysql_client_command(sql) == result
 
     def test_full_backup_without_tag(self, return_bck_obj):
         return_bck_obj.clean_full_backup_dir()
@@ -23,6 +19,54 @@ def test_full_backup_with_tag(self, return_bck_obj):
         # Making it None back for global object
         return_bck_obj.tag = None
         # Check if the backup tag file is created and contains given string
-        assert os.path.isfile("{}/backup_tags.txt".format(return_bck_obj.backupdir))
-        with open("{}/backup_tags.txt".format(return_bck_obj.backupdir), "r") as file:
+        assert os.path.isfile("{}/backup_tags.txt".format(return_bck_obj.builder_obj.backup_options.get('backup_dir')))
+        with open("{}/backup_tags.txt".format(return_bck_obj.builder_obj.backup_options.get('backup_dir')), "r") as file:
             assert "My first full backup" in file.read()
+
+    def test_full_backup_dry_run(self, return_bck_obj):
+        return_bck_obj.dry = True
+        assert return_bck_obj.full_backup() is True
+
+    def test_show_tags_with_wrong_file_name(self, return_bck_obj):
+        assert return_bck_obj.show_tags(return_bck_obj.builder_obj.backup_options.get('backup_dir'), "dummy.txt") \
+               is None
+
+    def test_show_tags_with_correct_file_name(self, return_bck_obj):
+        assert return_bck_obj.show_tags(return_bck_obj.builder_obj.backup_options.get('backup_dir')) is True
+
+    def test_last_full_backup_date(self, return_bck_obj):
+        os.makedirs('tests/DELETE_ME', mode=777, exist_ok=True)
+        os.makedirs('tests/DELETE_ME/2021-05-06_11-48-31', mode=777, exist_ok=True)
+        assert return_bck_obj.last_full_backup_date(path=f'{os.path.dirname(__file__)}/DELETE_ME',
+                                                    full_backup_interval=60) is True
+        assert return_bck_obj.last_full_backup_date(path=f'{os.path.dirname(__file__)}/DELETE_ME',
+                                                    full_backup_interval=6000000) is False
+
+    def test_clean_full_backup_dir_dummy_path(self, return_bck_obj):
+        assert return_bck_obj.clean_full_backup_dir(full_dir='NON_EXISTING_PATH_NAME') is True
+
+    def test_clean_full_backup_dir_real_path(self, return_bck_obj):
+        os.makedirs('tests/DELETE_ME', mode=777, exist_ok=True)
+        os.makedirs('tests/DELETE_ME/2021-05-06_11-48-31', mode=777, exist_ok=True)
+        os.makedirs('tests/DELETE_ME/2021-05-06_11-47-31', mode=777, exist_ok=True)
+        assert return_bck_obj.clean_full_backup_dir(full_dir=f'{os.path.dirname(__file__)}/DELETE_ME') is True
+        for file_ in os.listdir(f'{os.path.dirname(__file__)}/DELETE_ME'):
+            assert file_ == '2021-05-06_11-48-31'
+
+    def test_clean_full_backup_dir_with_remove_all(self, return_bck_obj):
+        os.makedirs('tests/DELETE_ME', mode=777, exist_ok=True)
+        os.makedirs('tests/DELETE_ME/2021-05-06_11-48-31', mode=777, exist_ok=True)
+        os.makedirs('tests/DELETE_ME/2021-05-06_11-47-31', mode=777, exist_ok=True)
+        assert return_bck_obj.clean_full_backup_dir(full_dir=f'{os.path.dirname(__file__)}/DELETE_ME',
+                                                    remove_all=True) is True
+        assert len(os.listdir(f'{os.path.dirname(__file__)}/DELETE_ME')) == 0
+
+    def test_clean_inc_backup_dir_with_dummy_path(self, return_bck_obj):
+        assert return_bck_obj.clean_inc_backup_dir(inc_dir='NON_EXISTING_PATH_NAME') is True
+
+    def test_clean_inc_backup_dir_real_path(self, return_bck_obj):
+        os.makedirs('tests/DELETE_ME', mode=777, exist_ok=True)
+        os.makedirs('tests/DELETE_ME/2021-05-06_11-48-31', mode=777, exist_ok=True)
+        os.makedirs('tests/DELETE_ME/2021-05-06_11-47-31', mode=777, exist_ok=True)
+        assert return_bck_obj.clean_inc_backup_dir(inc_dir=f'{os.path.dirname(__file__)}/DELETE_ME') is True
+        assert len(os.listdir(f'{os.path.dirname(__file__)}/DELETE_ME')) == 0

tests/test_helpers.py

@@ -0,0 +1,22 @@
+import shutil
+
+import pytest
+import os
+
+from mysql_autoxtrabackup.utils import helpers
+
+
+class TestHelpers:
+
+    def test_get_latest_dir_name(self):
+        os.makedirs('tests/DELETE_ME', mode=777, exist_ok=True)
+        os.makedirs('tests/DELETE_ME/2021-05-06_11-48-31', mode=777, exist_ok=True)
+        os.makedirs('tests/DELETE_ME/2021-05-06_11-47-31', mode=777, exist_ok=True)
+
+        assert helpers.get_latest_dir_name(path=f'{os.path.dirname(__file__)}/DELETE_ME') == '2021-05-06_11-48-31'
+
+    def test_create_backup_directory(self):
+        path_ = f'{os.path.dirname(__file__)}/DELETE_ME'
+        assert helpers.create_backup_directory(path_, 'TEST_DIR') == f'{path_}/TEST_DIR'
+        shutil.rmtree(f'{path_}/TEST_DIR')
+

tests/test_mysql_cli.py

@@ -0,0 +1,13 @@
+import os
+import pytest
+
+
+class TestMySQLCLi:
+    def test_create_mysql_client_command(self, return_bck_obj):
+        result = '/usr/bin/mysql --defaults-file= -uroot --password=12345 --socket=/var/run/mysqld/mysqld.sock -e "select 1"'
+        sql = "select 1"
+        assert return_bck_obj.mysql_cli.create_mysql_client_command(sql) == result
+
+    def test_mysql_run_command(self, return_bck_obj):
+        sql = "select 1"
+        assert return_bck_obj.mysql_cli.mysql_run_command(sql) is True