Add unix_path_join file operation util.

This simplifies a lot of calls to `'/'.join` in multiple operations and
provides a single util function for unix style path joins.

Author: Fizzadar

pyinfra/operations/files.py

@@ -54,6 +54,7 @@
     ensure_whole_line_match,
     get_timestamp,
     sed_replace,
+    unix_path_join,
 )
 
 
@@ -482,12 +483,11 @@ def sync(
                 # Join local as normal (unix, win)
                 full_filename,
                 # Join remote as unix like
-                '/'.join(
+                unix_path_join(*[
                     item for item in
-                    # Remove any existing / where we're going to join
-                    (dest.rstrip('/'), remote_dirpath.strip('/'), filename.lstrip('/'))
+                    (dest, remote_dirpath, filename)
                     if item
-                ),
+                ]),
             ))
 
     # Ensure the destination directory - if the destination is a link, ensure
@@ -507,7 +507,7 @@ def sync(
     # Ensure any remote dirnames
     for dirpath, dir_mode in ensure_dirnames:
         yield directory(
-            '/'.join((dest, dirpath)),
+            unix_path_join(dest, dirpath),
             user=user, group=group, mode=dir_mode,
             state=state, host=host,
         )
@@ -711,7 +711,7 @@ def put(
     remote_file = host.get_fact(File, path=dest)
 
     if not remote_file and host.get_fact(Directory, path=dest):
-        dest = '/'.join((dest.rstrip('/'), os_path.basename(src)))
+        dest = unix_path_join(dest, os_path.basename(src))
 
     if create_remote_dir:
         yield _create_remote_dir(state, host, dest, user, group)
@@ -1004,11 +1004,11 @@ def link(
             link_dirname = os_path.dirname(path)
 
             if not os_path.isabs(target):
-                target = os_path.normpath('/'.join((link_dirname, target)))
+                target = os_path.normpath(unix_path_join(link_dirname, target))
 
             if info and not os_path.isabs(info['link_target']):
                 info['link_target'] = os_path.normpath(
-                    '/'.join((link_dirname, info['link_target'])),
+                    unix_path_join(link_dirname, info['link_target']),
                 )
 
         changed = False

pyinfra/operations/git.py

@@ -12,7 +12,7 @@
 from pyinfra.facts.git import GitBranch, GitConfig, GitTrackingBranch
 
 from . import files, ssh
-from .util.files import chown
+from .util.files import chown, unix_path_join
 
 
 @operation(pipeline_facts={
@@ -49,7 +49,7 @@ def config(
         existing_config = host.get_fact(GitConfig)
 
     # Only get the config if the repo exists at this stage
-    elif host.get_fact(Directory, path='/'.join((repo, '.git'))):
+    elif host.get_fact(Directory, path=unix_path_join(repo, '.git')):
         existing_config = host.get_fact(GitConfig, repo=repo)
 
     if existing_config.get(key) != value:
@@ -117,7 +117,7 @@ def repo(
 
     # Store git commands for directory prefix
     git_commands = []
-    is_repo = host.get_fact(Directory, path='/'.join((dest, '.git')))
+    is_repo = host.get_fact(Directory, path=unix_path_join(dest, '.git'))
 
     # Cloning new repo?
     if not is_repo:
@@ -310,7 +310,10 @@ def worktree(
     if not host.get_fact(Directory, path=worktree) and present:
 
         # be sure that `repo` is a GIT repository
-        if not assume_repo_exists and not host.get_fact(Directory, path='/'.join((repo, '.git'))):
+        if (
+            not assume_repo_exists
+            and not host.get_fact(Directory, path=unix_path_join(repo, '.git'))
+        ):
             raise OperationError(
                 'The following folder is not a valid GIT repository : {0}'.format(repo),
             )

pyinfra/operations/util/files.py

@@ -5,6 +5,12 @@
 from pyinfra.api import QuoteString, StringCommand
 
 
+def unix_path_join(*parts):
+    parts = list(parts)
+    parts[0:-1] = [part.rstrip('/') for part in parts[0:-1]]
+    return '/'.join(parts)
+
+
 def ensure_mode_int(mode):
     # Already an int (/None)?
     if isinstance(mode, int) or mode is None:

tests/test_operations_utils.py

@@ -12,10 +12,10 @@
 import pytest
 
 from pyinfra.operations.util.compat import fspath
+from pyinfra.operations.util.files import unix_path_join
 
 
 class TestCompatFSPath(TestCase):
-
     def test_fspath(self):
         assert '/path/to/file' == fspath('/path/to/file')
         with pytest.raises(TypeError):
@@ -70,3 +70,17 @@ def __fspath__(cls, obj):
     )
     def test_fspath_with_pathlib_object(self):
         assert path.join('path', 'to', 'file') == fspath(Path('path/to/file'))
+
+
+class TestUnixPathJoin(TestCase):
+    def test_simple_path(self):
+        assert unix_path_join('home', 'pyinfra') == 'home/pyinfra'
+
+    def test_absolute_path(self):
+        assert unix_path_join('/', 'home', 'pyinfra') == '/home/pyinfra'
+
+    def test_multiple_slash_path(self):
+        assert unix_path_join('/', 'home/', 'pyinfra') == '/home/pyinfra'
+
+    def test_end_slash_path(self):
+        assert unix_path_join('/', 'home', 'pyinfra/') == '/home/pyinfra/'