Make tests more robust

nbgitpuller/pull.py

@@ -86,30 +86,25 @@ def branch_exists(self, branch):
         This checks to make sure the branch we are told to access
         exists in the repo
         """
-        try:
-            heads = subprocess.run(
-                ["git", "ls-remote", "--heads", "--", self.git_url],
-                capture_output=True,
-                text=True,
-                check=True
-            )
-            tags = subprocess.run(
-                ["git", "ls-remote", "--tags", "--", self.git_url],
-                capture_output=True,
-                text=True,
-                check=True
-            )
-            lines = heads.stdout.splitlines() + tags.stdout.splitlines()
-            branches = []
-            for line in lines:
-                _, ref = line.split()
-                refs, heads, branch_name = ref.split("/", 2)
-                branches.append(branch_name)
-            return branch in branches
-        except subprocess.CalledProcessError:
-            m = f"Problem accessing list of branches and/or tags: {self.git_url}"
-            logging.exception(m)
-            raise ValueError(m)
+        heads = subprocess.run(
+            ["git", "ls-remote", "--heads", "--", self.git_url],
+            capture_output=True,
+            text=True,
+            check=True
+        )
+        tags = subprocess.run(
+            ["git", "ls-remote", "--tags", "--", self.git_url],
+            capture_output=True,
+            text=True,
+            check=True
+        )
+        lines = heads.stdout.splitlines() + tags.stdout.splitlines()
+        branches = []
+        for line in lines:
+            _, ref = line.split()
+            refs, heads, branch_name = ref.split("/", 2)
+            branches.append(branch_name)
+        return branch in branches
 
     def resolve_default_branch(self):
         """

tests/repohelpers.py

@@ -0,0 +1,81 @@
+"""
+Helper classes for creating git repos
+"""
+import os
+import tempfile
+import shutil
+import subprocess as sp
+from uuid import uuid4
+
+from nbgitpuller import GitPuller
+
+
+class Repository:
+    def __init__(self, path=None):
+        if path is None:
+            path = os.path.join(tempfile.gettempdir(), str(uuid4()))
+        self.path = path
+
+    def __enter__(self):
+        os.makedirs(self.path, exist_ok=True)
+        self.git('init', '--bare')
+        return self
+
+    def __exit__(self, *args):
+        shutil.rmtree(self.path)
+
+    def write_file(self, path, content):
+        with open(os.path.join(self.path, path), 'w') as f:
+            f.write(content)
+
+    def read_file(self, path):
+        with open(os.path.join(self.path, path)) as f:
+            return f.read()
+
+    def git(self, *args):
+        return sp.check_output(
+            ['git'] + list(args),
+            cwd=self.path,
+            stderr=sp.STDOUT
+        ).decode().strip()
+
+
+class Remote(Repository):
+    pass
+
+
+class Pusher(Repository):
+    def __init__(self, remote, path=None):
+        self.remote = remote
+        super().__init__(path=path)
+
+    def __enter__(self):
+        sp.check_output(['git', 'clone', self.remote.path, self.path])
+        self.git('config', '--local', 'user.email', '[email protected]')
+        self.git('config', '--local', 'user.name', 'pusher')
+        return self
+
+    def push_file(self, path, content):
+        self.write_file(path, content)
+        self.git('add', path)
+        self.git('commit', '-am', 'Ignore the message')
+        self.git('push', 'origin', 'master')
+
+
+class Puller(Repository):
+    def __init__(self, remote, path=None, branch="master", *args, **kwargs):
+        super().__init__(path)
+        remotepath = "file://%s" % os.path.abspath(remote.path)
+        self.gp = GitPuller(remotepath, self.path, branch=branch, *args, **kwargs)
+
+    def pull_all(self):
+        for line in self.gp.pull():
+            print('{}: {}'.format(self.path, line.rstrip()))
+
+    def __enter__(self):
+        print()
+        self.pull_all()
+        self.git('config', '--local', 'user.email', '[email protected]')
+        self.git('config', '--local', 'user.name', 'puller')
+        return self
+

tests/test_api.py

@@ -6,6 +6,8 @@
 from uuid import uuid4
 import pytest
 
+from repohelpers import Pusher, Remote
+
 PORT = os.getenv('TEST_PORT', 18888)
 
 
@@ -58,17 +60,22 @@ def test_clone_default(self, tmpdir, backend_type):
         """
         jupyterdir = str(tmpdir)
         self.start_jupyter(jupyterdir, {}, backend_type)
-        params = {
-            'repo': 'https://github.com/binder-examples/jupyter-extension',
-            'branch': 'master',
-        }
-        r = request_api(params)
-        assert r.code == 200
-        s = r.read().decode()
-        print(s)
-        assert '--branch master' in s
-        assert "Cloning into '{}/{}'".format(jupyterdir, 'jupyter-extension') in s
-        assert os.path.isdir(os.path.join(jupyterdir, 'jupyter-extension', '.git'))
+
+        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 '--branch master' in s
+            assert f"Cloning into '{target_path}" in s
+            assert os.path.isdir(os.path.join(target_path, '.git'))
 
     @pytest.mark.parametrize(
         "backend_type",
@@ -84,17 +91,20 @@ def test_clone_targetpath(self, tmpdir, backend_type):
         jupyterdir = str(tmpdir)
         target = str(uuid4())
         self.start_jupyter(jupyterdir, {}, backend_type)
-        params = {
-            'repo': 'https://github.com/binder-examples/jupyter-extension',
-            'branch': 'master',
-            'targetpath': target,
-        }
-        r = request_api(params)
-        assert r.code == 200
-        s = r.read().decode()
-        print(s)
-        assert "Cloning into '{}/{}'".format(jupyterdir, target) in s
-        assert os.path.isdir(os.path.join(jupyterdir, target, '.git'))
+        with Remote() as remote, Pusher(remote) as pusher:
+            pusher.push_file('README.md', 'Testing some content')
+            params = {
+                'repo': remote.path,
+                'branch': 'master',
+                'targetpath': target,
+            }
+            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(os.path.join(target_path, '.git'))
 
     @pytest.mark.parametrize(
         "backend_type",
@@ -111,14 +121,18 @@ def test_clone_parenttargetpath(self, tmpdir, backend_type):
         parent = str(uuid4())
         target = str(uuid4())
         self.start_jupyter(jupyterdir, {'NBGITPULLER_PARENTPATH': parent}, backend_type)
-        params = {
-            'repo': 'https://github.com/binder-examples/jupyter-extension',
-            'branch': 'master',
-            'targetpath': target,
-        }
-        r = request_api(params)
-        assert r.code == 200
-        s = r.read().decode()
-        print(s)
-        assert "Cloning into '{}/{}/{}'".format(jupyterdir, parent, target) in s
-        assert os.path.isdir(os.path.join(jupyterdir, parent, target, '.git'))
+
+        with Remote() as remote, Pusher(remote) as pusher:
+            pusher.push_file('README.md', 'Testing some content')
+            params = {
+                'repo': remote.path,
+                'branch': 'master',
+                'targetpath': target,
+            }
+            r = request_api(params)
+            assert r.code == 200
+            s = r.read().decode()
+            print(s)
+            target_path = os.path.join(jupyterdir, parent, target)
+            assert f"Cloning into '{target_path}" in s
+            assert os.path.isdir(os.path.join(target_path, '.git'))