Add a Mercurial contentprovider
MyBinder could support Mercurial repositories
See
MODIFY
Pipfile
Pipfile
@@ -12,6 +12,8 @@ alabaster = "*"
Sphinx = ">=1.4,!=1.5.4"
alabaster_jupyterhub = "*"
sphinxcontrib-autoprogram = "*"
+mercurial = "*"
+hg-evolve = "*"
[packages]
repo2docker = {path=".", editable=true}

MODIFY
dev-requirements.txt
dev-requirements.txt
@@ -6,3 +6,5 @@ wheel
pytest-cov
pre-commit
requests
+mercurial
+hg-evolve

MODIFY
repo2docker/contentproviders/__init__.py
repo2docker/contentproviders/__init__.py
@@ -4,3 +4,4 @@ from .zenodo import Zenodo
from .figshare import Figshare
from .dataverse import Dataverse
from .hydroshare import Hydroshare
+from .mercurial import Mercurial

ADD
repo2docker/contentproviders/mercurial.py
+import subprocess
+
+from .base import ContentProvider, ContentProviderException
+from ..utils import execute_cmd
+
+
+hg_config = [
+ "--config",
+ "extensions.hggit=!",
+ "--config",
+ "extensions.evolve=",
+ "--config",
+ "extensions.topic=",
+]
+
+
+class Mercurial(ContentProvider):
+ """Provide contents of a remote Mercurial repository."""
+
+ def detect(self, source, ref=None, extra_args=None):
+ if "github.com/" in source or source.endswith(".git"):
+ return None
+ try:
+ subprocess.check_output(
+ ["hg", "identify", source] + hg_config, stderr=subprocess.DEVNULL,
+ )
+ except subprocess.CalledProcessError:
+ return None
+
+ return {"repo": source, "ref": ref}
+
+ def fetch(self, spec, output_dir, yield_output=False):
+ repo = spec["repo"]
+ ref = spec.get("ref", None)
+
+ # make a clone of the remote repository
+ try:
+ cmd = ["hg", "clone", repo, output_dir]
+ cmd.extend(hg_config)
+ if ref is not None:
+ # don't update so the clone will include an empty working
+ # directory, the given ref will be updated out later
+ cmd.extend(["--noupdate"])
+ for line in execute_cmd(cmd, capture=yield_output):
+ yield line
+
+ except subprocess.CalledProcessError as error:
+ msg = "Failed to clone repository from {repo}".format(repo=repo)
+ if ref is not None:
+ msg += " (ref {ref})".format(ref=ref)
+ msg += "."
+ raise ContentProviderException(msg) from error
+
+ # check out the specific ref given by the user
+ if ref is not None:
+ try:
+ for line in execute_cmd(
+ ["hg", "update", "--clean", ref] + hg_config,
+ cwd=output_dir,
+ capture=yield_output,
+ ):
+ yield line
+ except subprocess.CalledProcessError:
+ self.log.error(
+ "Failed to update to ref %s", ref, extra=dict(phase="failed")
+ )
+ raise ValueError("Failed to update to ref {}".format(ref))
+
+ cmd = ["hg", "identify"]
+ cmd.extend(hg_config)
+ sha1 = subprocess.Popen(cmd, stdout=subprocess.PIPE, cwd=output_dir)
+ self._sha1 = sha1.stdout.read().decode().strip()
+
+ @property
+ def content_id(self):
+ """A unique ID to represent the version of the content.
+ Uses the first seven characters of the git commit ID of the repository.
+ """
+ return self._sha1[:7]

ADD
tests/unit/contentproviders/test_mercurial.py
+from pathlib import Path
+import subprocess
+from tempfile import TemporaryDirectory
+
+import pytest
+
+from repo2docker.contentproviders import Mercurial
+
+
+def _add_content_to_hg(repo_dir):
+ """Add content to file 'test' in hg repository and commit."""
+ # use append mode so this can be called multiple times
+ with open(Path(repo_dir) / "test", "a") as f:
+ f.write("Hello")
+
+ subprocess.check_call(["hg", "add", "test"], cwd=repo_dir)
+ subprocess.check_call(["hg", "commit", "-m", "Test commit"], cwd=repo_dir)
+
+
+def _get_sha1(repo_dir):
+ """Get repository's current commit SHA1."""
+ sha1 = subprocess.Popen(["hg", "identify"], stdout=subprocess.PIPE, cwd=repo_dir)
+ return sha1.stdout.read().decode().strip()
+
+
+@pytest.fixture()
+def hg_repo():
+ """
+ Make a dummy git repo in which user can perform git operations
+
+ Should be used as a contextmanager, it will delete directory when done
+ """
+ with TemporaryDirectory() as gitdir:
+ subprocess.check_call(["hg", "init"], cwd=gitdir)
+ yield gitdir
+
+
+@pytest.fixture()
+def hg_repo_with_content(hg_repo):
+ """Create a hg repository with content"""
+ _add_content_to_hg(hg_repo)
+ sha1 = _get_sha1(hg_repo)
+
+ yield hg_repo, sha1
+
+
+def test_detect_mercurial(hg_repo_with_content, repo_with_content):
+ mercurial = Mercurial()
+ assert mercurial.detect("this-is-not-a-directory") is None
+ assert mercurial.detect("https://github.com/jupyterhub/repo2docker") is None
+
+ git_repo = repo_with_content[0]
+ assert mercurial.detect(git_repo) is None
+
+ hg_repo = hg_repo_with_content[0]
+ assert mercurial.detect(hg_repo) == {"repo": hg_repo, "ref": None}
+
+
+def test_clone(hg_repo_with_content):
+ """Test simple hg clone to a target dir"""
+ upstream, sha1 = hg_repo_with_content
+
+ with TemporaryDirectory() as clone_dir:
+ spec = {"repo": upstream}
+ mercurial = Mercurial()
+ for _ in mercurial.fetch(spec, clone_dir):
+ pass
+ assert (Path(clone_dir) / "test").exists()
+
+ assert mercurial.content_id == sha1[:7]
+
+
+def test_bad_ref(hg_repo_with_content):
+ """
+ Test trying to checkout a ref that doesn't exist
+ """
+ upstream, sha1 = hg_repo_with_content
+ with TemporaryDirectory() as clone_dir:
+ spec = {"repo": upstream, "ref": "does-not-exist"}
+ with pytest.raises(ValueError):
+ for _ in Mercurial().fetch(spec, clone_dir):
+ pass

Add a Mercurial contentprovider

MyBinder could support Mercurial repositories

See https://github.com/jupyterhub/binderhub/issues/1148