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

example

Browse Source
This commit is contained in:
Tomas Mirchev
2026-02-25 15:32:23 +02:00
parent 6cff65f288
commit 741173f617
24 changed files with 1205 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files

72
tests/test_bootstrap.py
View File

@@ -1,12 +1,14 @@
"""Tests for flow.commands.bootstrap helpers and schema behavior."""
import os
from pathlib import Path
import pytest
from flow.commands.bootstrap import (
_ensure_required_variables,
_get_profiles,
_install_binary_package,
_normalize_profile_package_entry,
_resolve_package_manager,
_resolve_package_spec,
@@ -141,3 +143,73 @@ def test_ensure_required_variables_accepts_vars(monkeypatch):
env["USER_EMAIL"] = "a@b"
env["TARGET_HOSTNAME"] = "devbox"
_ensure_required_variables({"requires": ["USER_EMAIL", "TARGET_HOSTNAME"]}, env)
class _FakeResponse:
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
return False
def read(self):
return b"archive"
def _patch_binary_download(monkeypatch, after_unpack=None):
monkeypatch.setattr(
"flow.commands.bootstrap.urllib.request.urlopen",
lambda *args, **kwargs: _FakeResponse(),
)
def _fake_unpack(_archive, extract_dir):
extracted = Path(extract_dir)
extracted.mkdir(parents=True, exist_ok=True)
if after_unpack:
after_unpack(extracted)
monkeypatch.setattr("flow.commands.bootstrap.shutil.unpack_archive", _fake_unpack)
def test_install_binary_package_rejects_absolute_declared_path(monkeypatch, tmp_path, ctx):
absolute_item = tmp_path / "outside-bin"
absolute_item.write_text("binary")
_patch_binary_download(monkeypatch)
monkeypatch.setattr(
"flow.commands.bootstrap._copy_install_item",
lambda *args, **kwargs: pytest.fail("_copy_install_item should not be called"),
)
spec = {
"name": "demo",
"type": "binary",
"source": "https://example.invalid/demo",
"asset-pattern": "demo.tar.gz",
"install": {"bin": [str(absolute_item)]},
}
with pytest.raises(RuntimeError, match="must be relative"):
_install_binary_package(ctx, spec, {}, dry_run=False)
def test_install_binary_package_rejects_parent_traversal_declared_path(monkeypatch, ctx):
def _after_unpack(extracted):
(extracted.parent / "escape-bin").write_text("binary")
_patch_binary_download(monkeypatch, after_unpack=_after_unpack)
monkeypatch.setattr(
"flow.commands.bootstrap._copy_install_item",
lambda *args, **kwargs: pytest.fail("_copy_install_item should not be called"),
)
spec = {
"name": "demo",
"type": "binary",
"source": "https://example.invalid/demo",
"asset-pattern": "demo.tar.gz",
"install": {"bin": ["../escape-bin"]},
}
with pytest.raises(RuntimeError, match="parent traversal"):
_install_binary_package(ctx, spec, {}, dry_run=False)

48
tests/test_completion.py
View File

@@ -36,6 +36,43 @@ def test_complete_package_remove(monkeypatch):
assert out == ["hello"]
def test_list_manifest_packages_is_consistent_for_list_and_dict_forms(monkeypatch):
manifests = [
{
"packages": [
{"name": "neovim", "type": "binary"},
{"name": "ripgrep", "type": "pkg"},
{"name": "fzf", "type": "binary"},
]
},
{
"packages": {
"neovim": {"type": "binary"},
"ripgrep": {"type": "pkg"},
"fzf": {"type": "binary"},
}
},
]
monkeypatch.setattr(completion, "_safe_manifest", lambda: manifests.pop(0))
from_list = completion._list_manifest_packages()
from_dict = completion._list_manifest_packages()
assert from_list == ["fzf", "neovim"]
assert from_dict == ["fzf", "neovim"]
def test_list_manifest_packages_uses_mapping_key_when_name_missing(monkeypatch):
monkeypatch.setattr(
completion,
"_safe_manifest",
lambda: {"packages": {"bat": {"type": "binary"}, "git": {"type": "pkg"}}},
)
assert completion._list_manifest_packages() == ["bat"]
def test_complete_dotfiles_profile_value(monkeypatch):
monkeypatch.setattr(completion, "_list_dotfiles_profiles", lambda: ["work", "personal"])
out = completion.complete(["flow", "dotfiles", "link", "--profile", "w"], 5)
@@ -47,6 +84,17 @@ def test_complete_dotfiles_repo_subcommands():
assert out == ["pull", "push"]
def test_complete_dotfiles_modules_subcommands():
out = completion.complete(["flow", "dotfiles", "modules", "s"], 4)
assert out == ["sync"]
def test_complete_dotfiles_modules_profile_value(monkeypatch):
monkeypatch.setattr(completion, "_list_dotfiles_profiles", lambda: ["work", "personal"])
out = completion.complete(["flow", "dotfiles", "modules", "list", "--profile", "w"], 6)
assert out == ["work"]
def test_complete_enter_targets(monkeypatch):
monkeypatch.setattr(completion, "_list_targets", lambda: ["personal@orb", "work@ec2"])
out = completion.complete(["flow", "enter", "p"], 3)

13
tests/test_config.py
View File

@@ -10,6 +10,7 @@ def test_load_config_missing_path(tmp_path):
assert isinstance(cfg, AppConfig)
assert cfg.dotfiles_url == ""
assert cfg.container_registry == "registry.tomastm.com"
assert cfg.dotfiles_pull_before_edit is True
def test_load_config_merged_yaml(tmp_path):
@@ -17,6 +18,7 @@ def test_load_config_merged_yaml(tmp_path):
"repository:\n"
" dotfiles-url: git@github.com:user/dots.git\n"
" dotfiles-branch: dev\n"
" pull-before-edit: false\n"
"paths:\n"
" projects-dir: ~/code\n"
"defaults:\n"
@@ -31,6 +33,7 @@ def test_load_config_merged_yaml(tmp_path):
cfg = load_config(tmp_path)
assert cfg.dotfiles_url == "git@github.com:user/dots.git"
assert cfg.dotfiles_branch == "dev"
assert cfg.dotfiles_pull_before_edit is False
assert cfg.projects_dir == "~/code"
assert cfg.container_registry == "my.registry.com"
assert cfg.container_tag == "v1"
@@ -40,6 +43,16 @@ def test_load_config_merged_yaml(tmp_path):
assert cfg.targets[1].ssh_identity == "~/.ssh/id_work"
def test_load_config_pull_before_edit_string_true(tmp_path):
(tmp_path / "10-config.yaml").write_text(
"repository:\n"
" pull-before-edit: yes\n"
)
cfg = load_config(tmp_path)
assert cfg.dotfiles_pull_before_edit is True
def test_load_manifest_missing_path(tmp_path):
result = load_manifest(tmp_path / "nonexistent")
assert result == {}

35
tests/test_dotfiles.py
View File

@@ -2,7 +2,10 @@
import pytest
from flow.commands.dotfiles import _discover_packages, _resolve_edit_target, _walk_package
from flow.commands.dotfiles import _collect_home_specs, _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
def _make_tree(tmp_path):
@@ -20,6 +23,15 @@ def _make_tree(tmp_path):
return tmp_path
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 test_discover_packages_shared_only(tmp_path):
tree = _make_tree(tmp_path)
packages = _discover_packages(tree)
@@ -43,8 +55,7 @@ def test_discover_packages_profile_overrides_shared(tmp_path):
(profile_zsh / ".zshrc").write_text("# 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)
_collect_home_specs(_ctx(), tree, tmp_path / "home", "work", set(), None)
def test_walk_package_returns_relative_paths(tmp_path):
@@ -70,6 +81,24 @@ def test_resolve_edit_target_repo_path(tmp_path):
assert target == tree / "_shared" / "zsh" / ".zshrc"
def test_resolve_edit_target_rejects_parent_traversal(tmp_path):
tree = _make_tree(tmp_path / "repo")
outside = tmp_path / "outside.txt"
outside.write_text("secret")
target = _resolve_edit_target("../outside.txt", dotfiles_dir=tree)
assert target is None
def test_resolve_edit_target_rejects_nested_repo_escape(tmp_path):
tree = _make_tree(tmp_path / "repo")
outside = tmp_path / "escape.txt"
outside.write_text("secret")
target = _resolve_edit_target("_shared/../../escape.txt", dotfiles_dir=tree)
assert target 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

207
tests/test_dotfiles_folding.py
View File

@@ -1,6 +1,7 @@
"""Tests for flat-layout dotfiles helpers and state format."""
"""Tests for dotfiles link planning, root markers, and module sources."""
import json
import subprocess
from pathlib import Path
import pytest
@@ -8,11 +9,25 @@ import pytest
from flow.commands.dotfiles import (
LinkSpec,
_collect_home_specs,
_collect_root_specs,
_list_profiles,
_load_link_specs_from_state,
_pull_requires_ack,
_resolved_package_source,
_save_link_specs_to_state,
_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:
@@ -25,12 +40,9 @@ def _make_flow_tree(tmp_path: Path) -> Path:
(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")
(flow_root / "_shared" / "dnsmasq" / "_root" / "etc").mkdir(parents=True)
(flow_root / "_shared" / "dnsmasq" / "_root" / "etc" / "hostname").write_text("devbox")
return flow_root
@@ -47,14 +59,33 @@ def test_collect_home_specs_conflict_fails(tmp_path):
home.mkdir()
with pytest.raises(RuntimeError, match="Conflicting dotfile targets"):
_collect_home_specs(flow_root, home, "work", set(), None)
_collect_home_specs(_ctx(), flow_root, home, "work", set(), None)
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_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):
@@ -92,3 +123,153 @@ def test_state_old_format_rejected(tmp_path, monkeypatch):
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

45
tests/test_package.py Normal file
View File

@@ -0,0 +1,45 @@
"""Tests for flow.commands.package."""
from types import SimpleNamespace
from flow.commands import package
def test_load_installed_returns_empty_on_malformed_json(tmp_path, monkeypatch):
state_file = tmp_path / "installed.json"
state_file.write_text("{broken", encoding="utf-8")
monkeypatch.setattr(package, "INSTALLED_STATE", state_file)
assert package._load_installed() == {}
def test_load_installed_returns_empty_on_non_mapping_json(tmp_path, monkeypatch):
state_file = tmp_path / "installed.json"
state_file.write_text('["neovim"]', encoding="utf-8")
monkeypatch.setattr(package, "INSTALLED_STATE", state_file)
assert package._load_installed() == {}
class _ConsoleCapture:
def __init__(self):
self.info_messages = []
def info(self, message):
self.info_messages.append(message)
def table(self, _headers, _rows):
raise AssertionError("table() should not be called when installed state is malformed")
def test_run_list_handles_malformed_installed_state(tmp_path, monkeypatch):
state_file = tmp_path / "installed.json"
state_file.write_text("{oops", encoding="utf-8")
monkeypatch.setattr(package, "INSTALLED_STATE", state_file)
monkeypatch.setattr(package, "_get_definitions", lambda _ctx: {})
ctx = SimpleNamespace(console=_ConsoleCapture())
package.run_list(ctx, SimpleNamespace(all=False))
assert ctx.console.info_messages == ["No packages installed."]

7
tests/test_paths.py
View File

@@ -10,6 +10,7 @@ from flow.core.paths import (
INSTALLED_STATE,
LINKED_STATE,
MANIFEST_FILE,
MODULES_DIR,
PACKAGES_DIR,
SCRATCH_DIR,
STATE_DIR,
@@ -45,6 +46,10 @@ def test_packages_dir():
assert PACKAGES_DIR == DATA_DIR / "packages"
def test_modules_dir():
assert MODULES_DIR == DATA_DIR / "modules"
def test_scratch_dir():
assert SCRATCH_DIR == DATA_DIR / "scratch"
@@ -58,6 +63,7 @@ def test_ensure_dirs(tmp_path, monkeypatch):
monkeypatch.setattr("flow.core.paths.CONFIG_DIR", tmp_path / "config")
monkeypatch.setattr("flow.core.paths.DATA_DIR", tmp_path / "data")
monkeypatch.setattr("flow.core.paths.STATE_DIR", tmp_path / "state")
monkeypatch.setattr("flow.core.paths.MODULES_DIR", tmp_path / "data" / "modules")
monkeypatch.setattr("flow.core.paths.PACKAGES_DIR", tmp_path / "data" / "packages")
monkeypatch.setattr("flow.core.paths.SCRATCH_DIR", tmp_path / "data" / "scratch")
@@ -66,5 +72,6 @@ def test_ensure_dirs(tmp_path, monkeypatch):
assert (tmp_path / "config").is_dir()
assert (tmp_path / "data").is_dir()
assert (tmp_path / "state").is_dir()
assert (tmp_path / "data" / "modules").is_dir()
assert (tmp_path / "data" / "packages").is_dir()
assert (tmp_path / "data" / "scratch").is_dir()

88
tests/test_sync.py Normal file
View File

@@ -0,0 +1,88 @@
"""Tests for flow.commands.sync."""
from types import SimpleNamespace
import pytest
from flow.commands import sync
def _git_clean_repo(_repo, *cmd, capture=True):
_ = capture
if cmd == ("rev-parse", "--abbrev-ref", "HEAD"):
return SimpleNamespace(returncode=0, stdout="main\n")
if cmd == ("diff", "--quiet"):
return SimpleNamespace(returncode=0, stdout="")
if cmd == ("diff", "--cached", "--quiet"):
return SimpleNamespace(returncode=0, stdout="")
if cmd == ("ls-files", "--others", "--exclude-standard"):
return SimpleNamespace(returncode=0, stdout="")
if cmd == ("rev-parse", "--abbrev-ref", "main@{u}"):
return SimpleNamespace(returncode=0, stdout="origin/main\n")
if cmd == ("rev-list", "--oneline", "main@{u}..main"):
return SimpleNamespace(returncode=0, stdout="")
if cmd == ("for-each-ref", "--format=%(refname:short)", "refs/heads"):
return SimpleNamespace(returncode=0, stdout="main\n")
raise AssertionError(f"Unexpected git command: {cmd!r}")
@pytest.mark.parametrize("git_style", ["dir", "file"])
def test_check_repo_detects_git_dir_and_worktree_file(tmp_path, monkeypatch, git_style):
repo = tmp_path / "repo"
repo.mkdir()
if git_style == "dir":
(repo / ".git").mkdir()
else:
(repo / ".git").write_text("gitdir: /tmp/worktrees/repo\n", encoding="utf-8")
monkeypatch.setattr(sync, "_git", _git_clean_repo)
name, issues = sync._check_repo(str(repo), do_fetch=False)
assert name == "repo"
assert issues == []
class _ConsoleCapture:
def __init__(self):
self.info_messages = []
self.error_messages = []
self.success_messages = []
def info(self, message):
self.info_messages.append(message)
def error(self, message):
self.error_messages.append(message)
def success(self, message):
self.success_messages.append(message)
def test_run_fetch_includes_worktree_style_repo(tmp_path, monkeypatch):
projects = tmp_path / "projects"
projects.mkdir()
worktree_repo = projects / "worktree"
worktree_repo.mkdir()
(worktree_repo / ".git").write_text("gitdir: /tmp/worktrees/worktree\n", encoding="utf-8")
(projects / "non_git").mkdir()
calls = []
def _git_fetch(repo, *cmd, capture=True):
_ = capture
calls.append((repo, cmd))
return SimpleNamespace(returncode=0, stdout="")
monkeypatch.setattr(sync, "_git", _git_fetch)
console = _ConsoleCapture()
ctx = SimpleNamespace(config=SimpleNamespace(projects_dir=str(projects)), console=console)
sync.run_fetch(ctx, SimpleNamespace())
assert calls == [(str(worktree_repo), ("fetch", "--all", "--quiet"))]
assert console.success_messages == ["All remotes fetched."]