Added option to delete repositories listed in a YAML file. (#40)

Signed-off-by: Leander Stephen D'Souza <[email protected]>

Author: leander-dsouza

README.rst

@@ -107,6 +107,29 @@ For those two types the ``version`` key is optional.
 If specified only entries from the archive which are in the subfolder specified by the version value are being extracted.
 
 
+Delete set of repositories
+~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+The ``vcs delete`` command removes all directories of repositories which are passed in via ``stdin`` in YAML format.
+By default, the command performs a dry-run and only lists the directories which would be deleted.
+In addition, it would convey warnings for missing directories and skip invalid paths upon which no action is taken.
+To actually delete the directories the ``-f/--force`` argument must be passed::
+
+.. code-block:: bash
+
+  $ vcs delete < test/list.repos
+
+  Warning: The following paths do not exist:
+    ./immutable/hash
+    ./immutable/hash_tar
+    ./immutable/hash_zip
+    ./immutable/tag
+    ./without_version
+  The following paths will be deleted:
+    ./vcs2l
+  Dry-run mode: No directories were deleted. Use -f/--force to actually delete them.
+
+
 Validate repositories file
 ~~~~~~~~~~~~~~~~~~~~~~~~~~
 

scripts/vcs-delete

@@ -0,0 +1,7 @@
+#!/usr/bin/env python3
+
+import sys
+
+from vcs2l.commands.delete import main
+
+sys.exit(main() or 0)

setup.py

@@ -66,6 +66,7 @@
             'vcs-branch = vcs2l.commands.branch:main',
             'vcs-bzr = vcs2l.commands.custom:bzr_main',
             'vcs-custom = vcs2l.commands.custom:main',
+            'vcs-delete = vcs2l.commands.delete:main',
             'vcs-diff = vcs2l.commands.diff:main',
             'vcs-export = vcs2l.commands.export:main',
             'vcs-git = vcs2l.commands.custom:git_main',

test/commands.txt

@@ -1 +1 @@
-branch custom diff export import log pull push remotes status validate
+branch custom delete diff export import log pull push remotes status validate

test/delete.txt

@@ -0,0 +1,7 @@
+The following directories will be deleted:
+  ./immutable/hash
+  ./immutable/hash_tar
+  ./immutable/hash_zip
+  ./immutable/tag
+  ./vcs2l
+  ./without_version

test/test_commands.py

@@ -328,6 +328,36 @@ def test_import_url(self):
         finally:
             rmtree(workdir)
 
+    def test_deletion(self):
+        """Test the delete command."""
+        workdir = os.path.join(TEST_WORKSPACE, 'deletion')
+        os.makedirs(workdir)
+        try:
+            run_command(
+                'import', ['--input', REPOS_FILE_URL, '.'], subfolder='deletion'
+            )
+            output = run_command(
+                'delete',
+                ['--force', '--input', REPOS_FILE_URL, '.'],
+                subfolder='deletion',
+            )
+            expected = get_expected_output('delete')
+            self.assertEqual(output, expected)
+
+            # check that repositories were actually deleted
+            self.assertFalse(os.path.exists(os.path.join(workdir, 'immutable/hash')))
+            self.assertFalse(
+                os.path.exists(os.path.join(workdir, 'immutable/hash_tar'))
+            )
+            self.assertFalse(
+                os.path.exists(os.path.join(workdir, 'immutable/hash_zip'))
+            )
+            self.assertFalse(os.path.exists(os.path.join(workdir, 'immutable/tag')))
+            self.assertFalse(os.path.exists(os.path.join(workdir, 'vcs2l')))
+            self.assertFalse(os.path.exists(os.path.join(workdir, 'without_version')))
+        finally:
+            rmtree(workdir)
+
     def test_validate(self):
         output = run_command('validate', ['--input', REPOS_FILE])
         expected = get_expected_output('validate')

vcs2l/commands/__init__.py

@@ -1,5 +1,6 @@
 from .branch import BranchCommand
 from .custom import CustomCommand
+from .delete import DeleteCommand
 from .diff import DiffCommand
 from .export import ExportCommand
 from .import_ import ImportCommand
@@ -13,6 +14,7 @@
 vcs2l_commands = []
 vcs2l_commands.append(BranchCommand)
 vcs2l_commands.append(CustomCommand)
+vcs2l_commands.append(DeleteCommand)
 vcs2l_commands.append(DiffCommand)
 vcs2l_commands.append(ExportCommand)
 vcs2l_commands.append(ImportCommand)

vcs2l/commands/delete.py

@@ -0,0 +1,151 @@
+"""Command to delete directories of repositories listed in a YAML file."""
+
+import argparse
+import os
+import sys
+import urllib.request as request
+from urllib.error import URLError
+
+from vcs2l.commands.import_ import file_or_url_type, get_repositories
+from vcs2l.executor import ansi
+from vcs2l.streams import set_streams
+from vcs2l.util import rmtree
+
+from .command import Command, existing_dir
+
+
+class DeleteCommand(Command):
+    """Delete directories of repositories listed in a YAML file."""
+
+    command = 'delete'
+    help = 'Remove the directories indicated by the list of given repositories.'
+
+
+def get_parser():
+    """CLI parser for the 'delete' command."""
+    _cls = DeleteCommand
+
+    parser = argparse.ArgumentParser(
+        description=_cls.help, prog='vcs {}'.format(_cls.command)
+    )
+    group = parser.add_argument_group('Command parameters')
+    group.add_argument(
+        '--input',
+        type=file_or_url_type,
+        default='-',
+        help='Where to read YAML from',
+        metavar='FILE_OR_URL',
+    )
+    group.add_argument(
+        'path',
+        nargs='?',
+        type=existing_dir,
+        default=os.curdir,
+        help='Base path to look for repositories',
+    )
+    group.add_argument(
+        '-f',
+        '--force',
+        action='store_true',
+        default=False,
+        help='Do the deletion instead of a dry-run',
+    )
+    return parser
+
+
+def get_repository_paths(input_source, base_path):
+    """Get repository paths from input source."""
+    try:
+        if isinstance(input_source, request.Request):
+            input_source = request.urlopen(input_source)
+        repos = get_repositories(input_source)
+        return [os.path.join(base_path, rel_path) for rel_path in repos]
+    except (RuntimeError, URLError) as e:
+        raise RuntimeError(f'Failed to read repositories: {e}') from e
+
+
+def validate_paths(paths):
+    """Validate that paths exist and are directories."""
+    valid_paths = []
+    missing_paths = []
+
+    for path in paths:
+        if os.path.exists(path) and os.path.isdir(path):
+            valid_paths.append(path)
+        else:
+            missing_paths.append(path)
+
+    return valid_paths, missing_paths
+
+
+def main(args=None, stdout=None, stderr=None):
+    """Entry point for the 'delete' command."""
+
+    set_streams(stdout=stdout, stderr=stderr)
+    parser = get_parser()
+    args = parser.parse_args(args)
+
+    try:
+        paths = get_repository_paths(args.input, args.path)
+    except RuntimeError as e:
+        print(ansi('redf') + f'Error: {e}' + ansi('reset'), file=sys.stderr)
+        return 1
+
+    if not paths:
+        print(
+            ansi('yellowf') + 'No repositories found to delete' + ansi('reset'),
+            file=sys.stderr,
+        )
+        return 0
+
+    # Validate paths existence
+    valid_paths, missing_paths = validate_paths(paths)
+
+    if not valid_paths:
+        print(
+            ansi('redf') + 'No valid directories to delete.' + ansi('reset'),
+            file=sys.stderr,
+        )
+        return 1
+    else:
+        if missing_paths:
+            print(
+                ansi('yellowf')
+                + 'Warning: The following directories do not exist:'
+                + ansi('reset'),
+                file=sys.stderr,
+            )
+            for path in missing_paths:
+                print(f'  {path}', file=sys.stderr)
+
+        print(
+            ansi('cyanf')
+            + 'The following directories will be deleted:'
+            + ansi('reset'),
+            file=sys.stderr,
+        )
+        for path in valid_paths:
+            print(f'  {path}', file=sys.stderr)
+
+    if not args.force:
+        print(
+            ansi('yellowf')
+            + 'Dry-run mode: No directories were deleted. Use -f/--force to delete them.'
+            + ansi('reset'),
+            file=sys.stderr,
+        )
+        return 0
+
+    # Actual deletion
+    for path in valid_paths:
+        try:
+            rmtree(path)
+        except OSError as e:
+            print(
+                ansi('redf') + f'Failed to delete {path}: {e}' + ansi('reset'),
+                file=sys.stderr,
+            )
+
+
+if __name__ == '__main__':
+    sys.exit(main())