Support creating hard links #301

gitman/models/source.py

@@ -14,6 +14,11 @@
 class Link:
     source: str = ""
     target: str = ""
+    symbolic: Optional[bool] = None
+
+    def __post_init__(self):
+        if self.symbolic is None:
+            self.symbolic = True
 
 
 @dataclass
@@ -201,9 +206,12 @@ def create_links(self, root: str, *, force: bool = False):
 
         for link in self.links:
             target = os.path.join(root, os.path.normpath(link.target))
-            relpath = os.path.relpath(os.getcwd(), os.path.dirname(target))
-            source = os.path.join(relpath, os.path.normpath(link.source))
-            create_sym_link(source, target, force=force)
+            if link.symbolic:
+                relpath = os.path.relpath(os.getcwd(), os.path.dirname(target))
+                source = os.path.join(relpath, os.path.normpath(link.source))
+            else:
+                source = os.path.join(os.getcwd(), os.path.normpath(link.source))
+            create_sym_link(source, target, symbolic=link.symbolic, force=force)
 
     def run_scripts(self, force: bool = False, show_shell_stdout: bool = False):
         log.info("Running install scripts...")
@@ -336,16 +344,27 @@ def _invalid_repository(self):
         return exceptions.InvalidRepository(msg)
 
 
-def create_sym_link(source: str, target: str, *, force: bool):
-    log.info("Creating a symbolic link...")
+def create_sym_link(source: str, target: str, symbolic, *, force: bool):
+
+    if symbolic:
+        log.info("Creating a symbolic link...")
+    else:
+        log.info("Creating a hard link...")
 
     if os.path.islink(target):
         os.remove(target)
     elif os.path.exists(target):
-        if force:
-            shell.rm(target)
-        else:
-            msg = "Preexisting link location at {}".format(target)
-            raise exceptions.UncommittedChanges(msg)
+        if symbolic:
+            if force:
+                shell.rm(target)
+            else:
+                msg = "Preexisting link location at {}".format(target)
+                raise exceptions.UncommittedChanges(msg)
+        elif not os.path.isdir(target):
+            if force:
+                shell.rm(target)
+            else:
+                msg = "Preexisting file location at {}".format(target)
+                raise exceptions.UncommittedChanges(msg)
 
-    shell.ln(source, target)
+    shell.ln(source, target, symbolic)

gitman/shell.py

@@ -118,11 +118,50 @@ def pwd(_show=True):
     return cwd
 
 
-def ln(source, target):
-    dirpath = os.path.dirname(target)
-    if not os.path.isdir(dirpath):
-        mkdir(dirpath)
-    os.symlink(source, target)
+def ln(source, target, symbolic):
+    if symbolic:
+        dirpath = os.path.dirname(target)
+        if not os.path.isdir(dirpath):
+            mkdir(dirpath)
+        os.symlink(source, target)
+    else:
+        if not os.path.isdir(target):
+            mkdir(target)
+        # sync files and directories to source not present in target
+        for (wd_source, dirs, files) in os.walk(source):
+            wd_target = os.path.normpath(
+                os.path.join(target, os.path.relpath(wd_source, source))
+            )
+            for dir in dirs:
+                mkdir(os.path.join(wd_target, dir))
+            for file in files:
+                file_source = os.path.join(wd_source, file)
+                file_target = os.path.join(wd_target, file)
+                if not os.path.exists(file_target):
+                    os.link(file_source, file_target)
+
+        # delete files and record directories from target not present in source
+        target_dirs = []
+        for (cwd, dirs, files) in os.walk(target):
+            wd_source = os.path.normpath(
+                os.path.join(source, os.path.relpath(cwd, target))
+            )
+
+            for file in files:
+                target_file = os.path.join(cwd, file)
+                source_file = os.path.join(wd_source, file)
+                if not os.path.isfile(source_file):
+                    rm(target_file)
+
+            for dir in dirs:
+                target_dir = os.path.join(cwd, dir)
+                source_dir = os.path.join(wd_source, dir)
+                if not os.path.isdir(source_dir):
+                    target_dirs.append(target_dir)
+
+        # delete directories from target not present in source
+        for dir in target_dirs:
+            rm(dir)
 
 
 def rm(path):

gitman/tests/test_shell.py

@@ -66,7 +66,7 @@ def test_pwd(self, mock_show, mock_call):
     @patch("os.symlink")
     def test_ln(self, mock_symlink, mock_call):
         """Verify the commands to create symbolic links."""
-        shell.ln("mock/target", "mock/source")
+        shell.ln("mock/target", "mock/source", True)
         mock_symlink.assert_called_once_with("mock/target", "mock/source")
         check_calls(mock_call, [])
 
@@ -75,7 +75,7 @@ def test_ln(self, mock_symlink, mock_call):
     @patch("os.symlink")
     def test_ln_missing_parent(self, mock_symlink, mock_call):
         """Verify the commands to create symbolic links (missing parent)."""
-        shell.ln("mock/target", "mock/source")
+        shell.ln("mock/target", "mock/source", True)
         mock_symlink.assert_called_once_with("mock/target", "mock/source")
         if os.name == "nt":
             check_calls(mock_call, ["mkdir mock"])