Merge pull request #10 from gcarrarom/feature/undo-flag

Feature/undo flag

.gitignore

@@ -11,3 +11,4 @@ tests/__pycache__
 coverage.xml
 htmlcov
 demo.yml
+tests/*.bak

azure-pipelines.yml

@@ -50,6 +50,7 @@ steps:
     python -m pip install --upgrade pip
     pip install pytest-cov
     pip install .
+    pip install testfixtures
   displayName: 'Install dependencies'
 
 - script: |

kcleaner.py

@@ -5,6 +5,10 @@
 import yaml
 from pathlib import Path
 from iterfzf import iterfzf
+import datetime
+
+backup_limit = 10
+backup_date_format = '%Y-%m-%d_%H-%M-%S'
 
 def ask_yn(yn_question, default='n'):
     tries = 0
@@ -18,43 +22,100 @@ def ask_yn(yn_question, default='n'):
             break
     return response
 
+def check_and_cleanup_backups(filename):
+    logging.debug(f"Checking if there's not more than {backup_limit} backup files in this directory")
+    dirpath = os.path.dirname(os.path.abspath(filename))
+    logging.debug(f"Getting all the files in {dirpath}")
+    files = os.listdir(dirpath)
+    logging.debug(f"These are all the files in the directory:\n{files}")
+    logging.debug(f"Checking for all kcleaner backup files")
+    files = [item for item in files if "kcleaner.bak" in item]
+    logging.debug(f"These are the backup files in this folder:\n{files}")
+    if len(files) > 10:
+        logging.info(f"Cleaning up excess of backup files - we have {len(files)} already... - Removing the {len(files) - 10} oldest files")
+        files.sort()
+        for file in files[0:(len(files)-10)]:
+            logging.debug(f"Removing File {file}")
+            os.remove(f"{dirpath}/{file}")
+
 def update_file(filename, yamldoc):
-    test_file_exists(filename)
-    logging.debug("Opening write stream for file {filename}")
+    file_exists(filename)
+    if not file_exists(filename) and not "bak" in filename:
+        logging.error("Cannot work with an empty file!, please check the path of your config file.")
+    if "bak" in filename:
+        check_and_cleanup_backups(filename)
+    if yamldoc == None:
+        logging.error("Yaml Value cannot be 'None'!")
+        exit(30)
+    elif yamldoc == "":
+        logging.error("Yaml Value cannot be empty!")
+        exit(31)
+    try:
+        yaml.safe_load(yamldoc)
+    except:
+        logging.exception("Yaml value is not valid!")
+        exit(32)
+    logging.debug(f"Opening write stream for file {filename}")
     with open(filename, 'w') as stream:
         try:
-            logging.debug("Dumping new yaml doc into the config file")
+            logging.debug("Writing new yaml doc into the config file")
             yaml.dump(yamldoc, stream)
         except yaml.YAMLError as exc:
-            print(exc)
+            logging.exception("Exception occured while trying to write Yaml file")
 
 def get_file(filename):
     logging.debug(f'Trying to retrieve contents of file {filename}')
-    test_file_exists(filename)
+    if not file_exists(filename):
+        logging.error("Cannot work with an empty file!, please check the path of your config file.")
+        exit(10)
     with open(filename, 'r') as stream:
         try:
             config_file = yaml.safe_load(stream)
         except yaml.YAMLError as exc:
-            print(exc)
+            logging.exception("Exception occured while trying to load Yaml file")
     logging.debug(f'File Contents\n{config_file}')
     logging.debug(f'Type of the file contents: {type(config_file)}')
     if config_file == None:
-        print("Config File is empty! Can't use it.")
+        logging.error("Config File is empty! Can't use it.")
         exit(11)
     elif type(config_file) == str:
-        print("Config File is not a valid yaml file!")
+        logging.error("Config File is not a valid yaml file!")
         exit(12)
     return config_file
 
-def test_file_exists(filename):
+def file_exists(filename):
     logging.debug(f"checking if file {filename} exists...")
+    if filename == None:
+        logging.error("Filename cannot be 'None'")
+        exit(20)
+    elif filename == "":
+        logging.error("Filename cannot be empty!")
+        exit(21)
     exists = os.path.isfile(filename)
     if exists:
         logging.debug("File exists!")
     else:
-        print('Config File Not found!')
-        # Keep presets
-        exit(10)
+        logging.info('Config File Not found!')
+    return exists
+
+def get_backup(backup_path):
+    logging.debug(f"Checking all backups available in the directory {backup_path}")
+    files = os.listdir(backup_path)
+    logging.debug(f"These are all the files in the directory:\n{files}")
+    logging.debug(f"Checking for all kcleaner backup files")
+    files = [item for item in files if "kcleaner.bak" in item]
+    logging.debug(f"These are the backup files in this folder:\n{files}")
+    files.sort(reverse=True)
+    dates = []
+    for file in files:
+        dates.append((datetime.datetime.strptime("_".join(file.split('_')[0:2]), backup_date_format).strftime("%c")))
+    logging.debug(dates)
+    backup_to_use = iterfzf(dates)
+    logging.debug(f'Backup chosen: {backup_to_use}')
+    backup_file_to_use = f"{datetime.datetime.strptime(backup_to_use, '%c').strftime(backup_date_format)}_kcleaner.bak"
+    logging.debug(f'Backup file: {backup_file_to_use}')
+    return get_file(f"{backup_path}/{backup_file_to_use}")
+
 
 def remove_resource(config_file, removing_type):
     logging.debug(f"Started removal of {removing_type}")
@@ -68,7 +129,7 @@ def remove_resource(config_file, removing_type):
     resources_to_remove = (iterfzf(resources_name_list, multi=True))
     logging.debug('List of resources selected: {resources_to_remove}')
     if resources_to_remove == None:
-        print("No resources to remove selected!")
+        logging.error("No resources to remove selected!")
         exit()
 
     logging.debug(f"{len(config_file[removing_type])} {removing_type} before the removal")
@@ -81,13 +142,12 @@ def remove_resource(config_file, removing_type):
         logging.debug('Removing resources...')
         config_file[removing_type] = [item for item in config_file[removing_type] if item['name'] not in resources_to_remove]
     except KeyError:
-        print(f"Something went wrong!!")
+        logging.exception(f"Something went wrong!!")
 
     logging.debug(f"{len(config_file[removing_type])} {removing_type} in the end")
 
     return config_file
 
-
 @click.command()
 @click.argument(
     "resource", 
@@ -101,31 +161,54 @@ def remove_resource(config_file, removing_type):
     default='contexts'
 )
 @click.option(
-    '--kubeconfig', '-k', default=f'{Path.home()}/.kube/config'
-    )
+    '--kubeconfig', '-k', default=f'{Path.home()}/.kube/config',
+    help="path to the config file to clean"
+)
 @click.option(
     '--name', '-n',
     help='Name of the entry to remove',
 )
-def cli(resource, name, kubeconfig):
+@click.option(
+    '--undo', '-u',
+    help='Use this to roll back latest changes',
+    is_flag=True
+)
+@click.option(
+    '--debug', '-d',
+    help='Use this to see debug level messages',
+    is_flag=True
+)
+def cli(resource, name, kubeconfig, undo, debug):
     """
     A little CLI tool to help keeping Config Files clean :)
     """
-    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
-    if resource == None:
-        resource = "clusters"
-    logging.debug(f'Using resource {resource}')
-    logging.debug(f'Config file to use: {kubeconfig}')
-    if name == None:
-        logging.debug(f'Name is empty, using fzf to search for the resource to remove')
+    if debug:
+        logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
     else:
-        logging.debug(f'Name of the resource requested to remove: {name}')
+        logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s')
 
-    config_file = get_file(kubeconfig)
-    config_file = remove_resource(config_file, resource)
-
+    kubeconfig_dir = os.path.dirname(os.path.abspath(kubeconfig))
+    kubeconfig_backup = f"{kubeconfig_dir}/{datetime.datetime.now().strftime(backup_date_format)}_kcleaner.bak"
 
-    update_file(kubeconfig, config_file)
+    if undo:
+        logging.info(f"Undo flag was set! checking for the backup file...")
+        logging.debug(f'Searching for backup config file {kubeconfig_backup}')
+        config_file_after = get_backup(kubeconfig_dir)
+    else:
+        config_file_before = get_file(kubeconfig)
+        logging.debug(f'Backing up config file at {kubeconfig_backup} before doing anything')
+        update_file(kubeconfig_backup, config_file_before)
+        logging.info(f'Using resource {resource}')
+        logging.debug(f'Config file to use: {kubeconfig}')
+        if name == None:
+            logging.debug(f'Name is empty, using fzf to search for the resource to remove')
+        else:
+            logging.debug(f'Name of the resource requested to remove: {name}')
+        config_file_after = remove_resource(config_file_before, resource)
+
+
+    logging.debug(f"New Config file content: \n{config_file_after}")
+    update_file(kubeconfig, config_file_after)
 
 if __name__ == '__main__':
     cli(obj={})

setup.py

@@ -2,7 +2,7 @@
 
 setup(
     name='kcleaner',
-    version='0.1.2',
+    version='0.2.0',
     author='Gui Martins',
     url='https://fancywhale.ca/',
     author_email='[email protected]',

tests/kcleaner_file_exists_test.py

@@ -0,0 +1,54 @@
+from  kcleaner import file_exists
+from testfixtures import log_capture
+import click
+from click.testing import CliRunner
+import pytest
+
+runner = CliRunner()
+
+@log_capture()
+def test_no_parameters(capture):
+
+    with pytest.raises(SystemExit) as pytest_wrapped_e:
+        file_exists(None)
+
+    assert pytest_wrapped_e.value.code == 20
+    capture.check_present(
+        ('root', 'ERROR', "Filename cannot be 'None'")
+    )
+
+@log_capture()
+def test_empty_string(capture):
+
+    with pytest.raises(SystemExit) as pytest_wrapped_e:
+        file_exists("")
+
+    assert pytest_wrapped_e.value.code == 21
+    capture.check_present(
+        ('root', 'ERROR', "Filename cannot be empty!")
+    )
+
+@log_capture()
+def test_existing_file(capture):
+    with runner.isolated_filesystem():
+        with open('./config', 'w') as f:
+            f.write('lololol')
+
+        exists = file_exists("./config")
+
+        assert exists == True
+        capture.check_present(
+            ('root', 'DEBUG', 'File exists!')
+        )
+
+@log_capture()
+def test_non_existing_file(capture):
+    with runner.isolated_filesystem():
+
+        exists = file_exists("./config")
+
+        assert exists == False
+        capture.check_present(
+            ('root', 'INFO', 'Config File Not found!')
+        )
+

tests/kcleaner_test.py

@@ -0,0 +1,49 @@
+import click
+from click.testing import CliRunner
+from  kcleaner import cli
+from testfixtures import log_capture
+
+
+runner = CliRunner()
+
+@log_capture()
+def test_clean_non_existant_file(capture):
+
+    results = runner.invoke(cli, ['-k', './non_existent_file'])
+    assert results.exit_code == 10
+    capture.check_present(
+        ('root', 'DEBUG', 'Trying to retrieve contents of file ./non_existent_file'),
+        ('root', 'DEBUG', 'checking if file ./non_existent_file exists...'),
+        ('root', 'INFO', 'Config File Not found!'),
+        ('root', 'ERROR', 'Cannot work with an empty file!, please check the path of your config file.'),
+    )
+
+@log_capture()
+def test_clean_empty_file(capture):
+    with runner.isolated_filesystem():
+        with open('./config', 'w') as f:
+            f.write('')
+
+        result = runner.invoke(cli, ['-k', './config'])
+        assert result.exit_code == 11
+        capture.check_present(
+            ('root', 'DEBUG', 'Trying to retrieve contents of file ./config'),
+            ('root', 'DEBUG', 'checking if file ./config exists...'),
+            ('root', 'DEBUG', 'File exists!'),
+            ('root', 'DEBUG', "Type of the file contents: <class 'NoneType'>"),
+            ('root', 'ERROR', "Config File is empty! Can't use it.")
+        )
+
+@log_capture()
+def test_non_valid_yaml(capture):
+    with runner.isolated_filesystem():
+        with open('./config', 'w') as f:
+            f.write('lololol')
+
+        result = runner.invoke(cli, ['-k', './config'])
+        assert result.exit_code == 12
+        capture.check_present(
+            ('root', 'DEBUG', 'checking if file ./config exists...'),
+            ('root', 'DEBUG', 'File exists!'),
+            ('root', 'ERROR', 'Config File is not a valid yaml file!'),
+        )