Enable including a repo by URL, cache repo clones (#19)

dotlink/core.py

@@ -10,9 +10,11 @@
 from tempfile import TemporaryDirectory
 from typing import Generator
 
+from platformdirs import user_cache_dir
+
 from .actions import Action, Copy, Plan, SSHTarball, Symlink
 from .types import Config, InvalidPlan, Method, Pair, Source, Target
-from .util import run
+from .util import run, sha1
 
 LOG = logging.getLogger(__name__)
 SUPPORTED_MAPPING_NAMES = (".dotlink", "dotlink")
@@ -42,7 +44,17 @@ def generate_config(root: Path) -> Config:
  continue
 
  if line.startswith(INCLUDE):
- subpath = root / line[1:]
+ subsource = Source.parse(line[1:], root=root)
+ if subsource.path:
+ try:
+ assert subsource.path.relative_to(root)
+ except ValueError as e:
+ raise InvalidPlan(
+ f"non-relative include paths not allowed ({line!r} given)"
+ ) from e
+
+ subpath = prepare_source(subsource)
+
  if subpath.is_dir():
  includes.append(generate_config(subpath))
  elif subpath.is_file():
@@ -64,17 +76,57 @@ def generate_config(root: Path) -> Config:
  )
 
 
+def repo_cache_dir(source: Source) -> Path:
+ assert source.url is not None
+ if source.ref:
+ key = f"{sha1(source.url)}-{source.stem}-{source.ref}"
+ else:
+ key = f"{sha1(source.url)}-{source.stem}"
+ cache_dir = Path(user_cache_dir("dotlink")) / key
+ return cache_dir
+
+
 def prepare_source(source: Source) -> Path:
  if source.path:
  return source.path.resolve()
 
  if source.url:
  # assume this is a git repo
- tmp = TemporaryDirectory(prefix="dotlink.")
- atexit.register(tmp.cleanup)
- repo = Path(tmp.name).resolve()
- run("git", "clone", "--depth=1", source.url, repo.as_posix())
- return repo
+ repo_dir = repo_cache_dir(source)
+ if not repo_dir.is_dir():
+ repo_dir.mkdir(parents=True, exist_ok=True)
+ run("git", "clone", "--depth=1", source.url, repo_dir.as_posix())
+
+ if source.ref:
+ run(
+ "git",
+ "-C",
+ repo_dir.as_posix(),
+ "fetch",
+ "--force",
+ "--update-head-ok",
+ "--depth=1",
+ "origin",
+ f"{source.ref}:{source.ref}",
+ )
+ run(
+ "git",
+ "-C",
+ repo_dir.as_posix(),
+ "checkout",
+ "--force",
+ source.ref,
+ )
+ else:
+ run(
+ "git",
+ "-C",
+ repo_dir.as_posix(),
+ "pull",
+ "--ff-only",
+ )
+
+ return repo_dir
 
  raise RuntimeError("unknown source value")
 
@@ -95,7 +147,7 @@ def resolve_actions(config: Config, target: Target, method: Method) -> list[Acti
  actions: list[Action] = []
 
  if target.remote:
- td = TemporaryDirectory(prefix="dotlink.")
+ td = TemporaryDirectory(prefix="dotlink-target-")
  atexit.register(td.cleanup)
  staging = Path(td.name).resolve()
  pairs = resolve_paths(config, staging)

dotlink/tests/core.py

@@ -8,7 +8,7 @@
 from unittest.mock import Mock, patch
 
 from dotlink import core
-from dotlink.types import Config, InvalidPlan
+from dotlink.types import Config, InvalidPlan, Source
 
 
 class CoreTest(TestCase):
@@ -109,6 +109,34 @@ def test_generate_config(self) -> None:
  with self.assertRaisesRegex(InvalidPlan, "bar not found"):
  core.generate_config(self.dir / "invalid")
 
+ @patch("dotlink.core.user_cache_dir")
+ def test_repo_cache_dir(self, ucd_mock: Mock) -> None:
+ with TemporaryDirectory() as td:
+ tdp = Path(td) / "dotlink"
+ ucd_mock.return_value = tdp.as_posix()
+
+ for source_str, expected in (
+ ("", None),
+ ("foo/bar", None),
+ ("https://github.com/amyreese/dotfiles.git", tdp / "d45a-dotfiles"),
+ (
+ "https://github.com/amyreese/dotfiles.git#feature-branch",
+ tdp / "d45a-dotfiles-feature-branch",
+ ),
+ ("https://github.com/actions/checkout", tdp / "0f8a-checkout"),
+ (
+ "https://github.com/actions/checkout#main",
+ tdp / "0f8a-checkout-main",
+ ),
+ ):
+ with self.subTest(source_str):
+ source = Source.parse(source_str)
+ if expected is None:
+ with self.assertRaises(AssertionError):
+ core.repo_cache_dir(source)
+ else:
+ self.assertEqual(expected, core.repo_cache_dir(source))
+
  @patch("dotlink.core.run")
  def test_prepare_source(self, run_mock: Mock) -> None:
  pass

dotlink/tests/types.py

@@ -12,9 +12,26 @@ def test_source(self) -> None:
  for value, expected in (
  ("", Source(path=Path(""))),
  (".", Source(path=Path("."))),
- ("/foo/bar", Source(path=Path("/foo/bar"))),
- ("git://github.com/a/b", Source(url="git://github.com/a/b")),
- ("https://github.com/a/b.git", Source(url="https://github.com/a/b.git")),
+ (
+ "/foo/bar",
+ Source(path=Path("/foo/bar"), stem="bar"),
+ ),
+ (
+ "git://github.com/a/b",
+ Source(url="git://github.com/a/b", stem="b"),
+ ),
+ (
+ "https://github.com/a/b.git",
+ Source(url="https://github.com/a/b.git", stem="b"),
+ ),
+ (
+ "git://github.com/a/b#main",
+ Source(url="git://github.com/a/b", stem="b", ref="main"),
+ ),
+ (
+ "https://github.com/a/b.git#abc123",
+ Source(url="https://github.com/a/b.git", stem="b", ref="abc123"),
+ ),
  ):
  with self.subTest(value):
  assert Source.parse(value) == expected

dotlink/tests/util.py

@@ -13,3 +13,12 @@ def test_run_hello_world(self) -> None:
  assert python
  result = util.run(python, "-c", 'print("hello world")', capture_output=True)
  assert result.stdout == "hello world\n"
+
+ def test_sha1(self) -> None:
+ for value, expected in (
+ ("", "da39"),
+ ("hello", "aaf4"),
+ ("https://github.com/amyreese/dotfiles", "01de"),
+ ):
+ with self.subTest(value):
+ self.assertEqual(expected, util.sha1(value))

dotlink/types.py

@@ -46,14 +46,23 @@ class Config:
 class Source:
  path: Path | None = None
  url: URL | None = None
+ ref: str = ""
+ stem: str = ""
 
  @classmethod
- def parse(cls, value: str) -> Self:
+ def parse(cls, value: str, root: Path | None = None) -> Self:
  url = urlparse(value)
  if url.scheme and url.netloc:
- return cls(url=URL(value))
+ return cls(
+ url=URL(url._replace(fragment="").geturl()),
+ ref=url.fragment,
+ stem=Path(url.path).stem,
+ )
+ path = Path(value)
+ if root:
+ return cls(path=root / path, stem=path.stem)
  else:
- return cls(path=Path(value))
+ return cls(path=path, stem=path.stem)
 
 
 @dataclass(frozen=True)

dotlink/util.py

@@ -3,6 +3,8 @@
 
 from __future__ import annotations
 
+import hashlib
+
 import shlex
 import subprocess
 from typing import Any
@@ -15,3 +17,8 @@ def run(*cmd: str, **kwargs: Any) -> subprocess.CompletedProcess[str]:
  kwargs.setdefault("check", True)
  proc = subprocess.run(cmd, **kwargs)
  return proc
+
+
+def sha1(value: str) -> str:
+ k = hashlib.sha1(value.encode("utf-8"))
+ return k.hexdigest()[:4]

pyproject.toml

@@ -18,6 +18,7 @@ dynamic = ["version", "description"]
 requires-python = ">=3.8"
 dependencies = [
  "click >= 8",
+ "platformdirs >= 4",
  "typing_extensions >= 4",
 ]