ENT-14408, CFE-4700: Added option to retain git history when using cfbs convert & Replaced cp/rsync with shutil.copy

cfbs/commands.py

@@ -50,6 +50,8 @@ def search_command(terms: List[str]):
 import copy
 import logging as log
 import json
+import shutil
+import tempfile
 from typing import Callable, List, Optional, Union
 from collections import OrderedDict
 from cfbs.analyze import analyze_policyset
@@ -118,6 +120,7 @@ def search_command(terms: List[str]):
 from cfbs.index import _VERSION_INDEX, Index
 from cfbs.git import (
     git_configure_and_initialize,
+    git_get_config,
     is_git_repo,
     CFBSGitError,
     head_commit_hash,
@@ -1208,9 +1211,27 @@ def analyze_command(
 
 @cfbs_command("convert")
 def convert_command(non_interactive=False, offline=False):
+    retained_git_history = False
+    backup_dir = None
+    project_is_functional = False
+
     def cfbs_convert_cleanup():
+        # once the project is functional, later failures (e.g. a patch-file
+        # conversion step) shouldn't roll back everything that came before
+        if project_is_functional:
+            return
         rm(cfbs_filename(), missing_ok=True)
-        rm(".git", missing_ok=True)
+        if backup_dir is not None:
+            # a pre-existing `.git`-directory was moved or removed below,
+            # so restore the original directory (including its `.git`) from
+            # the backup we took before touching anything
+            print("Restoring '%s' to its original state..." % path_string)
+            rm(path_string, missing_ok=True)
+            rm(".git", missing_ok=True)
+            cp(backup_dir, path_string)
+            rm(backup_dir, missing_ok=True)
+        elif not retained_git_history:
+            rm(".git", missing_ok=True)
 
     def cfbs_convert_git_commit(
         commit_message: str, add_scope: Union[str, Iterable[str]] = "all"
@@ -1237,15 +1258,53 @@ def cfbs_convert_git_commit(
             "A `.git`-directory already exists inside %s and it will not be possible to initialize a cfbs-project"
             % path_string
         )
-        if not prompt_user_yesno(non_interactive, "Do you want to remove this?"):
+        # back up the directory before doing anything destructive to it, so
+        # a failure can restore it to its original state instead of leaving it half-converted
+        backup_dir = tempfile.mkdtemp(prefix="cfbs-convert-backup-")
+        cp(path_string, backup_dir)
+        if prompt_user_yesno(
+            non_interactive,
+            "Would you like to retain the git history of this repository? "
+            "(the existing `.git`-directory will be moved to the root of "
+            "the new CFBS project and become its git history)",
+        ):
+            print(
+                "Moving the `.git`-directory from '%s' to the root of the new project..."
+                % path_string
+            )
+            shutil.move(os.path.join(path_string, ".git"), ".git")
+            retained_git_history = True
+            # the retained repo may not have a committer identity configured
+            if (
+                git_get_config("user.name") is None
+                or git_get_config("user.email") is None
+            ):
+                try:
+                    git_configure_and_initialize(
+                        get_args().git_user_name,
+                        get_args().git_user_email,
+                        non_interactive,
+                    )
+                except:
+                    cfbs_convert_cleanup()
+                    raise
+            cfbs_convert_git_commit(
+                "Moved CFEngine policy related files to subdirectory to convert this repository into a CFEngine Build project"
+            )
+        elif prompt_user_yesno(
+            non_interactive, "Do you want to remove the `.git`-directory instead?"
+        ):
+            rm(os.path.join(path_string, ".git"))
+        else:
+            rm(backup_dir, missing_ok=True)
             raise CFBSExitError("`.git`-directory was not removed, exiting.")
-        rm(os.path.join(path_string, ".git"))
 
     # validate the local module
     validate_module_name_content(path_string)
 
     promises_cf_path = os.path.join(dir_name, "promises.cf")
     if not os.path.isfile(promises_cf_path):
+        cfbs_convert_cleanup()
         raise CFBSUserError(
             "The file '"
             + promises_cf_path
@@ -1271,11 +1330,13 @@ def cfbs_convert_git_commit(
         )
     except:
         print("Analyzing the policy set failed, aborting conversion.")
+        cfbs_convert_cleanup()
         raise
 
     current_index = CFBSConfig.get_instance().index
     masterfiles = current_index.get_module_object("masterfiles")
     if masterfiles is None:
+        cfbs_convert_cleanup()
         raise CFBSExitError("Could not find the 'masterfiles' module in the index")
     default_version = masterfiles["version"]
 
@@ -1295,12 +1356,17 @@ def cfbs_convert_git_commit(
         "Do you want to continue making a new CFEngine Build project based on masterfiles %s?"
         % masterfiles_version,
     ):
+        cfbs_convert_cleanup()
         raise CFBSExitError("User did not proceed, exiting.")
 
     print("Initializing a new CFBS project...")
-    # since there should be no other files than the masterfiles-name directory, there shouldn't be a .git directory
-    assert not is_git_repo()
     assert not is_git_repo(path_string)
+    # if git history was retained, its .git directory was already moved to
+    # the root, otherwise there shouldn't be a .git directory yet
+    if retained_git_history:
+        assert is_git_repo()
+    else:
+        assert not is_git_repo()
     try:
         r = init_command(
             masterfiles="no", non_interactive=non_interactive, use_git=True
@@ -1315,7 +1381,7 @@ def cfbs_convert_git_commit(
         print("Initializing a new CFBS project failed, aborting conversion.")
         cfbs_convert_cleanup()
         raise
-    # the cfbs-init should've created a Git repository
+    # cfbs-init should've created a Git repository, or reused the retained one
     assert is_git_repo()
     if r != 0:
         print("Initializing a new CFBS project failed, aborting conversion.")
@@ -1371,6 +1437,11 @@ def cfbs_convert_git_commit(
         "Your project is now functional, can be built, and will produce a version of masterfiles %s with your modifications."
         % masterfiles_version
     )
+    # from here on, failures shouldn't undo the (now valid) project
+    project_is_functional = True
+    if backup_dir is not None:
+        rm(backup_dir, missing_ok=True)
+        backup_dir = None
     print(
         "The next conversion step is to handle files from other versions of masterfiles."
     )
@@ -1549,6 +1620,8 @@ def cfbs_convert_git_commit(
 
     print("\n")
     remove_empty_folders(os.getcwd())
+    if backup_dir is not None:
+        rm(backup_dir, missing_ok=True)
     print("Conversion finished successfully.")
 
     return 0

cfbs/utils.py

@@ -174,14 +174,11 @@ def rm(path: str, missing_ok=False):
 
 
 def cp(src, dst):
-    above = os.path.dirname(dst)
-    if not os.path.exists(above):
-        mkdir(above)
-    if dst.endswith("/") and not os.path.exists(dst):
-        mkdir(dst)
     if os.path.isfile(src):
-        return sh("rsync -r %s %s" % (src, dst))
-    return sh("rsync -r %s/ %s" % (src, dst))
+        os.makedirs(os.path.dirname(dst), exist_ok=True)
+        shutil.copy2(src, dst)
+    else:
+        copytree_merge(src, dst)
 
 
 def cp_dry_overwrites(src: str, dst: str) -> Tuple[List[str], List[str]]:
@@ -740,3 +737,19 @@ def remove_empty_folders(target_path):
             path.rmdir()
         except OSError:
             pass
+
+
+def copytree_merge(src, dst, ignore=None):
+    """Backwards-compatible replacement for shutil.copytree(src, dst, dirs_exist_ok=True)"""
+    os.makedirs(dst, exist_ok=True)
+    names = os.listdir(src)
+    ignored_names = ignore(src, names) if ignore else set()
+    for name in names:
+        if name in ignored_names:
+            continue
+        src_path = os.path.join(src, name)
+        dst_path = os.path.join(dst, name)
+        if os.path.isdir(src_path):
+            copytree_merge(src_path, dst_path, ignore=ignore)
+        else:
+            shutil.copy2(src_path, dst_path)