diff --git a/README.md b/README.md index c84e2a1..0b1e708 100644 --- a/README.md +++ b/README.md @@ -183,7 +183,9 @@ See [QUICKSTART.md](QUICKSTART.md) for the fuller tour. The command-line client. Runs any VICAR tool in the container, translating host paths, reusing one warm container, mounting calibration data, and forwarding X11 -so GUI tools such as `xvd` display on your desktop. +so GUI tools such as `xvd` display on your desktop. Also a Python API +(`from tig_cli import Tig`) that runs the same tools in the same container and +returns their output and the files they wrote. ๐Ÿ“ `tig-cli/` ยท ๐Ÿ“ฆ [`pip install tig-cli`](https://pypi.org/project/tig-cli/) ยท ๐Ÿ“– [README](tig-cli/README.md) diff --git a/tig-cli/README.md b/tig-cli/README.md index 0bdeafd..1b1591c 100644 --- a/tig-cli/README.md +++ b/tig-cli/README.md @@ -84,6 +84,50 @@ tool untouched: tig --writable-path /data/results marsmap INP=/data/in.vic OUT=/data/results/out.vic ``` +## Python API + +The same thing from Python, with the tool's output captured instead of +written to your terminal: + +```python +from tig_cli import Tig + +with Tig() as tig: + result = tig.run("gen", "test.vic", 64, 64) + result.exit_code # 0 + result.stdout # 'Beginning VICAR task GEN\n...' + result.outputs # [PosixPath('/home/you/work/test.vic')] + + listing = tig.run("label", "-list", inp="test.vic").check() + tig.run("marsxyz", inp=["left.vic", "right.vic"], out="xyz.vic", check=True) + tig.vicario("test.vic", "test.png") # tig.(...) is tig.run("", ...) +``` + +Positional arguments and `keyword=value` parameters are passed as on the command +line; keyword arguments become parameters, and a list becomes a parenthesized +value list such as `inp=(left.vic,right.vic)`. `Path` objects are accepted +anywhere a path is. + +`run()` returns a `Result` with `exit_code`, `stdout`, `stderr`, `ok`, and +`outputs`: the files named in the arguments that the run created or changed, +as host paths. The container writes straight onto the host filesystem, so those +files are in place as soon as `run()` returns - there is nothing to copy back. +Files a tool names itself (a `.mtl` beside an `.obj`) are not listed, but are in +the working directory all the same. + +A non-zero exit is reported, not raised, because VICAR programs often exit +non-zero on success; pass `check=True` or call `result.check()` to get a +`VicarError` instead. `stdin=`, `env=`, `timeout=` (raises `VicarTimeout` and +kills the program in the container) and a per-run `workdir=` are also available. + +`Tig()` takes the same settings as the command-line options - `image`, +`calibration_path`, `writable_paths`, `config`, `disable_path_translation`, +`selinux_label_disable` - and a `workdir` that relative paths are relative to +(the current directory by default). Anything not given comes from the config +files and environment variables described below, so a script and a shell see the +same setup and share the same container. `tig.tools()` lists the programs the +image provides; `tig.shutdown()` removes the container, as `tig --shutdown` does. + ## Configuration Settings can come from TOML config files, environment variables, or command-line diff --git a/tig-cli/src/tig_cli/__init__.py b/tig-cli/src/tig_cli/__init__.py index 2836690..d22d1c4 100644 --- a/tig-cli/src/tig_cli/__init__.py +++ b/tig-cli/src/tig_cli/__init__.py @@ -2,6 +2,20 @@ Configure the tool with layered TOML config files (system, user, project), the CONTAINER_IMAGE / TIG_* environment variables, or command-line flags. + +From Python, ``from tig_cli import Tig`` runs VICAR programs in the same +container; see :mod:`tig_cli.api`. """ __version__ = "0.2.0" + +__all__ = ["Tig", "Result", "VicarError", "VicarTimeout", "run", "__version__"] + + +def __getattr__(name: str): + # Imported on demand so the command-line fast path stays light. + if name in __all__: + from . import api + + return getattr(api, name) + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/tig-cli/src/tig_cli/api.py b/tig-cli/src/tig_cli/api.py new file mode 100644 index 0000000..6b8d546 --- /dev/null +++ b/tig-cli/src/tig_cli/api.py @@ -0,0 +1,435 @@ +"""Run VICAR programs in the TIG container from Python. + +The ``tig`` command hands a tool this process's own terminal; this module +captures what the tool writes and hands it back, together with the files it +produced:: + + from tig_cli import Tig + + with Tig() as tig: + result = tig.run("gen", "test.vic", 64, 64) + result.stdout # 'Beginning VICAR task GEN\\n...' + result.outputs # [PosixPath('/home/you/work/test.vic')] + + listing = tig.run("label", "-list", inp="test.vic").check() + tig.run("marsxyz", inp=["left.vic", "right.vic"], out="xyz.vic") + +Keyword arguments become VICAR ``keyword=value`` parameters; a list becomes +a parenthesized value list. Host paths are translated exactly as they are for +the command line, and the container writes straight onto the host filesystem, +so an output file is on the host as soon as the tool exits - nothing has to +be copied back. + +The container is the same long-lived one the ``tig`` command uses, so a script +and a shell share it, and ``tig --shutdown`` removes it. +""" +from __future__ import annotations + +import os +import signal +import subprocess +import uuid +from pathlib import Path +from typing import Any, Iterable, Mapping, Sequence, Union + +from .config import Config, ConfigError, load_config +from .container import ContainerManager +from .runtime import Runtime +from .spec import ( + EXEC_ID_ENV, + TigError, + container_display, + ensure_x11_ready, + get_calibration_path, + get_container_image, + kill_tree_command, + resolve_disable_path_translation, + resolve_selinux_label_disable, + resolve_writable_paths, +) + +# What run() accepts as a positional argument or a keyword's value. +Value = Union[str, "os.PathLike[str]", int, float] +Param = Union[Value, Sequence[Value]] + +# Seconds allowed for the in-container kill after a timeout. +KILL_TIMEOUT = 5.0 + + +class VicarError(TigError): + """A VICAR program exited non-zero when the caller asked for success.""" + + def __init__(self, result: "Result"): + detail = (result.stderr or result.stdout).strip() + message = f"{result.tool} exited with status {result.exit_code}" + if detail: + message += f": {detail.splitlines()[-1].strip()}" + super().__init__(message) + self.result = result + + +class VicarTimeout(TigError): + """A VICAR program was stopped for running longer than allowed.""" + + def __init__(self, tool: str, timeout: float, stdout: str, stderr: str): + super().__init__(f"{tool} did not finish within {timeout:g} seconds") + self.tool = tool + self.timeout = timeout + self.stdout = stdout + self.stderr = stderr + + +class Result: + """What one VICAR program run produced. + + Attributes: + tool: The program that was run. + command: The full command line as run in the container, paths + translated. + exit_code: The program's exit status. + stdout: Everything it wrote to standard output. + stderr: Everything it wrote to standard error. + outputs: Host paths named in the arguments that the run created or + modified, in argument order. Files a tool writes under names of + its own (``marsmesh``'s ``.mtl`` beside the ``.obj``, say) are + not listed but are in the working directory all the same. + """ + + def __init__( + self, + tool: str, + command: list[str], + exit_code: int, + stdout: str, + stderr: str, + outputs: list[Path], + ): + self.tool = tool + self.command = command + self.exit_code = exit_code + self.stdout = stdout + self.stderr = stderr + self.outputs = outputs + + @property + def ok(self) -> bool: + """Whether the program exited with status 0.""" + return self.exit_code == 0 + + @property + def output(self) -> Path | None: + """The single output file, when there is exactly one.""" + return self.outputs[0] if len(self.outputs) == 1 else None + + def check(self) -> "Result": + """Return this result, or raise :class:`VicarError` if it failed.""" + if not self.ok: + raise VicarError(self) + return self + + def __bool__(self) -> bool: + return self.ok + + def __repr__(self) -> str: + return ( + f"Result(tool={self.tool!r}, exit_code={self.exit_code}, " + f"outputs={[str(path) for path in self.outputs]!r})" + ) + + +class Tig: + """A handle on the TIG container, for running VICAR programs. + + Settings default to what the ``tig`` command would use - the layered config + files and the ``CONTAINER_IMAGE`` / ``MARS_CONFIG_PATH`` / ``TIG_*`` + environment variables - and each argument here overrides that one setting. + + Args: + image: Container image to run. + calibration_path: Host directory with MARS/VISOR calibration files. + writable_paths: Host directories to mount read-write, besides the + home directory and ``workdir``; replaces the configured ones. + workdir: Where programs run, and what relative paths are relative + to. Defaults to the current directory. + config: A :class:`~tig_cli.config.Config`, or the path of one config + file to load instead of the layered ones. + disable_path_translation: Pass host paths through untranslated. + selinux_label_disable: Run with ``--security-opt label=disable``. + env: Environment variables set for every program run. + runtime: Container runtime to drive, instead of finding one. + + A ``Tig`` holds a claim on its container while open, which keeps + concurrent ``tig`` invocations from removing it; use it as a context + manager or call :meth:`close`. + """ + + def __init__( + self, + image: str | None = None, + *, + calibration_path: str | os.PathLike[str] | None = None, + writable_paths: Iterable[str | os.PathLike[str]] | None = None, + workdir: str | os.PathLike[str] | None = None, + config: Config | str | os.PathLike[str] | None = None, + disable_path_translation: bool | None = None, + selinux_label_disable: bool | None = None, + env: Mapping[str, str] | None = None, + runtime: Runtime | None = None, + ): + try: + if config is None: + config = load_config() + elif not isinstance(config, Config): + config = load_config(Path(config)) + if writable_paths is None: + self.writable_paths = resolve_writable_paths(config) + else: + self.writable_paths = [os.fspath(path) for path in writable_paths] + if disable_path_translation is None: + translate_off = resolve_disable_path_translation(config) + else: + translate_off = disable_path_translation + label_disable = resolve_selinux_label_disable(config, selinux_label_disable) + except ConfigError as e: + raise TigError(str(e)) from e + + self.config = config + self.image = image or get_container_image(config) + self.workdir = str(Path(os.fspath(workdir) if workdir else os.getcwd()).resolve()) + self.env = dict(env or {}) + # Programs write where they run, so the working directory has to be + # writable like the one the CLI is invoked from. + if self.workdir not in self.writable_paths: + self.writable_paths.append(self.workdir) + + self.manager = ContainerManager( + self.image, + config=config, + disable_path_translation=translate_off, + calibration_path=( + os.fspath(calibration_path) + if calibration_path is not None + else get_calibration_path(config) + ), + selinux_label_disable=label_disable, + runtime=runtime, + ) + + def __enter__(self) -> "Tig": + return self + + def __exit__(self, *exc_info) -> bool: + self.close() + return False + + def __repr__(self) -> str: + return f"Tig(image={self.image!r}, workdir={self.workdir!r})" + + def __getattr__(self, name: str): + """``tig.marsxyz(...)`` as a spelling of ``tig.run("marsxyz", ...)``.""" + if name.startswith("_"): + raise AttributeError(name) + + def run_tool(*args: Param, **kwargs: Any) -> Result: + return self.run(name, *args, **kwargs) + + run_tool.__name__ = name + return run_tool + + @property + def container_name(self) -> str: + """The container programs run in, once one has been used.""" + return self.manager.container_name + + def start(self) -> "Tig": + """Make sure the container is running, pulling the image if needed. + + :meth:`run` does this itself; calling it first only moves the + (possibly slow) first-use cost to a moment of the caller's choosing. + """ + self.manager.ensure_container(writable_paths=self.writable_paths) + return self + + def close(self) -> None: + """Release this handle's claim on the container, leaving it running.""" + self.manager.release_claim() + + def shutdown(self) -> int: + """Remove the containers tig has created; returns how many.""" + return self.manager.shutdown() + + def tools(self) -> list[str]: + """The names of the VICAR programs the image provides.""" + self.start() + return self.manager.list_tools() + + def run( + self, + tool: str, + *args: Param, + stdin: str | bytes | None = None, + timeout: float | None = None, + check: bool = False, + workdir: str | os.PathLike[str] | None = None, + env: Mapping[str, str] | None = None, + **params: Param, + ) -> Result: + """Run one VICAR program and wait for it. + + Args: + tool: Program name, as on the ``tig`` command line. + *args: Positional arguments. Paths may be given as ``Path``. + stdin: Text or bytes fed to the program's standard input. + timeout: Seconds to allow; the program is killed after that + and :class:`VicarTimeout` raised. + check: Raise :class:`VicarError` on a non-zero exit status. + Off by default: VICAR programs often exit non-zero on + success, so look at ``stdout`` before relying on it. + workdir: Run in this directory instead of the handle's. It must + be under the home directory or one of ``writable_paths`` + for the program to be able to write there. + env: Environment variables for this run, over the handle's. + **params: VICAR parameters, written ``keyword=value``. A list + or tuple becomes ``keyword=(a,b)``. + + Returns: + The program's :class:`Result`. + """ + arguments = [_text(arg) for arg in args] + [ + f"{keyword}={_value(value)}" for keyword, value in params.items() + ] + host_workdir = str(Path(os.fspath(workdir)).resolve()) if workdir else self.workdir + candidates = _named_paths(arguments, host_workdir) + before = {path: _fingerprint(path) for path in candidates} + + self.start() + ensure_x11_ready() + + manager = self.manager + if manager.disable_path_translation: + command = [tool, *arguments] + container_workdir = host_workdir + else: + command = [tool, *manager.translator.translate_args(arguments)] + container_workdir = manager.translator.get_container_cwd(host_workdir) + + exec_id = uuid.uuid4().hex + environment = { + "DISPLAY": container_display(manager.runtime.name), + **self.env, + **(env or {}), + EXEC_ID_ENV: exec_id, + } + exec_args = ["exec", "-w", container_workdir] + if stdin is not None: + exec_args.append("-i") + for key, value in environment.items(): + exec_args += ["-e", f"{key}={value}"] + exec_args += [manager.container_name, *command] + + payload = stdin.encode() if isinstance(stdin, str) else stdin + try: + completed = subprocess.run( + manager.runtime.args(*exec_args), + input=payload, + stdin=None if stdin is not None else subprocess.DEVNULL, + capture_output=True, + timeout=timeout, + ) + except subprocess.TimeoutExpired as e: + self._kill(exec_id) + raise VicarTimeout( + tool, timeout or 0.0, _decode(e.stdout), _decode(e.stderr) + ) from None + except OSError as e: + raise TigError(f"Failed to run {manager.runtime.name}: {e}") from e + except BaseException: + # Ctrl-C or a cancelled thread: the client is gone, the tool is not. + self._kill(exec_id) + raise + + outputs = [ + path for path in candidates if _fingerprint(path) not in (None, before[path]) + ] + result = Result( + tool, + command, + completed.returncode, + _decode(completed.stdout), + _decode(completed.stderr), + outputs, + ) + return result.check() if check else result + + def _kill(self, exec_id: str) -> None: + """Kill a run's processes in the container after its client is gone.""" + try: + subprocess.run( + self.manager.runtime.args( + "exec", self.manager.container_name, + "sh", "-c", kill_tree_command(exec_id, signal.SIGKILL), + ), + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + timeout=KILL_TIMEOUT, + check=False, + ) + except (OSError, subprocess.SubprocessError): + pass + + +def run(tool: str, *args: Param, **kwargs: Any) -> Result: + """Run one VICAR program with default settings; see :meth:`Tig.run`.""" + with Tig() as tig: + return tig.run(tool, *args, **kwargs) + + +def _text(value: Value) -> str: + if isinstance(value, (str, int, float)): + return str(value) + return os.fspath(value) + + +def _value(value: Param) -> str: + """A keyword's value as VICAR spells it; a sequence is a value list.""" + if isinstance(value, (list, tuple)): + return "(" + ",".join(_text(item) for item in value) + ")" + return _text(value) + + +def _named_paths(arguments: list[str], workdir: str) -> list[Path]: + """Host paths the arguments may name, in order; flags and keywords aren't.""" + paths: list[Path] = [] + for argument in arguments: + if argument.startswith("-"): + continue + value = argument + if "=" in argument and not os.path.isabs(argument): + value = argument.partition("=")[2] + if value.startswith("(") and value.endswith(")"): + items = value[1:-1].split(",") + else: + items = [value] + for item in items: + item = item.strip() + if not item: + continue + path = Path(item) if os.path.isabs(item) else Path(workdir) / item + if path not in paths: + paths.append(path) + return paths + + +def _fingerprint(path: Path) -> tuple[int, int] | None: + """What tells one version of a file from the next; None if no file.""" + try: + info = path.stat() + except OSError: + return None + if not (info.st_mode & 0o170000) == 0o100000: + return None + return (info.st_mtime_ns, info.st_size) + + +def _decode(data: bytes | None) -> str: + return (data or b"").decode("utf-8", errors="replace") diff --git a/tig-cli/tests/integration/test_api.py b/tig-cli/tests/integration/test_api.py new file mode 100644 index 0000000..141834c --- /dev/null +++ b/tig-cli/tests/integration/test_api.py @@ -0,0 +1,88 @@ +"""Integration tests for the Python API against the real image. + +Run with: pytest -m integration +""" +import pytest + +from tig_cli import Tig, VicarError, VicarTimeout + + +@pytest.fixture +def tig(tmp_path): + # tmp_path is outside the home directory, so this exercises the + # read-write mount of the working directory and /host translation. + with Tig(workdir=tmp_path) as tig: + yield tig + + +@pytest.mark.integration +def test_generates_an_image_and_reports_it(tig, tmp_path): + result = tig.run("gen", "test.vic", 32, 32) + + assert result.ok, result.stdout + assert "GEN task completed" in result.stdout + assert result.outputs == [tmp_path / "test.vic"] + assert result.output.stat().st_size > 32 * 32 + + +@pytest.mark.integration +def test_pipelines_files_between_tools(tig, tmp_path): + tig.run("gen", "test.vic", 32, 32, check=True) + listing = tig.run("label", "-list", inp="test.vic", check=True) + png = tig.vicario("test.vic", "test.png").check() + + assert "File test.vic" in listing.stdout + assert listing.outputs == [] + assert png.output == tmp_path / "test.png" + assert png.output.read_bytes().startswith(b"\x89PNG") + + +@pytest.mark.integration +def test_absolute_host_paths_are_translated(tig, tmp_path): + result = tig.run("gen", tmp_path / "abs.vic", 16, 16, check=True) + + assert result.command[1] == f"/host{tmp_path / 'abs.vic'}" + assert result.outputs == [tmp_path / "abs.vic"] + + +@pytest.mark.integration +def test_a_failing_tool_reports_its_status(tig): + result = tig.run("copy", "missing.vic", "out.vic") + + assert not result.ok + assert result.outputs == [] + with pytest.raises(VicarError): + result.check() + + +@pytest.mark.integration +def test_stdin_and_environment_reach_the_tool(tig): + result = tig.run("sh", "-c", 'cat; echo "$GREETING"', + stdin="in\n", env={"GREETING": "hi"}) + + assert result.stdout == "in\nhi\n" + + +@pytest.mark.integration +def test_a_timed_out_tool_is_killed_in_the_container(tig): + with pytest.raises(VicarTimeout): + tig.run("sh", "-c", "echo started; sleep 60", timeout=2) + + survivors = tig.run("sh", "-c", "ps -eo stat,args | grep '[s]leep 60' || true") + assert all(line.startswith("Z") for line in survivors.stdout.splitlines()) + + +@pytest.mark.integration +def test_the_container_is_shared_between_handles_and_runs(tmp_path): + with Tig(workdir=tmp_path) as first: + first.run("gen", "a.vic", 8, 8, check=True) + name = first.container_name + with Tig(workdir=tmp_path) as second: + second.run("gen", "b.vic", 8, 8, check=True) + assert second.container_name == name + + +@pytest.mark.integration +def test_tools_include_the_vicar_programs(tig): + tools = tig.tools() + assert {"gen", "label", "marsxyz"} <= set(tools) diff --git a/tig-cli/tests/test_api.py b/tig-cli/tests/test_api.py new file mode 100644 index 0000000..3bf773b --- /dev/null +++ b/tig-cli/tests/test_api.py @@ -0,0 +1,584 @@ +"""Tests for the Python API, against a runtime answered from memory.""" +import os +import subprocess +from pathlib import Path +from unittest.mock import patch + +import pytest + +import tig_cli +from tig_cli import api +from tig_cli.api import Result, Tig, VicarError, VicarTimeout +from tig_cli.config import Config +from tig_cli.spec import EXEC_ID_ENV, TigError + +from tests.test_container import IMAGE, FakeRuntime + + +class FakeExec: + """Stands in for subprocess.run: records the exec, plays a scripted tool. + + ``script`` is called with the argument list after the tool name and may + write files, so that output tracking sees what a real tool would do. + """ + + def __init__(self, exit_code=0, stdout=b"", stderr=b"", script=None): + self.exit_code = exit_code + self.stdout = stdout + self.stderr = stderr + self.script = script + self.calls = [] + self.kwargs = [] + + def __call__(self, argv, **kwargs): + self.calls.append(list(argv)) + self.kwargs.append(kwargs) + if self.script is not None and argv[1] == "exec": + self.script(argv) + return subprocess.CompletedProcess( + argv, self.exit_code, self.stdout, self.stderr + ) + + @property + def exec(self): + """The one exec of a tool that was issued.""" + execs = [c for c in self.calls if c[1] == "exec" and "-w" in c] + assert len(execs) == 1, f"tools were run {len(execs)} times" + return execs[0] + + def tool_args(self): + """The exec's arguments past the container name.""" + argv = self.exec + container = next(a for a in argv if a.startswith("tig-vicar-")) + return argv[argv.index(container) + 1:] + + def env(self): + argv = self.exec + return dict( + argv[i + 1].split("=", 1) for i, a in enumerate(argv) if a == "-e" + ) + + +@pytest.fixture +def home(tmp_path, monkeypatch): + home = tmp_path / "home" / "user" + home.mkdir(parents=True) + monkeypatch.setenv("HOME", str(home)) + return home + + +@pytest.fixture +def workdir(home): + workdir = home / "work" + workdir.mkdir() + return workdir + + +@pytest.fixture +def no_x11(monkeypatch): + monkeypatch.setattr(api, "ensure_x11_ready", lambda: None) + + +def make_tig(workdir, runtime=None, **kwargs): + return Tig(IMAGE, workdir=workdir, runtime=runtime or FakeRuntime(), **kwargs) + + +def run_with(fake, tig, *args, **kwargs): + with patch.object(api.subprocess, "run", fake): + return tig.run(*args, **kwargs) + + +# --- Result --- + + +def result(exit_code=0, stdout="", stderr="", outputs=()): + return Result("gen", ["gen"], exit_code, stdout, stderr, list(outputs)) + + +def test_result_ok_follows_the_exit_code(): + assert result(0).ok + assert result(0) + assert not result(1).ok + assert not result(1) + + +def test_check_returns_a_passing_result(): + passing = result(0) + assert passing.check() is passing + + +def test_check_raises_on_failure_with_the_last_line_of_output(): + failing = result(255, stdout="Beginning VICAR task GEN\n ** ABEND called **\n") + with pytest.raises(VicarError, match=r"gen exited with status 255: \*\* ABEND") as e: + failing.check() + assert e.value.result is failing + assert isinstance(e.value, TigError) + + +def test_check_prefers_stderr_for_the_detail(): + with pytest.raises(VicarError, match="status 1: boom"): + result(1, stdout="chatter", stderr="boom\n").check() + + +def test_output_is_the_single_output_file(): + assert result(outputs=[Path("/a")]).output == Path("/a") + assert result(outputs=[]).output is None + assert result(outputs=[Path("/a"), Path("/b")]).output is None + + +def test_result_repr_is_compact(): + assert repr(result(outputs=[Path("/a.vic")])) == ( + "Result(tool='gen', exit_code=0, outputs=['/a.vic'])" + ) + + +# --- Tig construction --- + + +def test_defaults_come_from_the_config(home, workdir, monkeypatch): + calibration = home / "calib" + calibration.mkdir() + config = Config( + image="config-image:1", + calibration_path=str(calibration), + writable_paths=["/data"], + disable_path_translation=True, + ) + tig = Tig(workdir=workdir, config=config, runtime=FakeRuntime()) + assert tig.image == "config-image:1" + assert tig.manager.calibration_path == str(calibration) + assert tig.manager.disable_path_translation + assert "/data" in tig.writable_paths + + +def test_arguments_override_the_config(home, workdir): + config = Config(image="config-image:1") + tig = Tig("arg-image:2", workdir=workdir, config=config, runtime=FakeRuntime()) + assert tig.image == "arg-image:2" + + +def test_false_and_empty_overrides_beat_the_config(home, workdir, monkeypatch): + monkeypatch.setenv("TIG_DISABLE_PATH_TRANSLATION", "1") + config = Config(writable_paths=["/data"], disable_path_translation=True) + tig = Tig( + workdir=workdir, config=config, runtime=FakeRuntime(), + writable_paths=[], disable_path_translation=False, + ) + assert not tig.manager.disable_path_translation + assert "/data" not in tig.writable_paths + + +def test_writable_paths_replace_the_configured_ones(home, workdir): + config = Config(writable_paths=["/data"]) + tig = Tig( + workdir=workdir, config=config, runtime=FakeRuntime(), + writable_paths=[Path("/results")], + ) + assert "/results" in tig.writable_paths + assert "/data" not in tig.writable_paths + + +def test_config_may_be_a_file(home, workdir, tmp_path): + path = tmp_path / "tig.toml" + path.write_text('image = "file-image:3"\n') + tig = Tig(workdir=workdir, config=path, runtime=FakeRuntime()) + assert tig.image == "file-image:3" + + +def test_a_bad_config_file_is_user_facing(home, workdir, tmp_path): + path = tmp_path / "tig.toml" + path.write_text("image = [\n") + with pytest.raises(TigError): + Tig(workdir=workdir, config=path, runtime=FakeRuntime()) + + +def test_workdir_defaults_to_the_current_directory(home, workdir, monkeypatch): + monkeypatch.chdir(workdir) + assert Tig(IMAGE, runtime=FakeRuntime()).workdir == str(workdir.resolve()) + + +def test_workdir_is_made_writable(home, tmp_path): + outside = tmp_path / "elsewhere" + outside.mkdir() + tig = make_tig(outside) + assert str(outside.resolve()) in tig.writable_paths + + +def test_the_package_exports_the_api_lazily(): + assert tig_cli.Tig is Tig + assert tig_cli.run is api.run + with pytest.raises(AttributeError): + tig_cli.nonexistent + + +# --- Running a tool --- + + +def test_run_execs_the_tool_in_the_container(home, workdir, no_x11): + tig = make_tig(workdir) + fake = FakeExec(stdout=b"GEN task completed\n") + result = run_with(fake, tig, "gen", "test.vic", 64, 64) + + assert result.ok + assert result.stdout == "GEN task completed\n" + assert result.stderr == "" + assert result.tool == "gen" + assert fake.tool_args() == ["gen", "test.vic", "64", "64"] + assert result.command == ["gen", "test.vic", "64", "64"] + argv = fake.exec + assert argv[:2] == ["/usr/bin/docker", "exec"] + assert argv[argv.index("-w") + 1] == str(workdir.resolve()) + assert tig.container_name in argv + assert tig.container_name in tig.manager.runtime.containers + + +def test_run_starts_the_container_once(home, workdir, no_x11): + runtime = FakeRuntime() + tig = make_tig(workdir, runtime) + fake = FakeExec() + run_with(fake, tig, "gen", "a.vic") + run_with(fake, tig, "gen", "b.vic") + assert runtime.created == [tig.container_name] + + +def test_keywords_become_vicar_parameters(home, workdir, no_x11): + fake = FakeExec() + run_with(fake, make_tig(workdir), "label", "-list", inp="test.vic", nl=10) + assert fake.tool_args() == ["label", "-list", "inp=test.vic", "nl=10"] + + +def test_sequences_become_value_lists(home, workdir, no_x11): + fake = FakeExec() + run_with( + fake, make_tig(workdir), "marsxyz", + inp=[Path("left.vic"), "right.vic"], out=("xyz.vic",), + ) + assert fake.tool_args() == [ + "marsxyz", "inp=(left.vic,right.vic)", "out=(xyz.vic)" + ] + + +def test_host_paths_are_translated(home, workdir, no_x11): + fake = FakeExec() + run_with( + fake, make_tig(workdir), "copy", "/data/in.vic", out=Path("/data/out.vic") + ) + assert fake.tool_args() == [ + "copy", "/host/data/in.vic", "out=/host/data/out.vic" + ] + + +def test_paths_under_home_are_left_alone(home, workdir, no_x11): + fake = FakeExec() + run_with(fake, make_tig(workdir), "copy", workdir / "in.vic") + assert fake.tool_args() == ["copy", str(workdir / "in.vic")] + + +def test_translation_can_be_disabled(home, workdir, no_x11): + fake = FakeExec() + tig = make_tig(workdir, disable_path_translation=True) + run_with(fake, tig, "copy", "/data/in.vic") + assert fake.tool_args() == ["copy", "/data/in.vic"] + + +def test_the_workdir_is_translated_too(home, tmp_path, no_x11): + outside = tmp_path / "elsewhere" + outside.mkdir() + fake = FakeExec() + run_with(fake, make_tig(outside), "gen", "a.vic") + argv = fake.exec + assert argv[argv.index("-w") + 1] == f"/host{outside.resolve()}" + + +def test_run_may_use_another_workdir(home, workdir, no_x11): + other = workdir.parent / "other" + other.mkdir() + fake = FakeExec() + run_with(fake, make_tig(workdir), "gen", "a.vic", workdir=other) + argv = fake.exec + assert argv[argv.index("-w") + 1] == str(other.resolve()) + + +def test_each_run_gets_a_display_and_an_exec_id(home, workdir, no_x11): + fake = FakeExec() + run_with(fake, make_tig(workdir), "gen", "a.vic") + env = fake.env() + assert "DISPLAY" in env + assert len(env[EXEC_ID_ENV]) == 32 + + +def test_environment_layers_handle_then_run(home, workdir, no_x11): + fake = FakeExec() + tig = make_tig(workdir, env={"A": "handle", "B": "handle"}) + run_with(fake, tig, "gen", env={"B": "run", "C": "run"}) + env = fake.env() + assert (env["A"], env["B"], env["C"]) == ("handle", "run", "run") + + +def test_the_exec_id_cannot_be_overridden(home, workdir, no_x11): + fake = FakeExec() + tig = make_tig(workdir, env={EXEC_ID_ENV: "handle"}) + run_with(fake, tig, "gen", env={EXEC_ID_ENV: "run"}) + assert len(fake.env()[EXEC_ID_ENV]) == 32 + + +def test_stdin_is_closed_unless_given(home, workdir, no_x11): + fake = FakeExec() + run_with(fake, make_tig(workdir), "gen") + assert "-i" not in fake.exec + assert fake.kwargs[-1]["stdin"] is subprocess.DEVNULL + assert fake.kwargs[-1]["input"] is None + + +def test_stdin_text_is_fed_to_the_tool(home, workdir, no_x11): + fake = FakeExec() + run_with(fake, make_tig(workdir), "cat", stdin="hello\n") + assert "-i" in fake.exec + assert fake.kwargs[-1]["input"] == b"hello\n" + assert fake.kwargs[-1]["stdin"] is None + + +def test_stdin_bytes_are_fed_as_is(home, workdir, no_x11): + fake = FakeExec() + run_with(fake, make_tig(workdir), "cat", stdin=b"\x00\x01") + assert fake.kwargs[-1]["input"] == b"\x00\x01" + + +def test_output_is_decoded_leniently(home, workdir, no_x11): + fake = FakeExec(stdout=b"ok \xff\n", stderr=b"\xfe") + result = run_with(fake, make_tig(workdir), "gen") + assert result.stdout == "ok \ufffd\n" + assert result.stderr == "\ufffd" + + +def test_a_failing_tool_is_reported_not_raised(home, workdir, no_x11): + fake = FakeExec(exit_code=255, stdout=b" ** ABEND called **\n") + result = run_with(fake, make_tig(workdir), "gen") + assert result.exit_code == 255 + assert not result + + +def test_check_raises_for_a_failing_tool(home, workdir, no_x11): + fake = FakeExec(exit_code=255, stdout=b" ** ABEND called **\n") + with pytest.raises(VicarError, match="gen exited with status 255"): + run_with(fake, make_tig(workdir), "gen", check=True) + + +def test_timeout_is_passed_and_kills_the_tool(home, workdir, no_x11): + tig = make_tig(workdir) + calls = [] + + def fake(argv, **kwargs): + calls.append((list(argv), kwargs)) + if len(calls) == 1: + raise subprocess.TimeoutExpired(argv, 2.5, output=b"start\n") + return subprocess.CompletedProcess(argv, 0, b"", b"") + + with pytest.raises(VicarTimeout, match="sleep did not finish within 2.5") as e: + run_with(fake, tig, "sleep", "30", timeout=2.5) + + assert e.value.stdout == "start\n" + assert calls[0][1]["timeout"] == 2.5 + exec_id = dict( + a.split("=", 1) for a in calls[0][0] if a.startswith(EXEC_ID_ENV) + )[EXEC_ID_ENV] + kill_argv, kill_kwargs = calls[1] + assert kill_argv[:3] == ["/usr/bin/docker", "exec", tig.container_name] + assert kill_argv[3:5] == ["sh", "-c"] + assert exec_id in kill_argv[5] + assert "-9" in kill_argv[5] or "KILL" in kill_argv[5] + assert kill_kwargs["timeout"] == api.KILL_TIMEOUT + + +def test_an_interrupt_terminates_the_tool_and_propagates(home, workdir, no_x11): + tig = make_tig(workdir) + calls = [] + + def fake(argv, **kwargs): + calls.append(list(argv)) + if len(calls) == 1: + raise KeyboardInterrupt + return subprocess.CompletedProcess(argv, 0, b"", b"") + + with pytest.raises(KeyboardInterrupt): + run_with(fake, tig, "marsmap", "in.vic") + + exec_id = dict( + a.split("=", 1) for a in calls[0] if a.startswith(EXEC_ID_ENV) + )[EXEC_ID_ENV] + kill_argv = calls[1] + assert kill_argv[:5] == ["/usr/bin/docker", "exec", tig.container_name, "sh", "-c"] + assert exec_id in kill_argv[5] + assert "kill -9 " in kill_argv[5] + + +def test_a_failed_kill_does_not_hide_the_timeout(home, workdir, no_x11): + def fake(argv, **kwargs): + if kwargs.get("capture_output"): + raise subprocess.TimeoutExpired(argv, 1) + raise OSError("runtime gone") + + with pytest.raises(VicarTimeout): + run_with(fake, make_tig(workdir), "sleep", timeout=1) + + +def test_a_missing_runtime_binary_is_user_facing(home, workdir, no_x11): + def fake(argv, **kwargs): + raise FileNotFoundError(argv[0]) + + with pytest.raises(TigError, match="Failed to run docker"): + run_with(fake, make_tig(workdir), "gen") + + +def test_tools_are_methods(home, workdir, no_x11): + tig = make_tig(workdir) + fake = FakeExec() + with patch.object(api.subprocess, "run", fake): + result = tig.marsxyz("left.vic", out="xyz.vic") + assert isinstance(result, Result) + assert fake.tool_args() == ["marsxyz", "left.vic", "out=xyz.vic"] + assert tig.marsxyz.__name__ == "marsxyz" + + +def test_private_names_are_not_tools(home, workdir): + with pytest.raises(AttributeError): + make_tig(workdir)._marsxyz + + +def test_x11_is_readied_before_each_run(home, workdir, monkeypatch): + readied = [] + monkeypatch.setattr(api, "ensure_x11_ready", lambda: readied.append(True)) + monkeypatch.setattr( + "tig_cli.container.ensure_x11_ready", lambda: None + ) + run_with(FakeExec(), make_tig(workdir), "gen") + assert readied == [True] + + +# --- Output files --- + + +def touch(path, content=b"x"): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(content) + + +def writes(*names): + """A tool that writes the named files relative to the exec's workdir.""" + + def script(argv): + workdir = argv[argv.index("-w") + 1] + for name in names: + touch(Path(workdir) / name) + + return script + + +def test_created_files_named_in_the_arguments_are_outputs(home, workdir, no_x11): + fake = FakeExec(script=writes("test.vic")) + result = run_with(fake, make_tig(workdir), "gen", "test.vic", 64, 64) + assert result.outputs == [workdir / "test.vic"] + assert result.output == workdir / "test.vic" + + +def test_untouched_inputs_are_not_outputs(home, workdir, no_x11): + touch(workdir / "in.vic") + fake = FakeExec(script=writes("out.vic")) + result = run_with(fake, make_tig(workdir), "copy", "in.vic", "out.vic") + assert result.outputs == [workdir / "out.vic"] + + +def test_rewritten_files_are_outputs(home, workdir, no_x11): + touch(workdir / "out.vic", b"old") + os.utime(workdir / "out.vic", ns=(0, 0)) + fake = FakeExec(script=writes("out.vic")) + result = run_with(fake, make_tig(workdir), "gen", "out.vic") + assert result.outputs == [workdir / "out.vic"] + + +def test_keyword_values_and_lists_are_tracked(home, workdir, no_x11): + fake = FakeExec(script=writes("xyz.vic", "b.vic")) + result = run_with( + fake, make_tig(workdir), "marsxyz", inp=["a.vic", "b.vic"], out="xyz.vic" + ) + assert result.outputs == [workdir / "b.vic", workdir / "xyz.vic"] + + +def test_absolute_host_paths_are_tracked_untranslated(home, workdir, no_x11): + target = workdir / "abs.vic" + + def script(argv): + touch(target) + + result = run_with( + FakeExec(script=script), make_tig(workdir), "gen", str(target) + ) + assert result.outputs == [target] + + +def test_outputs_follow_the_runs_workdir(home, workdir, no_x11): + other = workdir.parent / "other" + other.mkdir() + fake = FakeExec(script=writes("a.vic")) + result = run_with(fake, make_tig(workdir), "gen", "a.vic", workdir=other) + assert result.outputs == [other / "a.vic"] + + +def test_flags_and_directories_are_not_outputs(home, workdir, no_x11): + (workdir / "-list").mkdir() + (workdir / "sub").mkdir() + fake = FakeExec() + result = run_with(fake, make_tig(workdir), "label", "-list", "sub", "64") + assert result.outputs == [] + + +def test_a_failed_run_still_reports_what_it_wrote(home, workdir, no_x11): + fake = FakeExec(exit_code=1, script=writes("partial.vic")) + result = run_with(fake, make_tig(workdir), "gen", "partial.vic") + assert result.outputs == [workdir / "partial.vic"] + assert not result.ok + + +# --- Lifecycle --- + + +def test_context_manager_releases_the_claim(home, workdir, no_x11): + with make_tig(workdir) as tig: + run_with(FakeExec(), tig, "gen") + claim = tig.manager.claim + assert claim is not None and claim.exists() + assert not claim.exists() + + +def test_close_leaves_the_container_running(home, workdir, no_x11): + runtime = FakeRuntime() + with make_tig(workdir, runtime) as tig: + run_with(FakeExec(), tig, "gen") + assert tig.container_name in runtime.containers + + +def test_shutdown_removes_the_container(home, workdir, no_x11): + runtime = FakeRuntime() + tig = make_tig(workdir, runtime) + run_with(FakeExec(), tig, "gen") + assert tig.shutdown() == 1 + assert tig.container_name not in runtime.containers + + +def test_tools_lists_what_the_image_provides(home, workdir): + runtime = FakeRuntime() + runtime.exec_output = "gen\nlabel\n" + tig = make_tig(workdir, runtime) + with patch("tig_cli.container.ensure_x11_ready", lambda: None): + tools = tig.tools() + assert tools == ["gen", "label"] + + +def test_module_run_uses_a_fresh_handle(home, workdir, monkeypatch, no_x11): + monkeypatch.chdir(workdir) + monkeypatch.setattr(api.Runtime, "detect", classmethod(lambda cls, config=None: FakeRuntime())) + fake = FakeExec(stdout=b"done\n") + with patch.object(api.subprocess, "run", fake): + result = api.run("gen", "test.vic") + assert result.stdout == "done\n" + assert fake.tool_args() == ["gen", "test.vic"]