Total rewrite with click and modern Python (#3)

* First step of rewriting, with pytest

* Add pytest dev dep

* Working local actions

* fix lint

* Windows

.flake8

@@ -8,6 +8,7 @@ ignore =
     E2
     E3
     E4
+    W503
 max-line-length = 88
 per-file-ignores =
     __init__.py: F401

dotlink/__init__.py

@@ -1,9 +1,10 @@
-# Copyright 2022 Amethyst Reese
+# Copyright Amethyst Reese
 # Licensed under the MIT license
 
 """
 Automate deployment of dotfiles to local paths or remote hosts
 """
 
+__author__ = "Amethyst Reese"
+
 from .__version__ import __version__
-from .dotlink import Dotlink, main, VERSION

dotlink/__main__.py

@@ -1,7 +1,7 @@
-# Copyright 2022 Amethyst Reese
+# Copyright Amethyst Reese
 # Licensed under the MIT license
 
-from dotlink import main
+from dotlink.cli import main
 
 if __name__ == "__main__":
-    main
+    main()

dotlink/actions.py

@@ -0,0 +1,97 @@
+# Copyright Amethyst Reese
+# Licensed under the MIT license
+
+from __future__ import annotations
+
+import shutil
+
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Generator
+
+from .types import Target
+
+
+@dataclass
+class Plan:
+    actions: list[Action]
+
+    def __str__(self) -> str:
+        lines = ["Plan:"] + [str(action) for action in self.actions]
+        return "\n  ".join(lines)
+
+    def execute(self) -> Generator[Action, None, None]:
+        for action in self.actions:
+            action.prepare()
+
+        for action in self.actions:
+            yield action
+            action.execute()
+
+
+class Action:
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        self.args = args
+        self.kwargs = kwargs
+
+    def __str__(self) -> str:
+        return f"{self.__class__.__name__}: {self.print()}"
+
+    def print(self) -> str:
+        return f"{self.args!r}, {self.kwargs!r}"
+
+    def prepare(self) -> None:
+        pass
+
+    def execute(self) -> None:
+        raise NotImplementedError
+
+
+class Copy(Action):
+    def __init__(self, src: Path, dest: Path) -> None:
+        self.src = src
+        self.dest = dest
+
+    def print(self) -> str:
+        return f"{self.src} -> {self.dest}"
+
+    def prepare(self) -> None:
+        if not self.src.exists():
+            raise FileNotFoundError(f"{self.src} does not exist")
+
+        if not self.dest.is_symlink() and (
+            (self.dest.is_dir() and self.src.is_file())
+            or (self.dest.is_file() and self.src.is_dir())
+        ):
+            raise RuntimeError(f"file/dir type mismatch {self.src} != {self.dest}")
+
+        self.dest.parent.mkdir(parents=True, exist_ok=True)
+
+    def execute(self) -> None:
+        if self.src.is_dir():
+            if self.dest.is_symlink():
+                self.dest.unlink(missing_ok=True)
+            shutil.copytree(self.src, self.dest, dirs_exist_ok=True)
+        else:
+            self.dest.unlink(missing_ok=True)
+            shutil.copyfile(self.src, self.dest)
+
+
+class Symlink(Copy):
+    def prepare(self) -> None:
+        if not self.dest.is_symlink() and self.dest.is_dir():
+            raise RuntimeError(f"symlink destination {self.dest} is a directory")
+        super().prepare()
+
+    def execute(self) -> None:
+        self.dest.unlink(missing_ok=True)
+        self.dest.symlink_to(self.src)
+
+
+class Deploy(Action):
+    def __init__(self, src: Path, target: Target) -> None:
+        self.src = src
+        self.target = target
+
+    def print(self) -> str:
+        return f"{self.src} -> {self.target}"

dotlink/cli.py

@@ -0,0 +1,73 @@
+# Copyright Amethyst Reese
+# Licensed under the MIT license
+
+from __future__ import annotations
+
+import logging
+import platform
+import sys
+
+import click
+
+from .__version__ import __version__
+from .core import dotlink
+from .types import Method, Source, Target
+
+LOG = logging.getLogger(__name__)
+
+
+@click.command("dotlink")
+@click.version_option(__version__, "--version", "-V")
+@click.option("--debug", "-D", is_flag=True, help="enable debug output")
+@click.option(
+    "--dry-run",
+    "--plan",
+    is_flag=True,
+    help="print planned actions without executing",
+)
+@click.option(
+    "--symlink / --copy",
+    default=True,
+    help="use symlinks or copies (default symlink)",
+)
+@click.argument("source", required=False, default=".")
+@click.argument("target")
+@click.pass_context
+def main(
+    ctx: click.Context,
+    debug: bool,
+    dry_run: bool,
+    symlink: bool,
+    source: str,
+    target: str,
+) -> None:
+    """
+    Copy or symlink dotfiles from a profile repository to a new location,
+    either a local path or a remote path accessible via ssh/scp.
+
+    Source must be a local path, or git:// or https:// git repo URL.
+    Defaults to current working directory.
+
+    Target must be a local path or remote SSH/SCP destination [[user@]host:path].
+
+    See https://github.com/amyreese/dotlink for more information.
+    """
+    logging.basicConfig(
+        level=(logging.DEBUG if debug else logging.WARNING),
+        stream=sys.stderr,
+    )
+
+    if symlink and platform.system() == "Windows":
+        ctx.fail("symlinks not supported on Windows, use --copy")
+
+    plan = dotlink(
+        source=Source.parse(source),
+        target=Target.parse(target),
+        method=Method.symlink if symlink else Method.copy,
+    )
+
+    if dry_run:
+        print(plan)
+    else:
+        for action in plan.execute():
+            print(action)

dotlink/core.py

@@ -0,0 +1,131 @@
+# Copyright Amethyst Reese
+# Licensed under the MIT license
+
+from __future__ import annotations
+
+import atexit
+import logging
+from pathlib import Path
+from pprint import pformat
+from tempfile import TemporaryDirectory
+from typing import Generator
+
+from .actions import Action, Copy, Deploy, Plan, Symlink
+from .types import Config, InvalidPlan, Method, Pair, Source, Target
+from .util import run
+
+LOG = logging.getLogger(__name__)
+SUPPORTED_MAPPING_NAMES = (".dotlink", "dotlink")
+COMMENT = "#"
+INCLUDE = "@"
+SEPARATOR = "="
+
+
+def discover_config(root: Path) -> Path:
+    for filename in SUPPORTED_MAPPING_NAMES:
+        if (path := root / filename).is_file():
+            LOG.debug("discover config file %s", path)
+            return path
+    raise FileNotFoundError(f"no dotlink mapping found in {root}")
+
+
+def generate_config(root: Path) -> Config:
+    root = root.resolve()
+    path = discover_config(root)
+    content = path.read_text()
+
+    paths: dict[Path, Path] = {}
+    includes: list[Config] = []
+
+    for line in content.splitlines():
+        if line.lstrip().startswith(COMMENT):
+            continue
+
+        if line.startswith(INCLUDE):
+            subpath = root / line[1:]
+            if subpath.is_dir():
+                includes.append(generate_config(subpath))
+            elif subpath.is_file():
+                raise InvalidPlan(f"{line} is a file")
+            else:
+                raise InvalidPlan(f"{line} not found")
+
+        elif SEPARATOR in line:
+            left, _, right = line.partition(SEPARATOR)
+            paths[Path(left.strip())] = Path(right.strip())
+
+        elif line := line.strip():
+            paths[Path(line)] = Path(line)
+
+    return Config(
+        root=root,
+        paths=paths,
+        includes=includes,
+    )
+
+
+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
+
+    raise RuntimeError("unknown source value")
+
+
+def resolve_paths(config: Config, out: Path) -> Generator[Pair, None, None]:
+    out = out.resolve()
+
+    for include in config.includes:
+        yield from resolve_paths(include, out)
+
+    for left, right in config.paths.items():
+        src = config.root / right
+        dest = out / left
+        yield src, dest
+
+
+def resolve_actions(config: Config, target: Target, method: Method) -> list[Action]:
+    actions: list[Action] = []
+
+    if target.remote:
+        td = TemporaryDirectory(prefix="dotlink.")
+        atexit.register(td.cleanup)
+        staging = Path(td.name).resolve()
+        pairs = resolve_paths(config, staging)
+        method = Method.copy
+    else:
+        pairs = resolve_paths(config, target.path)
+
+    if method == Method.copy:
+        actions += (Copy(src, dest) for src, dest in pairs)
+    elif method == Method.symlink:
+        actions += (Symlink(src, dest) for src, dest in pairs)
+    else:
+        raise ValueError(f"unknown {method = !r}")
+
+    if target.remote:
+        actions += [Deploy(staging, target)]
+
+    return actions
+
+
+def dotlink(source: Source, target: Target, method: Method) -> Plan:
+    LOG.debug("source = %r", source)
+    LOG.debug("target = %r", target)
+    LOG.debug("method = %r", method)
+
+    root = prepare_source(source)
+    config = generate_config(root)
+    LOG.debug("config = %s", pformat(config, indent=2))
+
+    plan = Plan(actions=resolve_actions(config, target, method))
+    LOG.debug("plan = %s", pformat(plan, indent=2))
+
+    return plan