"""Tests for DotfilesService.""" import json import time from pathlib import Path import pytest import yaml from flow.core.config import AppConfig, FlowContext from flow.core.console import Console from flow.core.errors import FlowError from flow.core.platform import PlatformInfo from flow.core.runtime import SystemRuntime from flow.core import paths from flow.services.dotfiles import DotfilesService from tests.fakes import FakeRunner class _CapturingConsole(Console): """Console that records every warn() call for assertions.""" def __init__(self): super().__init__(color=False) self.warnings: list[str] = [] def warn(self, msg: str) -> None: self.warnings.append(msg) super().warn(msg) def _make_ctx(tmp_path, console=None): """Build a FlowContext for testing.""" return FlowContext( config=AppConfig(), manifest={}, platform=PlatformInfo(), console=console or Console(color=False), runtime=SystemRuntime(), ) def _setup_dotfiles(tmp_path, packages_files): """Set up a fake dotfiles directory structure. packages_files: dict of {package_name: {relative_path: content}} """ dotfiles = tmp_path / "dotfiles" shared = dotfiles / "_shared" for pkg_name, files in packages_files.items(): pkg_dir = shared / pkg_name for rel_path, content in files.items(): file_path = pkg_dir / rel_path file_path.parent.mkdir(parents=True, exist_ok=True) file_path.write_text(content) return dotfiles class TestDotfilesServiceLink: def test_link_creates_symlinks(self, tmp_path, monkeypatch): home = tmp_path / "home" home.mkdir() dotfiles = _setup_dotfiles(tmp_path, { "zsh": {".zshrc": "# zsh config"}, "git": {".config/git/config": "[user]\n name = test"}, }) monkeypatch.setattr(paths, "HOME", home) monkeypatch.setattr(paths, "DOTFILES_DIR", dotfiles) monkeypatch.setattr(paths, "MODULES_DIR", tmp_path / "modules") monkeypatch.setattr(paths, "LINKED_STATE", tmp_path / "state" / "linked.json") ctx = _make_ctx(tmp_path) svc = DotfilesService(ctx) svc.link() assert (home / ".zshrc").is_symlink() assert (home / ".config" / "git" / "config").is_symlink() def test_link_with_module(self, tmp_path, monkeypatch): home = tmp_path / "home" home.mkdir() dotfiles = tmp_path / "dotfiles" modules = tmp_path / "modules" # Set up package with _module.yaml pkg_dir = dotfiles / "_shared" / "nvim" config_dir = pkg_dir / ".config" / "nvim" config_dir.mkdir(parents=True) (config_dir / "_module.yaml").write_text(yaml.dump({ "source": "github:test/nvim-config", "ref": {"branch": "main"}, })) # Set up local file outside mount path (pkg_dir / ".local" / "bin").mkdir(parents=True) (pkg_dir / ".local" / "bin" / "nvim-wrapper").write_text("#!/bin/sh") # Set up cloned module module_dir = modules / "_shared--nvim" module_dir.mkdir(parents=True) (module_dir / "init.lua").write_text("-- init") (module_dir / "lua").mkdir() (module_dir / "lua" / "plugins.lua").write_text("-- plugins") monkeypatch.setattr(paths, "HOME", home) monkeypatch.setattr(paths, "DOTFILES_DIR", dotfiles) monkeypatch.setattr(paths, "MODULES_DIR", modules) monkeypatch.setattr(paths, "LINKED_STATE", tmp_path / "state" / "linked.json") ctx = _make_ctx(tmp_path) svc = DotfilesService(ctx) svc.link() # Module files should be linked under .config/nvim/ assert (home / ".config" / "nvim" / "init.lua").is_symlink() assert (home / ".config" / "nvim" / "lua" / "plugins.lua").is_symlink() # Local file outside mount path should be linked assert (home / ".local" / "bin" / "nvim-wrapper").is_symlink() def test_unlink_removes_symlinks(self, tmp_path, monkeypatch): home = tmp_path / "home" home.mkdir() dotfiles = _setup_dotfiles(tmp_path, { "zsh": {".zshrc": "# zsh"}, }) monkeypatch.setattr(paths, "HOME", home) monkeypatch.setattr(paths, "DOTFILES_DIR", dotfiles) monkeypatch.setattr(paths, "MODULES_DIR", tmp_path / "modules") monkeypatch.setattr(paths, "LINKED_STATE", tmp_path / "state" / "linked.json") ctx = _make_ctx(tmp_path) svc = DotfilesService(ctx) # Link first svc.link() assert (home / ".zshrc").is_symlink() # Then unlink svc.unlink() assert not (home / ".zshrc").exists() def test_link_dry_run_no_changes(self, tmp_path, monkeypatch): home = tmp_path / "home" home.mkdir() dotfiles = _setup_dotfiles(tmp_path, { "zsh": {".zshrc": "# zsh"}, }) monkeypatch.setattr(paths, "HOME", home) monkeypatch.setattr(paths, "DOTFILES_DIR", dotfiles) monkeypatch.setattr(paths, "MODULES_DIR", tmp_path / "modules") monkeypatch.setattr(paths, "LINKED_STATE", tmp_path / "state" / "linked.json") ctx = _make_ctx(tmp_path) svc = DotfilesService(ctx) svc.link(dry_run=True) # No symlinks should exist assert not (home / ".zshrc").exists() def test_status_shows_packages(self, tmp_path, monkeypatch, capsys): home = tmp_path / "home" home.mkdir() dotfiles = _setup_dotfiles(tmp_path, { "zsh": {".zshrc": "# zsh"}, }) monkeypatch.setattr(paths, "HOME", home) monkeypatch.setattr(paths, "DOTFILES_DIR", dotfiles) monkeypatch.setattr(paths, "MODULES_DIR", tmp_path / "modules") monkeypatch.setattr(paths, "LINKED_STATE", tmp_path / "state" / "linked.json") ctx = _make_ctx(tmp_path) svc = DotfilesService(ctx) # Link first to populate state svc.link() # Check status svc.status() output = capsys.readouterr().out assert "zsh" in output def test_relink_does_not_remove_unmanaged_file(self, tmp_path, monkeypatch): home = tmp_path / "home" home.mkdir() dotfiles = _setup_dotfiles(tmp_path, { "zsh": {".zshrc": "# zsh"}, }) monkeypatch.setattr(paths, "HOME", home) monkeypatch.setattr(paths, "DOTFILES_DIR", dotfiles) monkeypatch.setattr(paths, "MODULES_DIR", tmp_path / "modules") monkeypatch.setattr(paths, "LINKED_STATE", tmp_path / "state" / "linked.json") ctx = _make_ctx(tmp_path) svc = DotfilesService(ctx) svc.link() target = home / ".zshrc" target.unlink() target.write_text("user managed file") svc.link() assert target.read_text() == "user managed file" assert not target.is_symlink() def test_status_shows_module_info(self, tmp_path, monkeypatch, capsys): home = tmp_path / "home" home.mkdir() dotfiles = tmp_path / "dotfiles" modules = tmp_path / "modules" # Set up package with _module.yaml pkg_dir = dotfiles / "_shared" / "nvim" config_dir = pkg_dir / ".config" / "nvim" config_dir.mkdir(parents=True) (config_dir / "_module.yaml").write_text(yaml.dump({ "source": "github:test/nvim-config", "ref": {"branch": "main"}, })) # Set up cloned module module_dir = modules / "_shared--nvim" module_dir.mkdir(parents=True) (module_dir / "init.lua").write_text("-- init") monkeypatch.setattr(paths, "HOME", home) monkeypatch.setattr(paths, "DOTFILES_DIR", dotfiles) monkeypatch.setattr(paths, "MODULES_DIR", modules) monkeypatch.setattr(paths, "LINKED_STATE", tmp_path / "state" / "linked.json") ctx = _make_ctx(tmp_path) svc = DotfilesService(ctx) svc.link() svc.status() output = capsys.readouterr().out assert "nvim" in output assert "branch:main" in output def test_repos_list_shows_dotfiles_and_modules(self, tmp_path, monkeypatch, capsys): home = tmp_path / "home" home.mkdir() dotfiles = tmp_path / "dotfiles" modules = tmp_path / "modules" pkg_dir = dotfiles / "_shared" / "nvim" config_dir = pkg_dir / ".config" / "nvim" config_dir.mkdir(parents=True) (config_dir / "_module.yaml").write_text(yaml.dump({ "source": "github:test/nvim-config", "ref": {"branch": "main"}, })) monkeypatch.setattr(paths, "HOME", home) monkeypatch.setattr(paths, "DOTFILES_DIR", dotfiles) monkeypatch.setattr(paths, "MODULES_DIR", modules) monkeypatch.setattr(paths, "LINKED_STATE", tmp_path / "state" / "linked.json") ctx = _make_ctx(tmp_path) svc = DotfilesService(ctx) svc.repos_list() output = capsys.readouterr().out assert "dotfiles" in output assert "nvim" in output assert "module" in output def test_repos_pull_includes_profile_module_repos(self, tmp_path, monkeypatch): home = tmp_path / "home" home.mkdir() dotfiles = tmp_path / "dotfiles" profile_pkg = dotfiles / "linux-work" / "nvim" / ".config" / "nvim" profile_pkg.mkdir(parents=True) (profile_pkg / "_module.yaml").write_text(yaml.dump({ "source": "github:test/nvim-config", "ref": {"branch": "main"}, })) monkeypatch.setattr(paths, "HOME", home) monkeypatch.setattr(paths, "DOTFILES_DIR", dotfiles) monkeypatch.setattr(paths, "MODULES_DIR", tmp_path / "modules") monkeypatch.setattr(paths, "LINKED_STATE", tmp_path / "state" / "linked.json") runtime = SystemRuntime() runner = FakeRunner() runtime.runner = runner runtime.git.runner = runner ctx = FlowContext( config=AppConfig(), manifest={}, platform=PlatformInfo(), console=Console(color=False), runtime=runtime, ) DotfilesService(ctx).repos_pull() assert any("linux-work--nvim" in " ".join(call) for call in runner.calls) def test_repos_status_shows_repo_names(self, tmp_path, monkeypatch, capsys): home = tmp_path / "home" home.mkdir() dotfiles = _setup_dotfiles(tmp_path, { "zsh": {".zshrc": "# zsh"}, }) monkeypatch.setattr(paths, "HOME", home) monkeypatch.setattr(paths, "DOTFILES_DIR", dotfiles) monkeypatch.setattr(paths, "MODULES_DIR", tmp_path / "modules") monkeypatch.setattr(paths, "LINKED_STATE", tmp_path / "state" / "linked.json") # Make dotfiles dir look like a git repo for status (dotfiles / ".git").mkdir() runtime = SystemRuntime() runner = FakeRunner() runtime.runner = runner runtime.git.runner = runner ctx = FlowContext( config=AppConfig(), manifest={}, platform=PlatformInfo(), console=Console(color=False), runtime=runtime, ) DotfilesService(ctx).repos_status() output = capsys.readouterr().out assert "dotfiles" in output def test_repos_push_calls_git_push(self, tmp_path, monkeypatch): home = tmp_path / "home" home.mkdir() dotfiles = _setup_dotfiles(tmp_path, { "zsh": {".zshrc": "# zsh"}, }) monkeypatch.setattr(paths, "HOME", home) monkeypatch.setattr(paths, "DOTFILES_DIR", dotfiles) monkeypatch.setattr(paths, "MODULES_DIR", tmp_path / "modules") monkeypatch.setattr(paths, "LINKED_STATE", tmp_path / "state" / "linked.json") runtime = SystemRuntime() runner = FakeRunner() runtime.runner = runner runtime.git.runner = runner ctx = FlowContext( config=AppConfig(), manifest={}, platform=PlatformInfo(), console=Console(color=False), runtime=runtime, ) DotfilesService(ctx).repos_push() assert any("push" in " ".join(call) for call in runner.calls) def test_repos_pull_dry_run_no_calls(self, tmp_path, monkeypatch, capsys): home = tmp_path / "home" home.mkdir() dotfiles = _setup_dotfiles(tmp_path, { "zsh": {".zshrc": "# zsh"}, }) monkeypatch.setattr(paths, "HOME", home) monkeypatch.setattr(paths, "DOTFILES_DIR", dotfiles) monkeypatch.setattr(paths, "MODULES_DIR", tmp_path / "modules") monkeypatch.setattr(paths, "LINKED_STATE", tmp_path / "state" / "linked.json") runtime = SystemRuntime() runner = FakeRunner() runtime.runner = runner runtime.git.runner = runner ctx = FlowContext( config=AppConfig(), manifest={}, platform=PlatformInfo(), console=Console(color=False), runtime=runtime, ) DotfilesService(ctx).repos_pull(dry_run=True) output = capsys.readouterr().out assert "Would" in output # No git calls should be made in dry run assert not runner.calls def test_status_filter_by_package(self, tmp_path, monkeypatch, capsys): home = tmp_path / "home" home.mkdir() dotfiles = _setup_dotfiles(tmp_path, { "zsh": {".zshrc": "# zsh"}, "git": {".gitconfig": "[user]"}, }) monkeypatch.setattr(paths, "HOME", home) monkeypatch.setattr(paths, "DOTFILES_DIR", dotfiles) monkeypatch.setattr(paths, "MODULES_DIR", tmp_path / "modules") monkeypatch.setattr(paths, "LINKED_STATE", tmp_path / "state" / "linked.json") ctx = _make_ctx(tmp_path) svc = DotfilesService(ctx) svc.link() capsys.readouterr() # discard link output svc.status(package_filter=["zsh"]) output = capsys.readouterr().out assert "zsh" in output # Only zsh should appear, not git assert "_shared/git" not in output def test_link_repairs_broken_symlinks(self, tmp_path, monkeypatch): home = tmp_path / "home" home.mkdir() dotfiles = _setup_dotfiles(tmp_path, { "zsh": {".zshrc": "# zsh config"}, }) monkeypatch.setattr(paths, "HOME", home) monkeypatch.setattr(paths, "DOTFILES_DIR", dotfiles) monkeypatch.setattr(paths, "MODULES_DIR", tmp_path / "modules") monkeypatch.setattr(paths, "LINKED_STATE", tmp_path / "state" / "linked.json") ctx = _make_ctx(tmp_path) svc = DotfilesService(ctx) # Link normally svc.link() assert (home / ".zshrc").is_symlink() # Break the symlink by removing its target real_target = (home / ".zshrc").resolve() (home / ".zshrc").unlink() (home / ".zshrc").symlink_to("/nonexistent/path") # Re-link should repair the broken symlink svc.link() assert (home / ".zshrc").is_symlink() assert (home / ".zshrc").resolve() == real_target.resolve() def _setup_module_pkg(tmp_path, *, ref_key: str, ref_value: str, profile: str = "_shared"): """Build a dotfiles tree with one module package and its module cache.""" dotfiles = tmp_path / "dotfiles" modules = tmp_path / "modules" pkg_dir = dotfiles / profile / "nvim" config_dir = pkg_dir / ".config" / "nvim" config_dir.mkdir(parents=True) (config_dir / "_module.yaml").write_text(yaml.dump({ "source": "github:test/nvim-config", "ref": {ref_key: ref_value}, })) return dotfiles, modules def _make_runner_ctx(tmp_path, home, dotfiles, modules, monkeypatch, console=None): """Build a FlowContext with a FakeRunner and the path monkeypatches set.""" monkeypatch.setattr(paths, "HOME", home) monkeypatch.setattr(paths, "DOTFILES_DIR", dotfiles) monkeypatch.setattr(paths, "MODULES_DIR", modules) monkeypatch.setattr(paths, "LINKED_STATE", tmp_path / "state" / "linked.json") runtime = SystemRuntime() runner = FakeRunner() runtime.runner = runner runtime.git.runner = runner ctx = FlowContext( config=AppConfig(), manifest={}, platform=PlatformInfo(), console=console or Console(color=False), runtime=runtime, ) return ctx, runner class TestCheckoutModuleRef: """The branch == 'main' early-return is gone: every ref runs git checkout.""" def test_branch_main_still_runs_checkout(self, tmp_path, monkeypatch): home = tmp_path / "home" home.mkdir() dotfiles, modules = _setup_module_pkg(tmp_path, ref_key="branch", ref_value="main") ctx, runner = _make_runner_ctx(tmp_path, home, dotfiles, modules, monkeypatch) DotfilesService(ctx).repos_pull() # Should have cloned (cache missing) and then checked out 'main'. clone_calls = [c for c in runner.calls if "clone" in c] checkout_calls = [c for c in runner.calls if "checkout" in c] assert clone_calls, f"expected git clone, calls: {runner.calls}" assert checkout_calls, f"expected git checkout, calls: {runner.calls}" assert any("main" in c for c in checkout_calls) def test_tag_ref_uses_tags_prefix(self, tmp_path, monkeypatch): home = tmp_path / "home" home.mkdir() dotfiles, modules = _setup_module_pkg(tmp_path, ref_key="tag", ref_value="v1.0") ctx, runner = _make_runner_ctx(tmp_path, home, dotfiles, modules, monkeypatch) DotfilesService(ctx).repos_pull() checkout_calls = [c for c in runner.calls if "checkout" in c] assert checkout_calls # tags/v1.0 form -- detached checkout. assert any("tags/v1.0" in arg for c in checkout_calls for arg in c) def test_commit_ref_uses_raw_sha(self, tmp_path, monkeypatch): home = tmp_path / "home" home.mkdir() sha = "deadbeefcafe1234567890abcdef1234567890ab" dotfiles, modules = _setup_module_pkg(tmp_path, ref_key="commit", ref_value=sha) ctx, runner = _make_runner_ctx(tmp_path, home, dotfiles, modules, monkeypatch) DotfilesService(ctx).repos_pull() checkout_calls = [c for c in runner.calls if "checkout" in c] assert checkout_calls # No tags/ prefix on commit refs. assert any(sha in arg for c in checkout_calls for arg in c) assert not any(f"tags/{sha}" in arg for c in checkout_calls for arg in c) class TestStaleStateReconciliation: """_load_state warns on stale entries and only persists when invoked from a mutating path.""" def _populate_stale_state(self, tmp_path): """Write a linked.json that points at a target with no symlink.""" state_path = tmp_path / "state" / "linked.json" state_path.parent.mkdir(parents=True) state_path.write_text(json.dumps({ "version": 2, "links": { "_shared/zsh": { str(tmp_path / "home" / ".zshrc"): { "source": str(tmp_path / "dotfiles" / "_shared" / "zsh" / ".zshrc"), "from_module": False, "needs_sudo": False, } } }, })) return state_path def test_load_state_warns_on_stale(self, tmp_path, monkeypatch): home = tmp_path / "home" home.mkdir() dotfiles = _setup_dotfiles(tmp_path, {"zsh": {".zshrc": "# zsh"}}) monkeypatch.setattr(paths, "HOME", home) monkeypatch.setattr(paths, "DOTFILES_DIR", dotfiles) monkeypatch.setattr(paths, "MODULES_DIR", tmp_path / "modules") state_path = self._populate_stale_state(tmp_path) monkeypatch.setattr(paths, "LINKED_STATE", state_path) console = _CapturingConsole() ctx = _make_ctx(tmp_path, console=console) svc = DotfilesService(ctx) state = svc._load_state() assert state.links == {} # stale entry dropped assert any("stale link record" in w for w in console.warnings) def test_status_does_not_rewrite_state(self, tmp_path, monkeypatch): home = tmp_path / "home" home.mkdir() dotfiles = _setup_dotfiles(tmp_path, {"zsh": {".zshrc": "# zsh"}}) monkeypatch.setattr(paths, "HOME", home) monkeypatch.setattr(paths, "DOTFILES_DIR", dotfiles) monkeypatch.setattr(paths, "MODULES_DIR", tmp_path / "modules") state_path = self._populate_stale_state(tmp_path) monkeypatch.setattr(paths, "LINKED_STATE", state_path) before_mtime = state_path.stat().st_mtime_ns before_content = state_path.read_text() # Sleep just enough that mtime would change if we rewrote. time.sleep(0.01) console = _CapturingConsole() ctx = _make_ctx(tmp_path, console=console) DotfilesService(ctx).status() assert state_path.stat().st_mtime_ns == before_mtime assert state_path.read_text() == before_content # Still warned about the stale entry. assert any("stale link record" in w for w in console.warnings) class TestUnclonedModuleWarning: def test_link_warns_when_module_cache_missing(self, tmp_path, monkeypatch): home = tmp_path / "home" home.mkdir() dotfiles, modules = _setup_module_pkg(tmp_path, ref_key="branch", ref_value="main") # No module cache: the module dir does not exist. monkeypatch.setattr(paths, "HOME", home) monkeypatch.setattr(paths, "DOTFILES_DIR", dotfiles) monkeypatch.setattr(paths, "MODULES_DIR", modules) monkeypatch.setattr(paths, "LINKED_STATE", tmp_path / "state" / "linked.json") console = _CapturingConsole() ctx = _make_ctx(tmp_path, console=console) svc = DotfilesService(ctx) svc.link() # must not crash assert any( "not cloned" in w and "repos pull" in w for w in console.warnings ), console.warnings class TestOrphanAdoption: """After a partial-failure rerun the planner adopts pre-existing matching symlinks. We simulate a failure during _apply_plan by replacing create_symlink mid-flight, then re-running link().""" def test_partial_apply_failure_recoverable_via_rerun(self, tmp_path, monkeypatch): home = tmp_path / "home" home.mkdir() dotfiles = _setup_dotfiles(tmp_path, { "zsh": {".zshrc": "# zsh"}, "git": {".gitconfig": "[user]\n name = t"}, }) monkeypatch.setattr(paths, "HOME", home) monkeypatch.setattr(paths, "DOTFILES_DIR", dotfiles) monkeypatch.setattr(paths, "MODULES_DIR", tmp_path / "modules") state_path = tmp_path / "state" / "linked.json" monkeypatch.setattr(paths, "LINKED_STATE", state_path) ctx = _make_ctx(tmp_path) svc = DotfilesService(ctx) # Patch create_symlink to fail on the SECOND call so the first # symlink lands on disk but the state is not persisted. real_create = ctx.runtime.fs.create_symlink calls = {"n": 0} def flaky_create(source, target, **kw): calls["n"] += 1 if calls["n"] >= 2: raise FlowError("simulated mid-apply failure") return real_create(source, target, **kw) monkeypatch.setattr(ctx.runtime.fs, "create_symlink", flaky_create) with pytest.raises(FlowError, match="simulated"): svc.link() # State file should NOT have been written (atomic semantics: we # only persist when _apply_plan completes). assert not state_path.exists() # First symlink landed on disk. existing_links = sorted( p.name for p in home.rglob("*") if p.is_symlink() ) assert len(existing_links) == 1 # Restore real implementation and re-run: orphan adoption kicks in. monkeypatch.setattr(ctx.runtime.fs, "create_symlink", real_create) svc.link() assert (home / ".zshrc").is_symlink() assert (home / ".gitconfig").is_symlink() assert state_path.exists() class TestStatePersistsAtomically: def test_save_state_uses_tmp_file_atomically(self, tmp_path, monkeypatch): home = tmp_path / "home" home.mkdir() dotfiles = _setup_dotfiles(tmp_path, {"zsh": {".zshrc": "# zsh"}}) monkeypatch.setattr(paths, "HOME", home) monkeypatch.setattr(paths, "DOTFILES_DIR", dotfiles) monkeypatch.setattr(paths, "MODULES_DIR", tmp_path / "modules") state_path = tmp_path / "state" / "linked.json" monkeypatch.setattr(paths, "LINKED_STATE", state_path) ctx = _make_ctx(tmp_path) DotfilesService(ctx).link() # No leftover tmp file. residue = list(state_path.parent.glob("*.tmp")) assert residue == [] # Final content is valid JSON. json.loads(state_path.read_text())