277 lines
11 KiB
Python
277 lines
11 KiB
Python
"""Tests for PackageService."""
|
|
|
|
import io
|
|
import subprocess
|
|
import tarfile
|
|
import urllib.error
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from flow.actions import ActionExecutor, ActionPlan, PrimitiveAction, RollbackPolicy
|
|
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.app.packages import PackageService
|
|
from tests.fakes import FakeRunner
|
|
|
|
|
|
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_install_does_not_write_state_when_package_manager_install_fails(
|
|
self, tmp_path, monkeypatch,
|
|
):
|
|
state_path = tmp_path / "installed.json"
|
|
monkeypatch.setattr(paths, "INSTALLED_STATE", state_path)
|
|
monkeypatch.setattr("flow.app.packages.detect_package_manager", lambda: "apt")
|
|
|
|
class FailingInstallRunner(FakeRunner):
|
|
def run(
|
|
self, argv, *, cwd=None, env=None, capture_output=True,
|
|
check=False, timeout=None,
|
|
):
|
|
parts = list(argv)
|
|
self.calls.append(parts)
|
|
self.timeouts.append(timeout)
|
|
if parts[:3] == ["sudo", "apt-get", "install"]:
|
|
return subprocess.CompletedProcess(
|
|
parts, 42, stdout="", stderr="install failed"
|
|
)
|
|
return subprocess.CompletedProcess(parts, 0, stdout="", stderr="")
|
|
|
|
rt = SystemRuntime()
|
|
rt.runner = FailingInstallRunner()
|
|
ctx = FlowContext(
|
|
config=AppConfig(),
|
|
manifest={},
|
|
platform=PlatformInfo(),
|
|
console=Console(color=False),
|
|
runtime=rt,
|
|
)
|
|
svc = PackageService(ctx)
|
|
pkg = PackageDef(
|
|
name="fd", type="pkg", sources={},
|
|
source=None, version=None, asset_pattern=None,
|
|
platform_map={}, extract_dir=None, install={},
|
|
post_install=None,
|
|
)
|
|
|
|
with pytest.raises(FlowError, match="install failed"):
|
|
svc.install([pkg])
|
|
|
|
assert not state_path.exists()
|
|
|
|
def test_post_install_with_sudo_runs_unchecked(self, tmp_path, monkeypatch):
|
|
"""No allow_sudo gate -- post-install scripts run as written."""
|
|
ctx = _make_ctx(tmp_path)
|
|
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",
|
|
)
|
|
primitive = svc._post_install_primitive(pkg)
|
|
assert primitive is not None
|
|
assert primitive.type == "process.shell_user_hook"
|
|
assert primitive.payload["command"] == "sudo groupadd docker || true"
|
|
assert primitive.rollback_policy == RollbackPolicy.BARRIER
|
|
|
|
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 allowed root"):
|
|
ActionExecutor(ctx, audit_path=tmp_path / "actions.jsonl").execute(
|
|
ActionPlan(
|
|
name="copy-escape",
|
|
primitive_actions=(
|
|
PrimitiveAction(
|
|
id="copy",
|
|
type="file.copy",
|
|
description="Copy escaped source",
|
|
payload={
|
|
"source": link / "escape",
|
|
"target": tmp_path / "target",
|
|
"source_root": extract_root.resolve(),
|
|
},
|
|
),
|
|
),
|
|
)
|
|
)
|