Backward compatibility for git suffixed existing repo

Author: Christophe Triquet

nbgitpuller/handlers.py

@@ -20,6 +20,16 @@
     ),
 )
 
+def _get_repo_basename(repo_parent_dir, repo):
+    repo_basename = os.path.splitext(os.path.basename(repo))[0]
+    local_repo_dir = os.path.join(repo_parent_dir, repo_basename)
+    # For backward compatibility
+    # restore .git extension
+    # if a repo with that name exists
+    if os.path.exists(local_repo_dir + ".git"):
+        repo_basename += ".git"
+    return repo_basename
+
 class SyncHandler(JupyterHandler):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
@@ -78,8 +88,7 @@ async def get(self):
             # must be expanded.
             repo_parent_dir = os.path.join(os.path.expanduser(self.settings['server_root_dir']),
                                            os.getenv('NBGITPULLER_PARENTPATH', ''))
-            repo_basename = os.path.splitext(os.path.basename(repo))[0]
-            repo_dir = os.path.join(repo_parent_dir, self.get_argument('targetpath', repo_basename))
+            repo_dir = os.path.join(repo_parent_dir, self.get_argument('targetpath', _get_repo_basename(repo_parent_dir, repo)))
 
             # We gonna send out event streams!
             self.set_header('content-type', 'text/event-stream')
@@ -155,9 +164,10 @@ async def get(self):
                   self.get_argument('subPath', '.')
         app = self.get_argument('app', app_env)
         parent_reldir = os.getenv('NBGITPULLER_PARENTPATH', '')
-        repo_basename = os.path.splitext(os.path.basename(repo))[0]
+        repo_parent_dir = os.path.join(os.path.expanduser(self.settings['server_root_dir']),
+                                       os.getenv('NBGITPULLER_PARENTPATH', ''))
         targetpath = self.get_argument('targetpath', None) or \
-                     self.get_argument('targetPath', repo_basename)
+                     self.get_argument('targetPath', _get_repo_basename(repo_parent_dir, repo))
 
         if urlPath:
             path = urlPath

tests/test_api.py

@@ -1,4 +1,5 @@
 import os
+import tempfile
 from http.client import HTTPConnection
 import subprocess
 import time
@@ -122,6 +123,55 @@ def test_clone_default(jupyterdir, jupyter_server):
         assert os.path.isdir(os.path.join(target_path, '.git'))
 
 
+def test_clone_suffix_dropped(jupyterdir, jupyter_server):
+    """
+    Tests drop of .git suffix from source repo path.
+    """
+    target = str(uuid4())
+    with Remote(path=os.path.join(tempfile.gettempdir(), target + '.git')) as remote, Pusher(remote) as pusher:
+        pusher.push_file('README.md', 'Testing some content')
+        print(f'path: {remote.path}')
+        params = {
+            'repo': remote.path,
+            'branch': 'master',
+        }
+        r = request_api(params)
+        assert r.code == 200
+        s = r.read().decode()
+        print(s)
+        target_path = os.path.join(jupyterdir, target)
+        assert f"Cloning into '{target_path}" in s
+        assert os.path.isdir(target_path)
+        assert not os.path.isdir(target_path + '.git')
+
+def test_clone_suffix_dropped_legacy(jupyterdir, jupyter_server):
+    """
+    Tests keep using legacy .git suffixed destination directory.
+    """
+    with Remote() as remote, Pusher(remote) as pusher:
+        pusher.push_file('README.md', 'Testing some content')
+        print(f'path: {remote.path}')
+        params = {
+            'repo': remote.path,
+            'branch': 'master',
+        }
+        r = request_api(params)
+        assert r.code == 200
+        s = r.read().decode()
+        print(s)
+        target_path = os.path.join(jupyterdir, os.path.basename(remote.path))
+        assert f"Cloning into '{target_path}" in s
+        assert os.path.isdir(target_path)
+
+        # rename target and run puller again
+        # simulate a cloned repo with nbgitpuller previous version
+        os.rename(target_path, target_path + '.git')
+        r = request_api(params)
+        assert r.code == 200
+        s = r.read().decode()
+        print(s)
+        assert "Already up to date" in s
+
 def test_clone_auth(jupyterdir, jupyter_server):
     """
     Tests use of 'repo' and 'branch' parameters.