554 lines
17 KiB
Python
554 lines
17 KiB
Python
"""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)
|