flow/tests/test_service_packages.py

"""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)
        svc.remove(["missing"])
        assert "No matching" in capsys.readouterr().out

    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.services.packages.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.services.packages.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",
            )