Files
flow/tests/test_domain_packages.py
Tomas Mirchev 6b7a48bb20 refactor: post-review hardening pass
Independent re-audit surfaced 11 follow-ups across two layers of review
(my fresh-eyes read + a parallel agent pass). Bundled into a single
commit because changes are small and intertwined.

Symlink / state consistency:
- FileSystem.same_symlink now uses raw readlink() instead of resolve().
  Aligns the three sites that ask "is this our link?" (_load_state,
  _check_overwrite_safe, remove_symlink) on a single rule: exact-readlink
  match. Following symlink chains would let externally-modified links
  pass as ours and be silently overwritten.
- LinkedState.from_dict raises ConfigError on missing required fields
  instead of .get(..., False) silent defaults. Matches InstalledState.
- LinkOp.source is now consistently None for remove_link ops; the
  service derives expected_source from current.links. Removes the
  asymmetry between in-state and orphan-broken removal ops.
- _apply_plan: rename shadowing local from link_target to spec.

Fail loud:
- _xdg() now treats XDG_CONFIG_HOME="" the same as unset. Previously
  an empty env var produced Path("") and state files were written to
  $PWD instead of ~/.local/state/flow.
- _resolve_target raises PlanConflict when a package contains a bare
  _root entry (no path components) instead of silently dropping it.
- _strip_prefix raises FlowError when a declared install path does not
  start with its section's expected prefix (e.g. etc/foo under install.bin).

Speculative abstraction removed (CLAUDE.md):
- core.template.substitute (the $VAR form) had no production callers --
  deleted along with its tests; only the {{var}} form remains.
- SetupModule base class -- five subclasses, no shared behaviour, no
  polymorphic call site. Deleted.
- Profile.arch -- parsed but never read. Deleted.
- PackagePlan.pm_command -- set but never read. Deleted (service
  recomputes pm_install_command at the call site).
- FileSystem.ensure_dir(mode=...), .copy_file(sudo=...), .read_text(
  default=...) -- no callers. Deleted along with their test.
- bootstrap _execute_action: the upfront `phase not in VALID_PHASES`
  check duplicated the trailing exhaustive raise. Kept the trailing
  raise as the single source of truth; phase set still documented in
  VALID_PHASES.

Completion ctx threading:
- Removed _config()/_manifest() helpers that re-loaded from disk on
  every completion call. _list_targets, _list_namespaces, _list_platforms,
  _list_bootstrap_profiles, _list_manifest_packages now take ctx and
  read from ctx.config / ctx.manifest.

Test coverage and e2e:
- e2e container test exercises a real `flow dotfiles link` (no dry-run)
  and asserts the resulting symlinks point into the dotfiles dir;
  reruns to verify idempotency.
- New tests: LinkedState corrupt-state ConfigError, LinkedState bad-version
  ConfigError, bare-_root PlanConflict, service-level _root path routing
  + skip semantics.
- 11 stale test imports removed (pyflakes clean across src/ + tests/).

357 unit tests + 1 e2e (gated) all pass.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-14 00:23:06 +03:00

248 lines
8.6 KiB
Python

"""Tests for packages catalog and resolution."""
from flow.domain.packages.catalog import normalize_profile_entry, parse_catalog
from flow.domain.packages.planning import plan_install
from flow.domain.packages.resolution import (
detect_package_manager,
pm_cask_install_command,
pm_install_command,
pm_update_command,
resolve_binary_asset,
resolve_download_url,
resolve_extract_dir,
resolve_source_name,
resolve_spec,
)
from flow.domain.packages.models import InstalledState, PackageDef, ProfilePackageRef
class TestParseCatalog:
def test_list_format(self):
manifest = {"packages": [
{"name": "fd", "type": "pkg", "apt": "fd-find"},
{"name": "ripgrep", "type": "pkg"},
]}
catalog = parse_catalog(manifest)
assert "fd" in catalog
assert catalog["fd"].sources.get("apt") == "fd-find"
assert "ripgrep" in catalog
def test_dict_format(self):
manifest = {"packages": {
"fd": {"type": "pkg", "apt": "fd-find"},
}}
catalog = parse_catalog(manifest)
assert "fd" in catalog
assert catalog["fd"].type == "pkg"
def test_empty_manifest(self):
assert parse_catalog({}) == {}
class TestNormalizeProfileEntry:
def test_string_shorthand_with_type(self):
ref = normalize_profile_entry("binary/neovim")
assert ref.name == "neovim"
assert ref.type == "binary"
def test_plain_name(self):
ref = normalize_profile_entry("fd")
assert ref.name == "fd"
assert ref.type is None
def test_dict_entry(self):
ref = normalize_profile_entry({"name": "nvim", "type": "binary", "version": "0.10"})
assert ref.name == "nvim"
assert ref.type == "binary"
assert ref.version == "0.10"
class TestResolveSpec:
def test_merge_with_catalog(self):
catalog = {"fd": PackageDef(
name="fd", type="pkg", sources={"apt": "fd-find"},
source=None, version=None, asset_pattern=None,
platform_map={}, extract_dir=None, install={},
post_install=None,
)}
ref = ProfilePackageRef(name="fd", type=None, source=None, version="1.0", asset_pattern=None)
result = resolve_spec(ref, catalog)
assert result.type == "pkg" # from catalog
assert result.version == "1.0" # from profile
assert result.sources == {"apt": "fd-find"} # from catalog
def test_not_in_catalog(self):
ref = ProfilePackageRef(name="unknown", type="binary", source="github:u/r", version=None, asset_pattern=None)
result = resolve_spec(ref, {})
assert result.name == "unknown"
assert result.type == "binary"
def test_profile_object_overrides_catalog(self):
catalog = {"docker": PackageDef(
name="docker", type="pkg", sources={"apt": "docker-ce"},
source=None, version=None, asset_pattern=None,
platform_map={}, extract_dir=None, install={},
post_install=None,
)}
ref = ProfilePackageRef(
name="docker",
type=None,
source=None,
version=None,
asset_pattern=None,
post_install="sudo groupadd docker || true",
)
result = resolve_spec(ref, catalog)
assert result.post_install == "sudo groupadd docker || true"
class TestResolveSourceName:
def test_with_pm_mapping(self):
pkg = PackageDef(
name="fd", type="pkg", sources={"apt": "fd-find"},
source=None, version=None, asset_pattern=None,
platform_map={}, extract_dir=None, install={},
post_install=None,
)
assert resolve_source_name(pkg, "apt") == "fd-find"
def test_fallback_to_name(self):
pkg = PackageDef(
name="fd", type="pkg", sources={},
source=None, version=None, asset_pattern=None,
platform_map={}, extract_dir=None, install={},
post_install=None,
)
assert resolve_source_name(pkg, "apt") == "fd"
class TestResolveBinaryAsset:
def test_platform_map(self):
pkg = PackageDef(
name="nvim", type="binary", sources={},
source="github:neovim/neovim",
version="v0.10.4",
asset_pattern=None,
platform_map={"linux-x64": "nvim-linux-x86_64.tar.gz"},
extract_dir=None, install={},
post_install=None,
)
assert resolve_binary_asset(pkg, "linux-x64") == "nvim-linux-x86_64.tar.gz"
def test_asset_pattern(self):
pkg = PackageDef(
name="fd", type="binary", sources={},
source="github:sharkdp/fd",
version="v10.2.0",
asset_pattern="fd-v10.2.0-{{arch}}-unknown-{{os}}-gnu.tar.gz",
platform_map={},
extract_dir=None, install={},
post_install=None,
)
result = resolve_binary_asset(pkg, "linux-x64")
assert "x64" in result
assert "linux" in result
def test_double_brace_pattern_uses_platform_map_context(self):
pkg = PackageDef(
name="nvim", type="binary", sources={},
source="github:neovim/neovim",
version="0.10.4",
asset_pattern="nvim-{{os}}-{{arch}}.tar.gz",
platform_map={"linux-x64": {"os": "linux", "arch": "x86_64"}},
extract_dir="nvim-{{os}}64", install={},
post_install=None,
)
assert resolve_binary_asset(pkg, "linux-x64") == "nvim-linux-x86_64.tar.gz"
assert resolve_extract_dir(pkg, "linux-x64") == "nvim-linux64"
class TestResolveDownloadUrl:
def test_github_shorthand_with_version(self):
pkg = PackageDef(
name="nvim", type="binary", sources={},
source="github:neovim/neovim",
version="v0.10.4",
asset_pattern=None, platform_map={},
extract_dir=None, install={},
post_install=None,
)
url = resolve_download_url(pkg, "nvim.tar.gz")
assert "github.com/neovim/neovim" in url
assert "v0.10.4" in url
def test_github_shorthand_prefixes_v(self):
pkg = PackageDef(
name="nvim", type="binary", sources={},
source="github:neovim/neovim",
version="0.10.4",
asset_pattern=None, platform_map={},
extract_dir=None, install={},
post_install=None,
)
url = resolve_download_url(pkg, "nvim.tar.gz", "linux-x64")
assert "/download/v0.10.4/" in url
def test_github_latest(self):
pkg = PackageDef(
name="nvim", type="binary", sources={},
source="github:neovim/neovim",
version=None,
asset_pattern=None, platform_map={},
extract_dir=None, install={},
post_install=None,
)
url = resolve_download_url(pkg, "nvim.tar.gz")
assert "latest" in url
def test_direct_url(self):
pkg = PackageDef(
name="x", type="binary", sources={},
source="https://example.com/download/",
version=None,
asset_pattern=None, platform_map={},
extract_dir=None, install={},
post_install=None,
)
url = resolve_download_url(pkg, "x.tar.gz")
assert url == "https://example.com/download/x.tar.gz"
class TestPmCommands:
def test_apt_update(self):
assert "apt-get update" in pm_update_command("apt")
def test_dnf_update(self):
assert "dnf" in pm_update_command("dnf")
def test_brew_install(self):
cmd = pm_install_command("brew", ["fd", "rg"])
assert "brew install" in cmd
assert "fd" in cmd
def test_apt_install(self):
cmd = pm_install_command("apt", ["fd-find"])
assert "apt-get install" in cmd
def test_brew_cask_install(self):
cmd = pm_cask_install_command("brew", ["wezterm"])
assert "--cask" in cmd
assert "wezterm" in cmd
def test_detect_package_manager_returns_something(self):
# Just verify it doesn't error
result = detect_package_manager()
assert result is None or result in ("apt", "dnf", "brew")
class TestPlanning:
def test_cask_package_is_planned(self):
pkg = PackageDef(
name="wezterm", type="cask", sources={"brew": "wezterm"},
source=None, version=None, asset_pattern=None,
platform_map={}, extract_dir=None, install={},
post_install=None,
)
plan = plan_install([pkg], InstalledState(), "macos-arm64", "brew")
assert plan.install_ops[0].method == "cask"