tm0/flow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

working version

Browse Source
This commit is contained in:
Tomas Mirchev
2026-02-13 12:15:46 +02:00
parent 1217337fbb
commit 6cff65f288
37 changed files with 2232 additions and 1872 deletions
Download Patch File Download Diff File Expand all files Collapse all files

235
tests/test_bootstrap.py
View File

@@ -1,12 +1,16 @@
"""Tests for flow.commands.bootstrap — action planning."""
"""Tests for flow.commands.bootstrap helpers and schema behavior."""
import os
import pytest
from flow.commands.bootstrap import (
_ensure_required_variables,
_get_profiles,
_plan_actions,
_normalize_profile_package_entry,
_resolve_package_manager,
_resolve_package_name,
_resolve_package_spec,
_resolve_pkg_source_name,
)
from flow.core.config import AppConfig, FlowContext
from flow.core.console import ConsoleLogger
@@ -18,127 +22,28 @@ def ctx():
return FlowContext(
config=AppConfig(),
manifest={
"binaries": {
"neovim": {
"version": "0.10.4",
"source": "github:neovim/neovim",
"asset-pattern": "nvim-{{os}}-{{arch}}.tar.gz",
"platform-map": {"linux-arm64": {"os": "linux", "arch": "arm64"}},
"install-script": "echo install",
"packages": [
{
"name": "fd",
"type": "pkg",
"sources": {"apt": "fd-find", "dnf": "fd-find", "brew": "fd"},
},
},
{
"name": "neovim",
"type": "binary",
"source": "github:neovim/neovim",
"version": "0.10.4",
"asset-pattern": "nvim-{{os}}-{{arch}}.tar.gz",
"platform-map": {"linux-x64": {"os": "linux", "arch": "x64"}},
"install": {"bin": ["bin/nvim"]},
},
]
},
platform=PlatformInfo(os="linux", arch="arm64", platform="linux-arm64"),
platform=PlatformInfo(os="linux", arch="x64", platform="linux-x64"),
console=ConsoleLogger(),
)
def test_plan_empty_profile(ctx):
actions = _plan_actions(ctx, "test", {}, {})
assert actions == []
def test_plan_hostname(ctx):
actions = _plan_actions(ctx, "test", {"hostname": "myhost"}, {})
types = [a.type for a in actions]
assert "set-hostname" in types
def test_plan_locale_and_shell(ctx):
actions = _plan_actions(ctx, "test", {"locale": "en_US.UTF-8", "shell": "zsh"}, {})
types = [a.type for a in actions]
assert "set-locale" in types
assert "set-shell" in types
def test_plan_packages(ctx):
env_config = {
"packages": {
"standard": ["git", "zsh", "tmux"],
"binary": ["neovim"],
},
}
actions = _plan_actions(ctx, "test", env_config, {})
types = [a.type for a in actions]
assert "pm-update" in types
assert "install-packages" in types
assert "install-binary" in types
def test_plan_packages_uses_package_map(ctx):
ctx.manifest["package-map"] = {
"fd": {"apt": "fd-find"},
}
env_config = {
"package-manager": "apt",
"packages": {
"standard": ["fd"],
},
}
actions = _plan_actions(ctx, "test", env_config, {})
install = [a for a in actions if a.type == "install-packages"][0]
assert install.data["packages"] == ["fd-find"]
def test_plan_ssh_keygen(ctx):
env_config = {
"ssh_keygen": [
{"type": "ed25519", "comment": "test@host", "filename": "id_ed25519"},
],
}
actions = _plan_actions(ctx, "test", env_config, {})
types = [a.type for a in actions]
assert "generate-ssh-key" in types
def test_plan_runcmd(ctx):
env_config = {"runcmd": ["echo hello", "mkdir -p ~/tmp"]}
actions = _plan_actions(ctx, "test", env_config, {})
run_cmds = [a for a in actions if a.type == "run-command"]
assert len(run_cmds) == 2
def test_plan_requires(ctx):
env_config = {"requires": ["VAR1", "VAR2"]}
actions = _plan_actions(ctx, "test", env_config, {})
checks = [a for a in actions if a.type == "check-variable"]
assert len(checks) == 2
assert all(not a.skip_on_error for a in checks)
def test_plan_full_profile(ctx):
"""Test planning with a realistic linux-vm profile."""
env_config = {
"requires": ["TARGET_HOSTNAME"],
"os": "linux",
"hostname": "$TARGET_HOSTNAME",
"shell": "zsh",
"locale": "en_US.UTF-8",
"packages": {
"standard": ["zsh", "tmux", "git"],
"binary": ["neovim"],
},
"ssh_keygen": [{"type": "ed25519", "comment": "test"}],
"configs": ["bin"],
"runcmd": ["mkdir -p ~/projects"],
}
actions = _plan_actions(ctx, "linux-vm", env_config, {"TARGET_HOSTNAME": "myvm"})
assert len(actions) >= 8
types = [a.type for a in actions]
assert "check-variable" in types
assert "set-hostname" in types
assert "set-locale" in types
assert "set-shell" in types
assert "pm-update" in types
assert "install-packages" in types
assert "install-binary" in types
assert "generate-ssh-key" in types
assert "link-config" in types
assert "run-command" in types
def test_get_profiles_from_manifest(ctx):
ctx.manifest = {"profiles": {"linux": {"os": "linux"}}}
assert "linux" in _get_profiles(ctx)
@@ -151,38 +56,88 @@ def test_get_profiles_rejects_environments(ctx):
def test_resolve_package_manager_explicit_value(ctx):
assert _resolve_package_manager(ctx, {"package-manager": "dnf"}) == "dnf"
assert _resolve_package_manager(ctx, {"os": "linux", "package-manager": "dnf"}) == "dnf"
def test_resolve_package_manager_linux_ubuntu(ctx):
os_release = "ID=ubuntu\nID_LIKE=debian"
assert _resolve_package_manager(ctx, {}, os_release_text=os_release) == "apt"
def test_resolve_package_manager_linux_auto_apt(monkeypatch, ctx):
monkeypatch.setattr("flow.commands.bootstrap.shutil.which", lambda name: "/usr/bin/apt" if name == "apt" else None)
assert _resolve_package_manager(ctx, {"os": "linux"}) == "apt"
def test_resolve_package_manager_linux_fedora(ctx):
os_release = "ID=fedora\nID_LIKE=rhel"
assert _resolve_package_manager(ctx, {}, os_release_text=os_release) == "dnf"
def test_resolve_package_manager_linux_auto_dnf(monkeypatch, ctx):
monkeypatch.setattr("flow.commands.bootstrap.shutil.which", lambda name: "/usr/bin/dnf" if name == "dnf" else None)
assert _resolve_package_manager(ctx, {"os": "linux"}) == "dnf"
def test_resolve_package_name_with_package_map(ctx):
ctx.manifest["package-map"] = {
def test_resolve_package_manager_requires_os(ctx):
with pytest.raises(RuntimeError, match="must be set"):
_resolve_package_manager(ctx, {})
def test_normalize_package_entry_string():
assert _normalize_profile_package_entry("git") == {"name": "git"}
def test_normalize_package_entry_type_prefix():
assert _normalize_profile_package_entry("cask/wezterm") == {"name": "wezterm", "type": "cask"}
def test_normalize_package_entry_object():
out = _normalize_profile_package_entry({"name": "docker", "allow_sudo": True})
assert out["name"] == "docker"
assert out["allow_sudo"] is True
def test_resolve_package_spec_uses_catalog_type(ctx):
catalog = {
"fd": {
"apt": "fd-find",
"dnf": "fd-find",
"brew": "fd",
"name": "fd",
"type": "pkg",
"sources": {"apt": "fd-find"},
}
}
assert _resolve_package_name(ctx, "fd", "apt") == "fd-find"
assert _resolve_package_name(ctx, "fd", "dnf") == "fd-find"
assert _resolve_package_name(ctx, "fd", "brew") == "fd"
resolved = _resolve_package_spec(catalog, {"name": "fd"})
assert resolved["type"] == "pkg"
assert resolved["sources"]["apt"] == "fd-find"
def test_resolve_package_name_falls_back_with_warning(ctx):
warnings = []
ctx.console.warn = warnings.append
ctx.manifest["package-map"] = {"python3-dev": {"apt": "python3-dev"}}
def test_resolve_package_spec_defaults_to_pkg(ctx):
resolved = _resolve_package_spec({}, {"name": "git"})
assert resolved["type"] == "pkg"
resolved = _resolve_package_name(ctx, "python3-dev", "dnf", warn_missing=True)
assert resolved == "python3-dev"
assert warnings
def test_resolve_package_spec_profile_override(ctx):
catalog = {
"neovim": {
"name": "neovim",
"type": "binary",
"version": "0.10.4",
}
}
resolved = _resolve_package_spec(catalog, {"name": "neovim", "post-install": "echo ok"})
assert resolved["type"] == "binary"
assert resolved["post-install"] == "echo ok"
def test_resolve_pkg_source_name_with_mapping(ctx):
spec = {"name": "fd", "sources": {"apt": "fd-find", "dnf": "fd-find", "brew": "fd"}}
assert _resolve_pkg_source_name(spec, "apt") == "fd-find"
assert _resolve_pkg_source_name(spec, "dnf") == "fd-find"
assert _resolve_pkg_source_name(spec, "brew") == "fd"
def test_resolve_pkg_source_name_fallback_to_name(ctx):
spec = {"name": "ripgrep", "sources": {"apt": "ripgrep"}}
assert _resolve_pkg_source_name(spec, "dnf") == "ripgrep"
def test_ensure_required_variables_missing_raises():
with pytest.raises(RuntimeError, match="Missing required environment variables"):
_ensure_required_variables({"requires": ["USER_EMAIL", "TARGET_HOSTNAME"]}, {"USER_EMAIL": "a@b"})
def test_ensure_required_variables_accepts_vars(monkeypatch):
env = dict(os.environ)
env["USER_EMAIL"] = "a@b"
env["TARGET_HOSTNAME"] = "devbox"
_ensure_required_variables({"requires": ["USER_EMAIL", "TARGET_HOSTNAME"]}, env)

4
tests/test_cli.py
View File

@@ -7,7 +7,9 @@ import sys
def _clean_env():
"""Return env dict without DF_* variables that trigger enter's guard."""
return {k: v for k, v in os.environ.items() if not k.startswith("DF_")}
env = {k: v for k, v in os.environ.items() if not k.startswith("DF_")}
env["FLOW_SKIP_SUDO_REFRESH"] = "1"
return env
def test_version():

76
tests/test_config.py
View File

@@ -1,37 +1,34 @@
"""Tests for flow.core.config."""
from pathlib import Path
import pytest
from flow.core.config import AppConfig, FlowContext, load_config, load_manifest
from flow.core.config import AppConfig, load_config, load_manifest
def test_load_config_missing_file(tmp_path):
def test_load_config_missing_path(tmp_path):
cfg = load_config(tmp_path / "nonexistent")
assert isinstance(cfg, AppConfig)
assert cfg.dotfiles_url == ""
assert cfg.container_registry == "registry.tomastm.com"
def test_load_config_ini(tmp_path):
config_file = tmp_path / "config"
config_file.write_text("""
[repository]
dotfiles_url=git@github.com:user/dots.git
dotfiles_branch=dev
def test_load_config_merged_yaml(tmp_path):
(tmp_path / "10-config.yaml").write_text(
"repository:\n"
" dotfiles-url: git@github.com:user/dots.git\n"
" dotfiles-branch: dev\n"
"paths:\n"
" projects-dir: ~/code\n"
"defaults:\n"
" container-registry: my.registry.com\n"
" container-tag: v1\n"
" tmux-session: main\n"
"targets:\n"
" personal: orb personal@orb\n"
" work@ec2: work.ec2.internal ~/.ssh/id_work\n"
)
[paths]
projects_dir=~/code
[defaults]
container_registry=my.registry.com
container_tag=v1
tmux_session=main
[targets]
personal=orb personal@orb
work=ec2 work.ec2.internal ~/.ssh/id_work
""")
cfg = load_config(config_file)
cfg = load_config(tmp_path)
assert cfg.dotfiles_url == "git@github.com:user/dots.git"
assert cfg.dotfiles_branch == "dev"
assert cfg.projects_dir == "~/code"
@@ -40,31 +37,28 @@ work=ec2 work.ec2.internal ~/.ssh/id_work
assert cfg.tmux_session == "main"
assert len(cfg.targets) == 2
assert cfg.targets[0].namespace == "personal"
assert cfg.targets[0].platform == "orb"
assert cfg.targets[0].ssh_host == "personal@orb"
assert cfg.targets[1].ssh_identity == "~/.ssh/id_work"
def test_load_manifest_missing_file(tmp_path):
result = load_manifest(tmp_path / "nonexistent.yaml")
def test_load_manifest_missing_path(tmp_path):
result = load_manifest(tmp_path / "nonexistent")
assert result == {}
def test_load_manifest_valid(tmp_path):
manifest = tmp_path / "manifest.yaml"
manifest.write_text("""
profiles:
linux-vm:
os: linux
hostname: test
""")
result = load_manifest(manifest)
assert "profiles" in result
def test_load_manifest_valid_directory(tmp_path):
(tmp_path / "manifest.yaml").write_text(
"profiles:\n"
" linux-vm:\n"
" os: linux\n"
" hostname: devbox\n"
)
result = load_manifest(tmp_path)
assert result["profiles"]["linux-vm"]["os"] == "linux"
def test_load_manifest_non_dict(tmp_path):
manifest = tmp_path / "manifest.yaml"
manifest.write_text("- a\n- b\n")
result = load_manifest(manifest)
assert result == {}
def test_load_manifest_non_dict_raises(tmp_path):
bad = tmp_path / "bad.yaml"
bad.write_text("- a\n- b\n")
with pytest.raises(RuntimeError, match="must contain a mapping"):
load_manifest(bad)

97
tests/test_dotfiles.py
View File

@@ -1,80 +1,75 @@
"""Tests for flow.commands.dotfiles — link/unlink/status logic."""
import json
from pathlib import Path
"""Tests for flow.commands.dotfiles discovery and path resolution."""
import pytest
from flow.commands.dotfiles import _discover_packages, _resolve_edit_target, _walk_package
from flow.core.config import AppConfig, FlowContext
from flow.core.console import ConsoleLogger
from flow.core.platform import PlatformInfo
@pytest.fixture
def dotfiles_tree(tmp_path):
"""Create a sample dotfiles directory structure."""
common = tmp_path / "common"
(common / "zsh").mkdir(parents=True)
(common / "zsh" / ".zshrc").write_text("# zshrc")
(common / "zsh" / ".zshenv").write_text("# zshenv")
(common / "tmux").mkdir(parents=True)
(common / "tmux" / ".tmux.conf").write_text("# tmux")
def _make_tree(tmp_path):
flow_root = tmp_path
shared = flow_root / "_shared"
(shared / "zsh").mkdir(parents=True)
(shared / "zsh" / ".zshrc").write_text("# zsh")
(shared / "tmux").mkdir(parents=True)
(shared / "tmux" / ".tmux.conf").write_text("# tmux")
profiles = tmp_path / "profiles" / "work"
(profiles / "git").mkdir(parents=True)
(profiles / "git" / ".gitconfig").write_text("[user]\nname = Work")
profile = flow_root / "work"
(profile / "git").mkdir(parents=True)
(profile / "git" / ".gitconfig").write_text("[user]\nname = Work")
return tmp_path
def test_discover_packages_common(dotfiles_tree):
packages = _discover_packages(dotfiles_tree)
def test_discover_packages_shared_only(tmp_path):
tree = _make_tree(tmp_path)
packages = _discover_packages(tree)
assert "zsh" in packages
assert "tmux" in packages
assert "git" not in packages # git is only in profiles
assert "git" not in packages
def test_discover_packages_with_profile(dotfiles_tree):
packages = _discover_packages(dotfiles_tree, profile="work")
def test_discover_packages_with_profile(tmp_path):
tree = _make_tree(tmp_path)
packages = _discover_packages(tree, profile="work")
assert "zsh" in packages
assert "tmux" in packages
assert "git" in packages
def test_discover_packages_profile_overrides(dotfiles_tree):
# Add zsh to work profile
work_zsh = dotfiles_tree / "profiles" / "work" / "zsh"
work_zsh.mkdir(parents=True)
(work_zsh / ".zshrc").write_text("# work zshrc")
def test_discover_packages_profile_overrides_shared(tmp_path):
tree = _make_tree(tmp_path)
profile_zsh = tree / "work" / "zsh"
profile_zsh.mkdir(parents=True)
(profile_zsh / ".zshrc").write_text("# work zsh")
packages = _discover_packages(dotfiles_tree, profile="work")
# Profile should override common
assert packages["zsh"] == work_zsh
with pytest.raises(RuntimeError, match="Conflicting dotfile targets"):
from flow.commands.dotfiles import _collect_home_specs
_collect_home_specs(tree, tmp_path / "home", "work", set(), None)
def test_walk_package(dotfiles_tree):
home = Path("/tmp/fakehome")
source = dotfiles_tree / "common" / "zsh"
pairs = list(_walk_package(source, home))
assert len(pairs) == 2
sources = {str(s.name) for s, _ in pairs}
assert ".zshrc" in sources
assert ".zshenv" in sources
targets = {str(t) for _, t in pairs}
assert str(home / ".zshrc") in targets
assert str(home / ".zshenv") in targets
def test_walk_package_returns_relative_paths(tmp_path):
tree = _make_tree(tmp_path)
source = tree / "_shared" / "zsh"
pairs = list(_walk_package(source))
assert len(pairs) == 1
src, rel = pairs[0]
assert src.name == ".zshrc"
assert str(rel) == ".zshrc"
def test_resolve_edit_target_package(dotfiles_tree):
target = _resolve_edit_target("zsh", dotfiles_dir=dotfiles_tree)
assert target == dotfiles_tree / "common" / "zsh"
def test_resolve_edit_target_package(tmp_path):
tree = _make_tree(tmp_path)
target = _resolve_edit_target("zsh", dotfiles_dir=tree)
assert target == tree / "_shared" / "zsh"
def test_resolve_edit_target_repo_path(dotfiles_tree):
target = _resolve_edit_target("common/zsh/.zshrc", dotfiles_dir=dotfiles_tree)
assert target == dotfiles_tree / "common" / "zsh" / ".zshrc"
def test_resolve_edit_target_repo_path(tmp_path):
tree = _make_tree(tmp_path)
target = _resolve_edit_target("_shared/zsh/.zshrc", dotfiles_dir=tree)
assert target == tree / "_shared" / "zsh" / ".zshrc"
def test_resolve_edit_target_missing_returns_none(dotfiles_tree):
assert _resolve_edit_target("does-not-exist", dotfiles_dir=dotfiles_tree) is None
def test_resolve_edit_target_missing_returns_none(tmp_path):
tree = _make_tree(tmp_path)
assert _resolve_edit_target("does-not-exist", dotfiles_dir=tree) is None

350
tests/test_dotfiles_folding.py
View File

@@ -1,298 +1,94 @@
"""Integration tests for dotfiles tree folding behavior."""
"""Tests for flat-layout dotfiles helpers and state format."""
import os
import json
from pathlib import Path
import pytest
from flow.commands.dotfiles import _discover_packages, _walk_package
from flow.core.config import AppConfig, FlowContext
from flow.core.console import ConsoleLogger
from flow.core.platform import PlatformInfo
from flow.core.stow import LinkTree, TreeFolder
from flow.commands.dotfiles import (
LinkSpec,
_collect_home_specs,
_collect_root_specs,
_list_profiles,
_load_link_specs_from_state,
_save_link_specs_to_state,
)
@pytest.fixture
def ctx():
"""Create a mock FlowContext."""
return FlowContext(
config=AppConfig(),
manifest={},
platform=PlatformInfo(),
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 / "work" / "nvim").mkdir(parents=True)
(flow_root / "work" / "nvim" / ".config" / "nvim").mkdir(parents=True)
(flow_root / "work" / "nvim" / ".config" / "nvim" / "init.lua").write_text("-- init")
(flow_root / "_root" / "general" / "etc").mkdir(parents=True)
(flow_root / "_root" / "general" / "etc" / "hostname").write_text("devbox")
return flow_root
@pytest.fixture
def dotfiles_with_nested(tmp_path):
"""Create dotfiles with nested directory structure for folding tests."""
common = tmp_path / "common"
# nvim package with nested config
nvim = common / "nvim" / ".config" / "nvim"
nvim.mkdir(parents=True)
(nvim / "init.lua").write_text("-- init")
(nvim / "lua").mkdir()
(nvim / "lua" / "config.lua").write_text("-- config")
(nvim / "lua" / "plugins.lua").write_text("-- plugins")
# zsh package with flat structure
zsh = common / "zsh"
zsh.mkdir(parents=True)
(zsh / ".zshrc").write_text("# zshrc")
(zsh / ".zshenv").write_text("# zshenv")
return tmp_path
def test_list_profiles_ignores_reserved_dirs(tmp_path):
flow_root = _make_flow_tree(tmp_path)
profiles = _list_profiles(flow_root)
assert profiles == ["work"]
@pytest.fixture
def home_dir(tmp_path):
"""Create a temporary home directory."""
def test_collect_home_specs_conflict_fails(tmp_path):
flow_root = _make_flow_tree(tmp_path)
home = tmp_path / "home"
home.mkdir()
return home
with pytest.raises(RuntimeError, match="Conflicting dotfile targets"):
_collect_home_specs(flow_root, home, "work", set(), None)
def test_tree_folding_single_package(dotfiles_with_nested, home_dir):
"""Test that a single package can be folded into directory symlink."""
# Discover nvim package
packages = _discover_packages(dotfiles_with_nested)
nvim_source = packages["nvim"]
# Build link tree
tree = LinkTree()
folder = TreeFolder(tree)
# Plan links for all nvim files
operations = []
for src, dst in _walk_package(nvim_source, home_dir):
ops = folder.plan_link(src, dst, "nvim")
operations.extend(ops)
# Execute operations
folder.execute_operations(operations, dry_run=False)
# Check that we created efficient symlinks
# In ideal case, we'd have one directory symlink instead of 3 file symlinks
nvim_config = home_dir / ".config" / "nvim"
# Verify links work
assert (nvim_config / "init.lua").exists()
assert (nvim_config / "lua" / "config.lua").exists()
def test_collect_root_specs_maps_to_absolute_paths(tmp_path):
flow_root = _make_flow_tree(tmp_path)
specs = _collect_root_specs(flow_root, set(), include_root=True)
assert Path("/etc/hostname") in specs
assert specs[Path("/etc/hostname")].package == "_root/general"
def test_tree_unfolding_conflict(dotfiles_with_nested, home_dir):
"""Test that tree unfolds when second package needs same directory."""
common = dotfiles_with_nested / "common"
def test_state_round_trip(tmp_path, monkeypatch):
state_file = tmp_path / "linked.json"
monkeypatch.setattr("flow.commands.dotfiles.LINKED_STATE", state_file)
# Create second package that shares .config
tmux = common / "tmux" / ".config" / "tmux"
tmux.mkdir(parents=True)
(tmux / "tmux.conf").write_text("# tmux")
# First, link nvim (can fold .config/nvim)
tree = LinkTree()
folder = TreeFolder(tree)
nvim_source = common / "nvim"
for src, dst in _walk_package(nvim_source, home_dir):
ops = folder.plan_link(src, dst, "nvim")
folder.execute_operations(ops, dry_run=False)
# Now link tmux (should unfold if needed)
tmux_source = common / "tmux"
for src, dst in _walk_package(tmux_source, home_dir):
ops = folder.plan_link(src, dst, "tmux")
folder.execute_operations(ops, dry_run=False)
# Both packages should be linked
assert (home_dir / ".config" / "nvim" / "init.lua").exists()
assert (home_dir / ".config" / "tmux" / "tmux.conf").exists()
def test_state_format_with_directory_links(dotfiles_with_nested, home_dir):
"""Test that state file correctly tracks directory vs file links."""
tree = LinkTree()
# Add a directory link
tree.add_link(
home_dir / ".config" / "nvim",
dotfiles_with_nested / "common" / "nvim" / ".config" / "nvim",
"nvim",
is_dir_link=True,
)
# Add a file link
tree.add_link(
home_dir / ".zshrc",
dotfiles_with_nested / "common" / "zsh" / ".zshrc",
"zsh",
is_dir_link=False,
)
# Convert to state
state = tree.to_state()
# Verify format
assert state["version"] == 2
nvim_link = state["links"]["nvim"][str(home_dir / ".config" / "nvim")]
assert nvim_link["is_directory_link"] is True
zsh_link = state["links"]["zsh"][str(home_dir / ".zshrc")]
assert zsh_link["is_directory_link"] is False
def test_state_backward_compatibility_rejected(home_dir):
"""Old state format should be rejected (no backward compatibility)."""
old_state = {
"links": {
"zsh": {
str(home_dir / ".zshrc"): str(home_dir.parent / "dotfiles" / "zsh" / ".zshrc"),
}
}
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"):
LinkTree.from_state(old_state)
def test_discover_packages_with_flow_package(tmp_path):
"""Test discovering the flow package itself from dotfiles."""
common = tmp_path / "common"
# Create flow package
flow_pkg = common / "flow" / ".config" / "flow"
flow_pkg.mkdir(parents=True)
(flow_pkg / "manifest.yaml").write_text("profiles: {}")
(flow_pkg / "config").write_text("[repository]\n")
packages = _discover_packages(tmp_path)
# Flow package should be discovered like any other
assert "flow" in packages
assert packages["flow"] == common / "flow"
def test_walk_flow_package(tmp_path):
"""Test walking the flow package structure."""
flow_pkg = tmp_path / "flow"
flow_config = flow_pkg / ".config" / "flow"
flow_config.mkdir(parents=True)
(flow_config / "manifest.yaml").write_text("profiles: {}")
(flow_config / "config").write_text("[repository]\n")
home = Path("/tmp/fakehome")
pairs = list(_walk_package(flow_pkg, home))
# Should find both files
assert len(pairs) == 2
targets = [str(t) for _, t in pairs]
assert str(home / ".config" / "flow" / "manifest.yaml") in targets
assert str(home / ".config" / "flow" / "config") in targets
def test_conflict_detection_before_execution(dotfiles_with_nested, home_dir):
"""Test that conflicts are detected before any changes are made."""
# Create existing file that conflicts
existing = home_dir / ".zshrc"
existing.parent.mkdir(parents=True, exist_ok=True)
existing.write_text("# existing zshrc")
# Try to link package that wants .zshrc
tree = LinkTree()
folder = TreeFolder(tree)
zsh_source = dotfiles_with_nested / "common" / "zsh"
operations = []
for src, dst in _walk_package(zsh_source, home_dir):
ops = folder.plan_link(src, dst, "zsh")
operations.extend(ops)
# Should detect conflict
conflicts = folder.detect_conflicts(operations)
assert len(conflicts) > 0
assert any("already exists" in c for c in conflicts)
# Original file should be unchanged
assert existing.read_text() == "# existing zshrc"
def test_profile_switching_relink(tmp_path):
"""Test switching between profiles maintains correct links."""
# Create profiles
common = tmp_path / "common"
profiles = tmp_path / "profiles"
# Common zsh
(common / "zsh").mkdir(parents=True)
(common / "zsh" / ".zshrc").write_text("# common zsh")
# Work profile override
(profiles / "work" / "zsh").mkdir(parents=True)
(profiles / "work" / "zsh" / ".zshrc").write_text("# work zsh")
# Personal profile override
(profiles / "personal" / "zsh").mkdir(parents=True)
(profiles / "personal" / "zsh" / ".zshrc").write_text("# personal zsh")
# Test that profile discovery works correctly
work_packages = _discover_packages(tmp_path, profile="work")
personal_packages = _discover_packages(tmp_path, profile="personal")
# Both should find zsh, but from different sources
assert "zsh" in work_packages
assert "zsh" in personal_packages
assert work_packages["zsh"] != personal_packages["zsh"]
def test_can_fold_empty_directory():
"""Test can_fold with empty directory."""
tree = LinkTree()
target_dir = Path("/home/user/.config/nvim")
# Empty directory - should be able to fold
assert tree.can_fold(target_dir, "nvim")
def test_can_fold_with_subdirectories():
"""Test can_fold with nested directory structure."""
tree = LinkTree()
base = Path("/home/user/.config/nvim")
# Add nested files from same package
tree.add_link(base / "init.lua", Path("/dotfiles/nvim/init.lua"), "nvim")
tree.add_link(base / "lua" / "config.lua", Path("/dotfiles/nvim/lua/config.lua"), "nvim")
tree.add_link(base / "lua" / "plugins" / "init.lua", Path("/dotfiles/nvim/lua/plugins/init.lua"), "nvim")
# Should be able to fold at base level
assert tree.can_fold(base, "nvim")
# Add file from different package
tree.add_link(base / "other.lua", Path("/dotfiles/other/other.lua"), "other")
# Now cannot fold
assert not tree.can_fold(base, "nvim")
def test_execute_operations_creates_parent_dirs(tmp_path):
"""Test that execute_operations creates necessary parent directories."""
tree = LinkTree()
folder = TreeFolder(tree)
source = tmp_path / "dotfiles" / "nvim" / ".config" / "nvim" / "init.lua"
target = tmp_path / "home" / ".config" / "nvim" / "init.lua"
# Create source
source.parent.mkdir(parents=True)
source.write_text("-- init")
# Target parent doesn't exist yet
assert not target.parent.exists()
# Plan and execute
ops = folder.plan_link(source, target, "nvim")
folder.execute_operations(ops, dry_run=False)
# Parent should be created
assert target.parent.exists()
assert target.is_symlink()
_load_link_specs_from_state()

8
tests/test_paths.py
View File

@@ -18,15 +18,15 @@ from flow.core.paths import (
def test_config_dir_under_home():
assert ".config/devflow" in str(CONFIG_DIR)
assert ".config/flow" in str(CONFIG_DIR)
def test_data_dir_under_home():
assert ".local/share/devflow" in str(DATA_DIR)
assert ".local/share/flow" in str(DATA_DIR)
def test_state_dir_under_home():
assert ".local/state/devflow" in str(STATE_DIR)
assert ".local/state/flow" in str(STATE_DIR)
def test_manifest_file_in_config_dir():
@@ -34,7 +34,7 @@ def test_manifest_file_in_config_dir():
def test_config_file_in_config_dir():
assert CONFIG_FILE == CONFIG_DIR / "config"
assert CONFIG_FILE == CONFIG_DIR / "config.yaml"
def test_dotfiles_dir():

3
tests/test_platform.py
View File

@@ -11,7 +11,7 @@ def test_detect_platform_returns_platforminfo():
info = detect_platform()
assert isinstance(info, PlatformInfo)
assert info.os in ("linux", "macos")
assert info.arch in ("amd64", "arm64")
assert info.arch in ("x64", "arm64")
assert info.platform == f"{info.os}-{info.arch}"
@@ -27,4 +27,3 @@ def test_detect_platform_unsupported_arch(monkeypatch):
detect_platform()

220
tests/test_self_hosting.py
View File

@@ -1,215 +1,81 @@
"""Tests for self-hosting flow config from dotfiles repository."""
"""Tests for self-hosted merged YAML config loading."""
from pathlib import Path
from unittest.mock import patch
import pytest
import yaml
from flow.core import paths as paths_module
from flow.core.config import load_config, load_manifest
@pytest.fixture
def mock_paths(tmp_path, monkeypatch):
"""Mock path constants for testing."""
config_dir = tmp_path / "config"
dotfiles_dir = tmp_path / "dotfiles"
def mock_roots(tmp_path, monkeypatch):
local_root = tmp_path / "local-flow"
dotfiles_root = tmp_path / "dotfiles" / "_shared" / "flow" / ".config" / "flow"
config_dir.mkdir()
dotfiles_dir.mkdir()
local_root.mkdir(parents=True)
dotfiles_root.mkdir(parents=True)
test_paths = {
"config_dir": config_dir,
"dotfiles_dir": dotfiles_dir,
"local_config": config_dir / "config",
"local_manifest": config_dir / "manifest.yaml",
"dotfiles_config": dotfiles_dir / "flow" / ".config" / "flow" / "config",
"dotfiles_manifest": dotfiles_dir / "flow" / ".config" / "flow" / "manifest.yaml",
monkeypatch.setattr(paths_module, "CONFIG_DIR", local_root)
monkeypatch.setattr(paths_module, "DOTFILES_FLOW_CONFIG", dotfiles_root)
return {
"local": local_root,
"dotfiles": dotfiles_root,
}
# Patch at the paths module level
monkeypatch.setattr(paths_module, "CONFIG_FILE", test_paths["local_config"])
monkeypatch.setattr(paths_module, "MANIFEST_FILE", test_paths["local_manifest"])
monkeypatch.setattr(paths_module, "DOTFILES_CONFIG", test_paths["dotfiles_config"])
monkeypatch.setattr(paths_module, "DOTFILES_MANIFEST", test_paths["dotfiles_manifest"])
return test_paths
def test_load_manifest_priority_dotfiles_first(mock_roots):
(mock_roots["local"] / "profiles.yaml").write_text("profiles:\n local: {os: linux}\n")
(mock_roots["dotfiles"] / "profiles.yaml").write_text("profiles:\n dotfiles: {os: macos}\n")
def test_load_manifest_priority_dotfiles_first(mock_paths):
"""Test that dotfiles manifest takes priority over local."""
# Create both manifests
local_manifest = mock_paths["local_manifest"]
dotfiles_manifest = mock_paths["dotfiles_manifest"]
local_manifest.write_text("profiles:\n local:\n os: linux")
dotfiles_manifest.parent.mkdir(parents=True)
dotfiles_manifest.write_text("profiles:\n dotfiles:\n os: macos")
# Should load from dotfiles
manifest = load_manifest()
assert "dotfiles" in manifest.get("profiles", {})
assert "local" not in manifest.get("profiles", {})
def test_load_manifest_fallback_to_local(mock_paths):
"""Test fallback to local manifest when dotfiles doesn't exist."""
local_manifest = mock_paths["local_manifest"]
local_manifest.write_text("profiles:\n local:\n os: linux")
def test_load_manifest_fallback_to_local(mock_roots):
(mock_roots["local"] / "profiles.yaml").write_text("profiles:\n local: {os: linux}\n")
# Remove dotfiles yaml file so local takes over.
dot_yaml = mock_roots["dotfiles"] / "profiles.yaml"
if dot_yaml.exists():
dot_yaml.unlink()
# Dotfiles manifest doesn't exist
manifest = load_manifest()
assert "local" in manifest.get("profiles", {})
def test_load_manifest_empty_when_none_exist(mock_paths):
"""Test empty dict returned when no manifests exist."""
def test_load_manifest_empty_when_none_exist(mock_roots):
manifest = load_manifest()
assert manifest == {}
def test_load_config_priority_dotfiles_first(mock_paths):
"""Test that dotfiles config takes priority over local."""
local_config = mock_paths["local_config"]
dotfiles_config = mock_paths["dotfiles_config"]
# Create local config
local_config.write_text(
"[repository]\n"
"dotfiles_url = https://github.com/user/dotfiles-local.git\n"
def test_load_config_from_merged_yaml(mock_roots):
(mock_roots["dotfiles"] / "config.yaml").write_text(
"repository:\n"
" dotfiles-url: git@github.com:user/dotfiles.git\n"
"defaults:\n"
" container-registry: registry.example.com\n"
)
# Create dotfiles config
dotfiles_config.parent.mkdir(parents=True)
dotfiles_config.write_text(
"[repository]\n"
"dotfiles_url = https://github.com/user/dotfiles-from-repo.git\n"
)
# Should load from dotfiles
config = load_config()
assert "dotfiles-from-repo" in config.dotfiles_url
cfg = load_config()
assert cfg.dotfiles_url == "git@github.com:user/dotfiles.git"
assert cfg.container_registry == "registry.example.com"
def test_load_config_fallback_to_local(mock_paths):
"""Test fallback to local config when dotfiles doesn't exist."""
local_config = mock_paths["local_config"]
local_config.write_text(
"[repository]\n"
"dotfiles_url = https://github.com/user/dotfiles-local.git\n"
)
def test_yaml_merge_is_alphabetical_last_writer_wins(mock_roots):
(mock_roots["local"] / "10-a.yaml").write_text("profiles:\n a: {os: linux}\n")
(mock_roots["local"] / "20-b.yaml").write_text("profiles:\n b: {os: linux}\n")
# Dotfiles config doesn't exist
config = load_config()
assert "dotfiles-local" in config.dotfiles_url
manifest = load_manifest(mock_roots["local"])
assert "b" in manifest.get("profiles", {})
assert "a" not in manifest.get("profiles", {})
def test_load_config_empty_when_none_exist(mock_paths):
"""Test default config returned when no configs exist."""
config = load_config()
assert config.dotfiles_url == ""
assert config.dotfiles_branch == "main"
def test_explicit_file_path_loads_single_yaml(tmp_path):
one_file = tmp_path / "single.yaml"
one_file.write_text("profiles:\n only: {os: linux}\n")
def test_self_hosting_workflow(tmp_path, monkeypatch):
"""Test complete self-hosting workflow.
Simulates:
1. User has dotfiles repo with flow config
2. Flow links its own config from dotfiles
3. Flow reads from self-hosted location
"""
# Setup paths
home = tmp_path / "home"
dotfiles = tmp_path / "dotfiles"
home.mkdir()
dotfiles.mkdir()
# Create flow package in dotfiles
flow_pkg = dotfiles / "flow" / ".config" / "flow"
flow_pkg.mkdir(parents=True)
# Create manifest in dotfiles
manifest_content = {
"profiles": {
"test-env": {
"os": "linux",
"packages": {"standard": ["git", "vim"]},
}
}
}
(flow_pkg / "manifest.yaml").write_text(yaml.dump(manifest_content))
# Create config in dotfiles
(flow_pkg / "config").write_text(
"[repository]\n"
"dotfiles_url = https://github.com/user/dotfiles.git\n"
)
# Mock paths to use our temp directories
monkeypatch.setattr(paths_module, "DOTFILES_MANIFEST", flow_pkg / "manifest.yaml")
monkeypatch.setattr(paths_module, "DOTFILES_CONFIG", flow_pkg / "config")
monkeypatch.setattr(paths_module, "MANIFEST_FILE", home / ".config" / "devflow" / "manifest.yaml")
monkeypatch.setattr(paths_module, "CONFIG_FILE", home / ".config" / "devflow" / "config")
# Load config and manifest - should come from dotfiles
manifest = load_manifest()
config = load_config()
assert "test-env" in manifest.get("profiles", {})
assert "github.com/user/dotfiles.git" in config.dotfiles_url
def test_manifest_cascade_with_symlink(tmp_path, monkeypatch):
"""Test that loading works correctly when symlink is used."""
# Setup
dotfiles = tmp_path / "dotfiles"
home_config = tmp_path / "home" / ".config" / "flow"
flow_pkg = dotfiles / "flow" / ".config" / "flow"
flow_pkg.mkdir(parents=True)
home_config.mkdir(parents=True)
# Create manifest in dotfiles
manifest_content = {"profiles": {"from-dotfiles": {"os": "linux"}}}
(flow_pkg / "manifest.yaml").write_text(yaml.dump(manifest_content))
# Create symlink from home config to dotfiles
manifest_link = home_config / "manifest.yaml"
manifest_link.symlink_to(flow_pkg / "manifest.yaml")
# Mock paths
monkeypatch.setattr(paths_module, "DOTFILES_MANIFEST", flow_pkg / "manifest.yaml")
monkeypatch.setattr(paths_module, "MANIFEST_FILE", manifest_link)
# Load - should work through symlink
manifest = load_manifest()
assert "from-dotfiles" in manifest.get("profiles", {})
def test_config_priority_documentation(mock_paths):
"""Document the config loading priority for users."""
# This test serves as documentation of the cascade behavior
# Priority 1: Dotfiles repo (self-hosted)
dotfiles_manifest = mock_paths["dotfiles_manifest"]
dotfiles_manifest.parent.mkdir(parents=True)
dotfiles_manifest.write_text("profiles:\n priority-1: {}")
manifest = load_manifest()
assert "priority-1" in manifest.get("profiles", {})
# If we remove dotfiles, falls back to Priority 2: Local override
dotfiles_manifest.unlink()
local_manifest = mock_paths["local_manifest"]
local_manifest.write_text("profiles:\n priority-2: {}")
manifest = load_manifest()
assert "priority-2" in manifest.get("profiles", {})
# If neither exists, Priority 3: Empty fallback
local_manifest.unlink()
manifest = load_manifest()
assert manifest == {}
manifest = load_manifest(one_file)
assert "only" in manifest["profiles"]

5
tests/test_variables.py
View File

@@ -50,3 +50,8 @@ def test_substitute_template_non_string():
def test_substitute_template_no_placeholders():
result = substitute_template("plain text", {"os": "linux"})
assert result == "plain text"
def test_substitute_template_env_namespace():
result = substitute_template("{{ env.USER_EMAIL }}", {"env": {"USER_EMAIL": "you@example.com"}})
assert result == "you@example.com"