"""Tests for PackageService.""" import io import tarfile import urllib.error from pathlib import Path import pytest from flow.core.config import AppConfig, FlowContext from flow.core.console import Console from flow.core.errors import FlowError from flow.core.platform import PlatformInfo from flow.core.runtime import SystemRuntime from flow.core import paths from flow.domain.packages.models import InstalledPackage, InstalledState, PackageDef from flow.services.packages import PackageService def _make_ctx(tmp_path, manifest=None): return FlowContext( config=AppConfig(), manifest=manifest or {}, platform=PlatformInfo(), console=Console(color=False), runtime=SystemRuntime(), ) class TestPackageService: def test_list_empty(self, tmp_path, monkeypatch, capsys): monkeypatch.setattr(paths, "INSTALLED_STATE", tmp_path / "installed.json") ctx = _make_ctx(tmp_path) svc = PackageService(ctx) svc.list_packages() assert "No packages" in capsys.readouterr().out def test_list_shows_installed(self, tmp_path, monkeypatch, capsys): state = InstalledState(packages={ "fd": InstalledPackage(name="fd", version="10.2", type="pkg"), }) state_path = tmp_path / "installed.json" import json state_path.parent.mkdir(parents=True, exist_ok=True) with open(state_path, "w") as f: json.dump(state.as_dict(), f) monkeypatch.setattr(paths, "INSTALLED_STATE", state_path) ctx = _make_ctx(tmp_path) svc = PackageService(ctx) svc.list_packages() output = capsys.readouterr().out assert "fd" in output assert "10.2" in output def test_remove_not_installed(self, tmp_path, monkeypatch, capsys): monkeypatch.setattr(paths, "INSTALLED_STATE", tmp_path / "installed.json") ctx = _make_ctx(tmp_path) svc = PackageService(ctx) with pytest.raises(FlowError, match="not installed"): svc.remove(["missing"]) def test_install_requires_args(self, tmp_path, monkeypatch): monkeypatch.setattr(paths, "INSTALLED_STATE", tmp_path / "installed.json") ctx = _make_ctx(tmp_path) svc = PackageService(ctx) with pytest.raises(FlowError, match="Specify"): svc.install() def test_list_all_known_packages(self, tmp_path, monkeypatch, capsys): monkeypatch.setattr(paths, "INSTALLED_STATE", tmp_path / "installed.json") manifest = {"packages": [{"name": "fd", "type": "pkg"}]} ctx = _make_ctx(tmp_path, manifest) svc = PackageService(ctx) svc.list_packages(show_all=True) assert "fd" in capsys.readouterr().out def test_install_binary_honors_declared_install_map(self, tmp_path, monkeypatch): home = tmp_path / "home" home.mkdir() monkeypatch.setenv("HOME", str(home)) monkeypatch.setattr(paths, "HOME", home) monkeypatch.setattr(paths, "DATA_DIR", tmp_path / "data") monkeypatch.setattr(paths, "INSTALLED_STATE", tmp_path / "installed.json") archive = io.BytesIO() with tarfile.open(fileobj=archive, mode="w:gz") as tar: files = { "nvim-linux64/bin/nvim": b"#!/bin/sh\n", "nvim-linux64/share/nvim/runtime.txt": b"runtime\n", "nvim-linux64/share/man/man1/nvim.1": b"manpage\n", } for name, content in files.items(): info = tarfile.TarInfo(name=name) info.size = len(content) tar.addfile(info, io.BytesIO(content)) archive_bytes = archive.getvalue() class FakeResponse: def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return False def read(self): return archive_bytes monkeypatch.setattr("flow.adapters.download.urllib.request.urlopen", lambda *args, **kwargs: FakeResponse()) manifest = { "packages": [{ "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"}}, "extract-dir": "nvim-{{os}}64", "install": { "bin": ["bin/nvim"], "share": ["share/nvim"], "man": ["share/man/man1/nvim.1"], }, }], } ctx = _make_ctx(tmp_path, manifest) svc = PackageService(ctx) packages = svc.resolve_install_packages(package_names=["neovim"]) svc.install(packages) assert (home / ".local" / "bin" / "nvim").exists() assert (home / ".local" / "share" / "nvim" / "runtime.txt").exists() assert (home / ".local" / "share" / "man" / "man1" / "nvim.1").exists() def test_post_install_with_sudo_runs_unchecked(self, tmp_path, monkeypatch): """No allow_sudo gate -- post-install scripts run as written.""" home = tmp_path / "home" home.mkdir() monkeypatch.setattr(paths, "HOME", home) monkeypatch.setattr(paths, "INSTALLED_STATE", tmp_path / "installed.json") calls: list[str] = [] class _Runner: def run_shell(self, command, **kwargs): calls.append(command) class _Result: returncode = 0 stdout = "" stderr = "" return _Result() ctx = _make_ctx(tmp_path) ctx.runtime.runner = _Runner() svc = PackageService(ctx) pkg = PackageDef( name="docker", type="pkg", sources={}, source=None, version=None, asset_pattern=None, platform_map={}, extract_dir=None, install={}, post_install="sudo groupadd docker || true", ) svc._run_post_install(pkg) assert calls == ["sudo groupadd docker || true"] def test_install_binary_url_failure_raises_flow_error(self, tmp_path, monkeypatch): home = tmp_path / "home" home.mkdir() monkeypatch.setattr(paths, "HOME", home) monkeypatch.setattr(paths, "DATA_DIR", tmp_path / "data") monkeypatch.setattr(paths, "INSTALLED_STATE", tmp_path / "installed.json") def _raise(*args, **kwargs): raise urllib.error.URLError("Network unreachable") monkeypatch.setattr( "flow.adapters.download.urllib.request.urlopen", _raise, ) manifest = { "packages": [{ "name": "neovim", "type": "binary", "source": "github:neovim/neovim", "version": "0.10.4", "platform-map": {"linux-x64": "nvim-linux-x64.tar.gz"}, "install": {"bin": ["bin/nvim"]}, }], } ctx = _make_ctx(tmp_path, manifest) svc = PackageService(ctx) packages = svc.resolve_install_packages(package_names=["neovim"]) with pytest.raises(FlowError, match="Failed to download"): svc.install(packages) def test_install_path_absolute_raises(self, tmp_path, monkeypatch): monkeypatch.setattr(paths, "HOME", tmp_path / "home") ctx = _make_ctx(tmp_path) svc = PackageService(ctx) with pytest.raises(FlowError, match="must be relative"): svc._validate_install_path("pkg", Path("/etc/passwd")) def test_install_path_parent_traversal_raises(self, tmp_path, monkeypatch): monkeypatch.setattr(paths, "HOME", tmp_path / "home") ctx = _make_ctx(tmp_path) svc = PackageService(ctx) with pytest.raises(FlowError, match="parent traversal"): svc._validate_install_path("pkg", Path("../etc/passwd")) def test_install_path_escapes_extract_dir_raises(self, tmp_path, monkeypatch): """A relative path whose resolved location is outside the extract dir.""" home = tmp_path / "home" home.mkdir() monkeypatch.setattr(paths, "HOME", home) ctx = _make_ctx(tmp_path) svc = PackageService(ctx) extract_root = tmp_path / "extract" extract_root.mkdir() sibling = tmp_path / "sibling" sibling.mkdir() # Symlink inside the extract root pointing outside -- the resolved # source escapes the root. link = extract_root / "evil" link.symlink_to(sibling) with pytest.raises(FlowError, match="escapes extract-dir"): svc._copy_install_item( "pkg", extract_root, extract_root.resolve(), "bin", "evil/escape", )