"""Tests for dotfiles link planning, root markers, and module sources.""" from argparse import Namespace import json import subprocess from pathlib import Path import pytest from flow.commands.dotfiles import ( LinkSpec, _collect_home_specs, _list_profiles, _load_link_specs_from_state, _load_state, _pull_requires_ack, _resolved_package_source, _run_sudo, run_undo, _save_link_specs_to_state, _sync_to_desired, _sync_modules, ) from flow.core.config import AppConfig, FlowContext from flow.core.console import ConsoleLogger from flow.core.platform import PlatformInfo def _ctx() -> FlowContext: return FlowContext( config=AppConfig(), manifest={"profiles": {"work": {"os": "linux", "configs": {"skip": []}}}}, platform=PlatformInfo(os="linux", arch="x64", platform="linux-x64"), console=ConsoleLogger(), ) def _make_flow_tree(tmp_path: Path) -> Path: flow_root = tmp_path (flow_root / "_shared" / "git").mkdir(parents=True) (flow_root / "_shared" / "git" / ".gitconfig").write_text("shared") (flow_root / "_shared" / "tmux").mkdir(parents=True) (flow_root / "_shared" / "tmux" / ".tmux.conf").write_text("tmux") (flow_root / "work" / "git").mkdir(parents=True) (flow_root / "work" / "git" / ".gitconfig").write_text("profile") (flow_root / "_shared" / "dnsmasq" / "_root" / "etc").mkdir(parents=True) (flow_root / "_shared" / "dnsmasq" / "_root" / "etc" / "hostname").write_text("devbox") return flow_root def test_list_profiles_ignores_reserved_dirs(tmp_path): flow_root = _make_flow_tree(tmp_path) profiles = _list_profiles(flow_root) assert profiles == ["work"] def test_collect_home_specs_conflict_fails(tmp_path): flow_root = _make_flow_tree(tmp_path) home = tmp_path / "home" home.mkdir() with pytest.raises(RuntimeError, match="Conflicting dotfile targets"): _collect_home_specs(_ctx(), flow_root, home, "work", set(), None) def test_collect_home_specs_maps_root_marker_to_absolute(tmp_path): flow_root = tmp_path (flow_root / "_shared" / "dnsmasq" / "_root" / "opt" / "homebrew" / "etc").mkdir(parents=True) src = flow_root / "_shared" / "dnsmasq" / "_root" / "opt" / "homebrew" / "etc" / "dnsmasq.conf" src.write_text("conf") home = tmp_path / "home" home.mkdir() specs = _collect_home_specs(_ctx(), flow_root, home, None, set(), None) assert Path("/opt/homebrew/etc/dnsmasq.conf") in specs assert specs[Path("/opt/homebrew/etc/dnsmasq.conf")].source == src def test_collect_home_specs_skip_root_marker(tmp_path): flow_root = tmp_path (flow_root / "_shared" / "dnsmasq" / "_root" / "etc").mkdir(parents=True) (flow_root / "_shared" / "dnsmasq" / "_root" / "etc" / "hostname").write_text("devbox") home = tmp_path / "home" home.mkdir() specs = _collect_home_specs(_ctx(), flow_root, home, None, {"_root"}, None) assert Path("/etc/hostname") not in specs def test_state_round_trip(tmp_path, monkeypatch): state_file = tmp_path / "linked.json" monkeypatch.setattr("flow.commands.dotfiles.LINKED_STATE", state_file) specs = { Path("/home/user/.gitconfig"): LinkSpec( source=Path("/repo/_shared/git/.gitconfig"), target=Path("/home/user/.gitconfig"), package="_shared/git", ) } _save_link_specs_to_state(specs) loaded = _load_link_specs_from_state() assert Path("/home/user/.gitconfig") in loaded assert loaded[Path("/home/user/.gitconfig")].package == "_shared/git" def test_state_old_format_rejected(tmp_path, monkeypatch): state_file = tmp_path / "linked.json" monkeypatch.setattr("flow.commands.dotfiles.LINKED_STATE", state_file) state_file.write_text( json.dumps( { "links": { "zsh": { "/home/user/.zshrc": "/repo/.zshrc", } } } ) ) with pytest.raises(RuntimeError, match="Unsupported linked state format"): _load_link_specs_from_state() def test_module_source_requires_sync(tmp_path): package_dir = tmp_path / "_shared" / "nvim" package_dir.mkdir(parents=True) (package_dir / "_module.yaml").write_text( "source: github:dummy/example\n" "ref:\n" " branch: main\n" ) with pytest.raises(RuntimeError, match="Run 'flow dotfiles sync' first"): _resolved_package_source(_ctx(), "_shared/nvim", package_dir) def test_sync_modules_populates_cache_and_resolves_source(tmp_path, monkeypatch): module_src = tmp_path / "module-src" module_src.mkdir() subprocess.run(["git", "init", "-b", "main", str(module_src)], check=True) (module_src / ".config" / "nvim").mkdir(parents=True) (module_src / ".config" / "nvim" / "init.lua").write_text("-- module") subprocess.run(["git", "-C", str(module_src), "add", "."], check=True) subprocess.run( [ "git", "-C", str(module_src), "-c", "user.name=Flow Test", "-c", "user.email=flow-test@example.com", "commit", "-m", "init module", ], check=True, ) dotfiles = tmp_path / "dotfiles" package_dir = dotfiles / "_shared" / "nvim" package_dir.mkdir(parents=True) (package_dir / "_module.yaml").write_text( f"source: {module_src}\n" "ref:\n" " branch: main\n" ) (package_dir / "notes.txt").write_text("ignore me") monkeypatch.setattr("flow.commands.dotfiles.DOTFILES_DIR", dotfiles) monkeypatch.setattr("flow.commands.dotfiles.MODULES_DIR", tmp_path / "modules") _sync_modules(_ctx(), verbose=False) resolved = _resolved_package_source(_ctx(), "_shared/nvim", package_dir) assert (resolved / ".config" / "nvim" / "init.lua").exists() def test_module_backed_link_specs_exclude_git_internals(tmp_path, monkeypatch): module_src = tmp_path / "module-src" module_src.mkdir() subprocess.run(["git", "init", "-b", "main", str(module_src)], check=True) (module_src / ".config" / "nvim").mkdir(parents=True) (module_src / ".config" / "nvim" / "init.lua").write_text("-- module") subprocess.run(["git", "-C", str(module_src), "add", "."], check=True) subprocess.run( [ "git", "-C", str(module_src), "-c", "user.name=Flow Test", "-c", "user.email=flow-test@example.com", "commit", "-m", "init module", ], check=True, ) dotfiles = tmp_path / "dotfiles" package_dir = dotfiles / "_shared" / "nvim" package_dir.mkdir(parents=True) (package_dir / "_module.yaml").write_text( f"source: {module_src}\n" "ref:\n" " branch: main\n" ) monkeypatch.setattr("flow.commands.dotfiles.DOTFILES_DIR", dotfiles) monkeypatch.setattr("flow.commands.dotfiles.MODULES_DIR", tmp_path / "modules") _sync_modules(_ctx(), verbose=False) home = tmp_path / "home" home.mkdir() specs = _collect_home_specs(_ctx(), dotfiles, home, None, set(), None) assert home / ".config" / "nvim" / "init.lua" in specs assert not any(target.relative_to(home).parts[0] == ".git" for target in specs) def test_sync_modules_resolves_relative_source_independent_of_cwd(tmp_path, monkeypatch): module_src = tmp_path / "module-src" module_src.mkdir() subprocess.run(["git", "init", "-b", "main", str(module_src)], check=True) (module_src / ".config" / "nvim").mkdir(parents=True) (module_src / ".config" / "nvim" / "init.lua").write_text("-- module") subprocess.run(["git", "-C", str(module_src), "add", "."], check=True) subprocess.run( [ "git", "-C", str(module_src), "-c", "user.name=Flow Test", "-c", "user.email=flow-test@example.com", "commit", "-m", "init module", ], check=True, ) dotfiles = tmp_path / "dotfiles" package_dir = dotfiles / "_shared" / "nvim" package_dir.mkdir(parents=True) relative_source = Path("../../../module-src") (package_dir / "_module.yaml").write_text( f"source: {relative_source}\n" "ref:\n" " branch: main\n" ) unrelated_cwd = tmp_path / "unrelated-cwd" unrelated_cwd.mkdir() monkeypatch.chdir(unrelated_cwd) monkeypatch.setattr("flow.commands.dotfiles.DOTFILES_DIR", dotfiles) monkeypatch.setattr("flow.commands.dotfiles.MODULES_DIR", tmp_path / "modules") _sync_modules(_ctx(), verbose=False) resolved = _resolved_package_source(_ctx(), "_shared/nvim", package_dir) assert (resolved / ".config" / "nvim" / "init.lua").exists() def test_pull_requires_ack_only_on_real_updates(): assert _pull_requires_ack("Already up to date.\n", "") is False assert _pull_requires_ack("Updating 123..456\n", "") is True def test_sync_to_desired_dry_run_force_is_read_only(tmp_path, monkeypatch): state_file = tmp_path / "linked.json" monkeypatch.setattr("flow.commands.dotfiles.LINKED_STATE", state_file) monkeypatch.setattr("flow.commands.dotfiles._is_in_home", lambda _path, _home: True) source = tmp_path / "source" / ".zshrc" source.parent.mkdir(parents=True) source.write_text("# new") target = tmp_path / "home" / ".zshrc" target.parent.mkdir(parents=True) target.write_text("# old") desired = { target: LinkSpec( source=source, target=target, package="_shared/zsh", ) } _sync_to_desired( _ctx(), desired, force=True, dry_run=True, copy=False, ) assert target.exists() assert not target.is_symlink() assert target.read_text() == "# old" assert not state_file.exists() def test_sync_to_desired_force_fails_before_any_writes_on_directory_conflict(tmp_path, monkeypatch): state_file = tmp_path / "linked.json" monkeypatch.setattr("flow.commands.dotfiles.LINKED_STATE", state_file) monkeypatch.setattr("flow.commands.dotfiles._is_in_home", lambda _path, _home: True) source_root = tmp_path / "source" source_root.mkdir() source_ok = source_root / "ok" source_ok.write_text("ok") source_conflict = source_root / "conflict" source_conflict.write_text("conflict") home = tmp_path / "home" home.mkdir() target_ok = home / "a-file" target_conflict = home / "z-dir" target_conflict.mkdir() desired = { target_ok: LinkSpec(source=source_ok, target=target_ok, package="_shared/test"), target_conflict: LinkSpec(source=source_conflict, target=target_conflict, package="_shared/test"), } with pytest.raises(RuntimeError, match="cannot be overwritten"): _sync_to_desired( _ctx(), desired, force=True, dry_run=False, copy=False, ) assert not target_ok.exists() assert not target_ok.is_symlink() assert not state_file.exists() def test_undo_restores_previous_file_and_link_state(tmp_path, monkeypatch): state_file = tmp_path / "linked.json" monkeypatch.setattr("flow.commands.dotfiles.LINKED_STATE", state_file) monkeypatch.setattr("flow.commands.dotfiles.LINK_BACKUP_DIR", tmp_path / "link-backups") monkeypatch.setattr("flow.commands.dotfiles._is_in_home", lambda _path, _home: True) source = tmp_path / "source" / ".zshrc" source.parent.mkdir(parents=True) source.write_text("# managed") target = tmp_path / "home" / ".zshrc" target.parent.mkdir(parents=True) target.write_text("# previous") desired = { target: LinkSpec( source=source, target=target, package="_shared/zsh", ) } _sync_to_desired( _ctx(), desired, force=True, dry_run=False, copy=False, ) assert target.is_symlink() state_after_link = _load_state() assert "last_transaction" in state_after_link tx = state_after_link["last_transaction"] assert isinstance(tx, dict) assert tx.get("targets") run_undo(_ctx(), Namespace()) assert target.exists() assert not target.is_symlink() assert target.read_text() == "# previous" state_after_undo = _load_state() assert state_after_undo.get("links") == {} assert "last_transaction" not in state_after_undo def test_sync_to_desired_persists_incomplete_transaction_on_failure(tmp_path, monkeypatch): state_file = tmp_path / "linked.json" monkeypatch.setattr("flow.commands.dotfiles.LINKED_STATE", state_file) monkeypatch.setattr("flow.commands.dotfiles.LINK_BACKUP_DIR", tmp_path / "link-backups") monkeypatch.setattr("flow.commands.dotfiles._is_in_home", lambda _path, _home: True) source = tmp_path / "source" source.mkdir() src_a = source / "a" src_b = source / "b" src_a.write_text("a") src_b.write_text("b") home = tmp_path / "home" home.mkdir() target_a = home / "a" target_b = home / "b" target_a.write_text("old-a") desired = { target_a: LinkSpec(source=src_a, target=target_a, package="_shared/test"), target_b: LinkSpec(source=src_b, target=target_b, package="_shared/test"), } call_count = {"n": 0} def _failing_apply(spec, *, copy, dry_run): # noqa: ARG001 call_count["n"] += 1 if call_count["n"] == 2: raise RuntimeError("simulated failure") spec.target.parent.mkdir(parents=True, exist_ok=True) spec.target.symlink_to(spec.source) return True monkeypatch.setattr("flow.commands.dotfiles._apply_link_spec", _failing_apply) with pytest.raises(RuntimeError, match="simulated failure"): _sync_to_desired( _ctx(), desired, force=True, dry_run=False, copy=False, ) state_after_failure = _load_state() tx = state_after_failure.get("last_transaction") assert isinstance(tx, dict) assert tx.get("incomplete") is True assert target_a.is_symlink() run_undo(_ctx(), Namespace()) assert target_a.exists() assert not target_a.is_symlink() assert target_a.read_text() == "old-a" assert not target_b.exists() assert _load_state().get("links") == {} def test_sync_to_desired_requires_force_to_remove_modified_managed_target(tmp_path, monkeypatch): state_file = tmp_path / "linked.json" monkeypatch.setattr("flow.commands.dotfiles.LINKED_STATE", state_file) monkeypatch.setattr("flow.commands.dotfiles._is_in_home", lambda _path, _home: True) source = tmp_path / "source" / ".old" source.parent.mkdir(parents=True) source.write_text("managed") target = tmp_path / "home" / ".zshrc" target.parent.mkdir(parents=True) target.write_text("user-edited") _save_link_specs_to_state( { target: LinkSpec( source=source, target=target, package="_shared/zsh", ) } ) with pytest.raises(RuntimeError, match="Use --force"): _sync_to_desired( _ctx(), {}, force=False, dry_run=False, copy=False, ) assert target.exists() assert not target.is_symlink() assert target.read_text() == "user-edited" assert target in _load_link_specs_from_state() def test_sync_to_desired_requires_force_to_replace_modified_managed_target(tmp_path, monkeypatch): state_file = tmp_path / "linked.json" monkeypatch.setattr("flow.commands.dotfiles.LINKED_STATE", state_file) monkeypatch.setattr("flow.commands.dotfiles._is_in_home", lambda _path, _home: True) old_source = tmp_path / "source" / ".old" new_source = tmp_path / "source" / ".new" old_source.parent.mkdir(parents=True) old_source.write_text("managed-old") new_source.write_text("managed-new") target = tmp_path / "home" / ".gitconfig" target.parent.mkdir(parents=True) target.write_text("manual-file") _save_link_specs_to_state( { target: LinkSpec( source=old_source, target=target, package="_shared/git", ) } ) desired = { target: LinkSpec( source=new_source, target=target, package="_shared/git", ) } with pytest.raises(RuntimeError, match="Use --force"): _sync_to_desired( _ctx(), desired, force=False, dry_run=False, copy=False, ) assert target.exists() assert not target.is_symlink() assert target.read_text() == "manual-file" assert _load_link_specs_from_state()[target].source == old_source def test_run_sudo_errors_when_binary_missing(monkeypatch): monkeypatch.setattr("flow.commands.dotfiles.shutil.which", lambda _name: None) with pytest.raises(RuntimeError, match="sudo is required"): _run_sudo(["true"], dry_run=False)