From a72efdef88839e52518963b5be4bb3c3c4f70c2e Mon Sep 17 00:00:00 2001 From: Joel Natividad <1980690+jqnatividad@users.noreply.github.com> Date: Thu, 8 Oct 2026 15:39:31 -0400 Subject: [PATCH 1/2] feat(qsv): run qsv through qsv-client QSVCommand now runs qsv through the qsv-client package. Its public surface is unchanged: the same methods, a subprocess.CompletedProcess on success, JobError on failure (or the {"stdout", "stderr"} dict with uses_stdio), and the result itself with check=False. What changes: - A timeout kills qsv's whole process group, so children qsv started cannot outlive it or hold its pipes open and hang the worker. - qsv is asked for structured errors (QSV_ERROR_FORMAT=json); with qsv builds that support them, JobError carries the error kind and message. Older builds print text, which is used as before. - The version check reads `qsv --capabilities`, falling back to `--version` on older builds. - `env=` is merged into the worker's environment instead of replacing it (subprocess.run's env= dropped PATH and everything else). - describegpt's api_key goes in QSV_LLM_APIKEY instead of --api-key: describegpt copies its command line into the attribution it writes, and argv is visible in ps. The "NONE" sentinel, which qsv honors only on the flag, stays on the command line. Any non-zero exit is still a failure, including 255, which qsv-client itself counts as success. Co-Authored-By: Claude Opus 5.5 --- ckanext/datapusher_plus/qsv_utils.py | 129 ++++++++++-------- requirements.txt | 4 + tests/test_ai_suggestions.py | 61 +++++++-- tests/test_qsv_command_client.py | 189 +++++++++++++++++++++++++++ 4 files changed, 311 insertions(+), 72 deletions(-) create mode 100644 tests/test_qsv_command_client.py diff --git a/ckanext/datapusher_plus/qsv_utils.py b/ckanext/datapusher_plus/qsv_utils.py index 8029b32..fd27955 100644 --- a/ckanext/datapusher_plus/qsv_utils.py +++ b/ckanext/datapusher_plus/qsv_utils.py @@ -14,6 +14,9 @@ import ckanext.datapusher_plus.utils as utils from typing import Optional, Dict, Any, List, Union +from qsv_client import Qsv, QsvError, QsvNotFound, QsvTimeout +from qsv_client.errors import error_from_run + from ckanext.datapusher_plus.logging_utils import TRACE logger = logging.getLogger(__name__) @@ -40,6 +43,10 @@ def __init__(self, logger: Optional[logging.Logger] = None) -> None: # Verify qsv binary exists if not Path(self.qsv_bin).is_file(): raise utils.JobError(f"{self.qsv_bin} not found.") + try: + self._qsv = Qsv(binary=self.qsv_bin, timeout=conf.QSV_COMMAND_TIMEOUT) + except QsvNotFound as e: + raise utils.JobError(f"{self.qsv_bin} not found.") from e # Check qsv version try: @@ -58,97 +65,96 @@ def _run_command( timeout: Optional[float] = None, ) -> Union[subprocess.CompletedProcess, Dict[str, Any], str]: """ - Run a qsv command with the given arguments. + Run a qsv command with the given arguments, via ``qsv-client``. + + ``qsv-client`` runs qsv in its own process group, so a timeout kills + qsv *and* any children it started, and it asks qsv for structured + errors (``QSV_ERROR_FORMAT=json``), whose kind and message go into + the ``JobError``. Any non-zero exit code is a failure, as with + ``subprocess.run(check=True)``. Args: args: List of arguments for the qsv command check: Whether to raise an exception if the command fails - capture_output: Whether to capture stdout and stderr + capture_output: Ignored: stdout and stderr are always captured. text: Whether to return output as text - env: Optional environment variables + env: Optional environment variables, added to the worker's own + environment (not replacing it). + uses_stdio: On failure with ``check``, return + ``{"stdout": ..., "stderr": ...}`` instead of raising. timeout: Per-call override (seconds). Defaults to ``conf.QSV_COMMAND_TIMEOUT`` so a hung qsv invocation never pins an RQ worker indefinitely. Returns: - The result of subprocess.run + A ``subprocess.CompletedProcess``. Raises: utils.JobError: If the command fails and check is True, or if the command times out. """ - - args = [self.qsv_bin] + args - - # Convert all args to str to avoid TypeError with Path objects str_args = [str(arg) for arg in args] - + command, rest = (str_args[0], str_args[1:]) if str_args else ("", []) + cmdline = " ".join([str(self.qsv_bin)] + str_args) effective_timeout = timeout if timeout is not None else conf.QSV_COMMAND_TIMEOUT + self.logger.trace(f"Running qsv command: {cmdline}") try: - self.logger.trace(f"Running qsv command: {' '.join(str_args)}") - result = subprocess.run( - str_args, - check=check, - capture_output=capture_output, - text=text, + res = self._qsv.run( + command, + *rest, env=env, timeout=effective_timeout, + check=False, + text=text, ) - return result - except subprocess.TimeoutExpired as e: - error_msg = ( - f"qsv command timed out after {effective_timeout}s: " - f"{' '.join(str_args)}" - ) + except QsvTimeout as e: + error_msg = f"qsv command timed out after {effective_timeout}s: {cmdline}" self.logger.error(error_msg) raise utils.JobError(error_msg) from e - except subprocess.CalledProcessError as e: - if uses_stdio: - # Callers that opt into stdio capture handle their own failure - # logic and just want the raw stdout/stderr back. - return {"stdout": e.stdout, "stderr": e.stderr} - - # Always log AND always raise when check=True — previously the - # raise was nested inside `if e.stderr:`, so failures with empty - # stderr silently returned None/"" and slipped past callers that - # had asked for check=True. Build the message with stderr when we - # have it, fall back to the exception's str otherwise. - error_msg = f"qsv command failed: {e}" - if getattr(e, "stderr", None): - error_msg += f" - {e.stderr}" - self.logger.error(error_msg) - if check: - raise utils.JobError(error_msg) from e - return e.stderr + + stdout = res.stdout if text else res.stdout_bytes + stderr: Union[str, bytes] = res.stderr if text else res.stderr.encode("utf-8") + result = subprocess.CompletedProcess( + args=list(res.args), returncode=res.exit_code, stdout=stdout, stderr=stderr + ) + if res.exit_code == 0 or not check: + return result + if uses_stdio: + # Callers that opt into stdio capture handle their own failure + # logic and just want the raw stdout/stderr back. + return {"stdout": stdout, "stderr": stderr} + + # Always log AND always raise when check=True, even with empty stderr. + err = error_from_run( + exit_code=res.exit_code, stderr=res.stderr, args=res.args, command=command + ) + error_msg = f"qsv command failed: {err}" + if not err.structured and res.stderr.strip(): + error_msg += f" - {res.stderr}" + self.logger.error(error_msg) + raise utils.JobError(error_msg) from err def version(self) -> str: """ Get the qsv version. + Read from ``qsv --capabilities`` on qsv versions that have it, and + from ``qsv --version`` otherwise. + Returns: - The qsv version string + The qsv version string, e.g. ``"24.0.0"`` Raises: utils.JobError: If the version command fails + ValueError: If the version output cannot be parsed """ - result = self._run_command(["--version"]) - version_info = result.stdout.strip() - - if not version_info: + try: + return self._qsv.version + except QsvError as e: raise utils.JobError( - f"We expect qsv version info to be returned. Command: {self.qsv_bin} --version. Response: {version_info}" - ) - - # Extract version number - version_start = version_info.find(" ") - version_end = version_info.find("-") - if version_start > 0 and version_end > version_start: - version = version_info[version_start:version_end].lstrip() - else: - version = version_info - - return version + f"Cannot get qsv version info. Command: {self.qsv_bin} --version. {e}" + ) from e def check_version(self) -> bool: """ @@ -712,7 +718,11 @@ def describegpt( env: Optional environment overrides for the subprocess (e.g. ``{"OPENAI_API_KEY": "..."}`` if the caller has a key it wants to inject out-of-band). - api_key: Optional value for ``--api-key``. Required when + api_key: Optional API key, passed to qsv as ``QSV_LLM_APIKEY`` + rather than ``--api-key``: describegpt copies its command + line into the attribution of what it generates, and argv is + visible in ``ps``. Only the ``"NONE"`` sentinel, which qsv + honors on the flag alone, goes on the command line. Required when the LLM endpoint isn't on ``localhost`` (qsv treats non-localhost base URLs as "not a local LLM" and refuses to start without an API key). For unauthenticated @@ -759,7 +769,10 @@ def describegpt( # is nonsensical for these flags, but matching the documented # contract keeps the wrapper predictable. if api_key is not None: - args.extend(["--api-key", api_key]) + if api_key.upper() == "NONE": + args.extend(["--api-key", api_key]) + else: + env = {**(env or {}), "QSV_LLM_APIKEY": api_key} if base_url is not None: args.extend(["--base-url", base_url]) diff --git a/requirements.txt b/requirements.txt index 0ee1502..1b1354c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -29,3 +29,7 @@ blake3>=1.0.7 # explicitly because DP+ imports ``babel.numbers`` directly now and # shouldn't rely on the upstream pin staying loose. babel>=2.9 +# qsv-client: runs qsv for QSVCommand. Kills the whole process group on +# timeout (qsv and anything it started) and reads qsv's structured errors +# (QSV_ERROR_FORMAT=json) where the binary supports them. No dependencies. +qsv-client>=0.1.0 diff --git a/tests/test_ai_suggestions.py b/tests/test_ai_suggestions.py index 45c0ff0..225f681 100644 --- a/tests/test_ai_suggestions.py +++ b/tests/test_ai_suggestions.py @@ -9,7 +9,7 @@ flow continue. These tests exercise: * ``QSVCommand.describegpt`` argument assembly (no real qsv binary - required — subprocess.run is mocked). + required — qsv-client's process layer is mocked). * ``AISuggestionsStage.should_skip`` honors the config flag. * ``AISuggestionsStage.process`` happy / failure paths. * ``helpers.scheming_get_ai_suggestion`` lookup logic across the @@ -53,6 +53,20 @@ def _completed(stdout: str = "", returncode: int = 0) -> subprocess.CompletedPro ) +# ``QSVCommand`` runs qsv through qsv-client, whose process layer is the +# subprocess seam: its first positional argument is the argv, and it takes +# ``env=`` and ``timeout=`` keywords, just like ``subprocess.run``. +RUN_SYNC = "qsv_client._process.run_sync" + + +def _raw(stdout: str = "", exit_code: int = 0): + from qsv_client._process import RawResult + + return RawResult( + exit_code=exit_code, stdout=stdout.encode(), stderr=b"", timed_out=False + ) + + def test_describegpt_default_flags(qsv_command): """Default invocation emits --description --dictionary --tags --format JSON. @@ -61,7 +75,7 @@ def test_describegpt_default_flags(qsv_command): invocations to fail with ``Unknown flag: '--json'``). Output format is controlled by ``--format ``. """ - with mock.patch("subprocess.run", return_value=_completed("{}")) as run: + with mock.patch(RUN_SYNC, return_value=_raw("{}")) as run: qsv_command.describegpt(input_file="/tmp/sample.csv") args = run.call_args[0][0] @@ -79,7 +93,7 @@ def test_describegpt_default_flags(qsv_command): def test_describegpt_prompt_file_threaded_through(qsv_command): """``prompt_file`` is passed as ``--prompt-file ``.""" - with mock.patch("subprocess.run", return_value=_completed("{}")) as run: + with mock.patch(RUN_SYNC, return_value=_raw("{}")) as run: qsv_command.describegpt( input_file="/tmp/sample.csv", prompt_file="/etc/qsv/describegpt.toml", @@ -93,7 +107,7 @@ def test_describegpt_prompt_file_threaded_through(qsv_command): def test_describegpt_output_file_threaded_through(qsv_command): """``output_file`` is passed as ``--output ``.""" - with mock.patch("subprocess.run", return_value=_completed("{}")) as run: + with mock.patch(RUN_SYNC, return_value=_raw("{}")) as run: qsv_command.describegpt( input_file="/tmp/sample.csv", output_file="/tmp/out.json" ) @@ -106,7 +120,7 @@ def test_describegpt_output_file_threaded_through(qsv_command): def test_describegpt_skips_disabled_flags(qsv_command): """Falsy flags omit their CLI counterpart entirely.""" - with mock.patch("subprocess.run", return_value=_completed("{}")) as run: + with mock.patch(RUN_SYNC, return_value=_raw("{}")) as run: qsv_command.describegpt( input_file="/tmp/sample.csv", description=False, @@ -129,7 +143,7 @@ def test_describegpt_default_timeout_from_config(qsv_command): from ckanext.datapusher_plus import config as conf with mock.patch.object(conf, "DESCRIBEGPT_TIMEOUT_SECONDS", 42), \ - mock.patch("subprocess.run", return_value=_completed("{}")) as run: + mock.patch(RUN_SYNC, return_value=_raw("{}")) as run: qsv_command.describegpt(input_file="/tmp/sample.csv") assert run.call_args.kwargs["timeout"] == 42 @@ -140,7 +154,7 @@ def test_describegpt_explicit_timeout_overrides_config(qsv_command): from ckanext.datapusher_plus import config as conf with mock.patch.object(conf, "DESCRIBEGPT_TIMEOUT_SECONDS", 42), \ - mock.patch("subprocess.run", return_value=_completed("{}")) as run: + mock.patch(RUN_SYNC, return_value=_raw("{}")) as run: qsv_command.describegpt(input_file="/tmp/sample.csv", timeout=7) assert run.call_args.kwargs["timeout"] == 7 @@ -156,7 +170,7 @@ def test_describegpt_api_key_threaded_through(qsv_command): container-host hostnames (``host.docker.internal`` etc.) the documented incantation is ``--api-key NONE``. """ - with mock.patch("subprocess.run", return_value=_completed("{}")) as run: + with mock.patch(RUN_SYNC, return_value=_raw("{}")) as run: qsv_command.describegpt(input_file="/tmp/sample.csv", api_key="NONE") args = run.call_args[0][0] @@ -166,7 +180,7 @@ def test_describegpt_api_key_threaded_through(qsv_command): def test_describegpt_base_url_threaded_through(qsv_command): """``--base-url`` overrides the endpoint without rewriting the prompt-file.""" - with mock.patch("subprocess.run", return_value=_completed("{}")) as run: + with mock.patch(RUN_SYNC, return_value=_raw("{}")) as run: qsv_command.describegpt( input_file="/tmp/sample.csv", base_url="http://host.docker.internal:1234/v1", @@ -182,7 +196,7 @@ def test_describegpt_base_url_threaded_through(qsv_command): def test_describegpt_optional_flags_omitted_by_default(qsv_command): """``api_key`` / ``base_url`` are None by default — let qsv discover.""" - with mock.patch("subprocess.run", return_value=_completed("{}")) as run: + with mock.patch(RUN_SYNC, return_value=_raw("{}")) as run: qsv_command.describegpt(input_file="/tmp/sample.csv") args = run.call_args[0][0] @@ -503,12 +517,31 @@ def test_describegpt_empty_api_key_passes_through(qsv_command): """``api_key=""`` is unusual but the docstring contract is ``None`` ⇒ omit, anything else ⇒ pass through. ``""`` should not be silently dropped.""" - with mock.patch("subprocess.run", return_value=_completed("{}")) as run: + with mock.patch(RUN_SYNC, return_value=_raw("{}")) as run: qsv_command.describegpt(input_file="/tmp/sample.csv", api_key="") - args = run.call_args[0][0] - assert "--api-key" in args - assert args[args.index("--api-key") + 1] == "" + assert run.call_args.kwargs["env"]["QSV_LLM_APIKEY"] == "" + assert "--api-key" not in run.call_args[0][0] + + +def test_describegpt_api_key_goes_in_the_environment_not_argv(qsv_command): + """describegpt copies its command line into the attribution of the + dictionary/description/tags it generates, and argv shows up in ``ps``, + so a real key must travel in ``QSV_LLM_APIKEY``. The worker's own + environment is kept, and caller ``env`` entries are merged in.""" + key = "sk-or-v1-0123456789abcdef" + with mock.patch(RUN_SYNC, return_value=_raw("{}")) as run: + qsv_command.describegpt( + input_file="/tmp/sample.csv", api_key=key, env={"EXTRA": "1"} + ) + + argv = run.call_args[0][0] + env = run.call_args.kwargs["env"] + assert "--api-key" not in argv + assert not any(key in a for a in argv) + assert env["QSV_LLM_APIKEY"] == key + assert env["EXTRA"] == "1" + assert "PATH" in env def test_process_reshape_skips_unknown_envelope_keys(stage_cls, context_factory): diff --git a/tests/test_qsv_command_client.py b/tests/test_qsv_command_client.py new file mode 100644 index 0000000..a563872 --- /dev/null +++ b/tests/test_qsv_command_client.py @@ -0,0 +1,189 @@ +# -*- coding: utf-8 -*- +"""``QSVCommand`` runs qsv through qsv-client. + +These tests run ``QSVCommand`` against a fake ``qsvdp`` (a Python script) and +check the contract callers rely on: a ``subprocess.CompletedProcess`` on +success, ``JobError`` on failure (or the ``{"stdout", "stderr"}`` dict with +``uses_stdio``), the result itself with ``check=False``, and a timeout that +kills qsv's children too. +""" + +import json +import os +import subprocess +import sys +import time +from unittest import mock + +import pytest + +pytestmark = pytest.mark.skipif( + sys.platform == "win32", reason="the fake qsv is a POSIX script" +) + +FAKE_QSV = r''' +import json, os, subprocess, sys, time + +args = sys.argv[1:] +cmd = args[0] if args else "" +if cmd == "--capabilities": + print(json.dumps({"binary": "qsvdp", "version": os.environ.get("FAKE_QSV_VERSION", "99.0.0"), + "features": [], "commands": ["count"], "error_formats": ["text", "json"]})) +elif cmd == "ok": + print("out") + print("progress", file=sys.stderr) +elif cmd == "fail-json": + print("out-before-failure") + print(json.dumps({"error": {"kind": "csv", "level": "error", "message": "ragged row 3", + "exit_code": 1, "command": "fail-json", "qsv_version": "99.0.0"}}), + file=sys.stderr) + sys.exit(1) +elif cmd == "fail-text": + print("plain failure text", file=sys.stderr) + sys.exit(1) +elif cmd == "fail-silent": + sys.exit(1) +elif cmd == "warn": + print("broken pipe", file=sys.stderr) + sys.exit(255) +elif cmd == "env": + print(json.dumps({k: os.environ.get(k) for k in ("FOO", "PATH", "QSV_ERROR_FORMAT")})) +elif cmd == "hang": + child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"]) + with open(args[1], "w") as f: + f.write(str(child.pid)) + time.sleep(60) +''' + + +@pytest.fixture +def fake_qsv(tmp_path): + path = tmp_path / "qsvdp" + path.write_text(f"#!{sys.executable}\n{FAKE_QSV}") + path.chmod(0o755) + return path + + +@pytest.fixture +def qsv(fake_qsv): + pytest.importorskip("ckan") + from ckanext.datapusher_plus import qsv_utils + + with mock.patch.object(qsv_utils.conf, "QSV_BIN", fake_qsv): + yield qsv_utils.QSVCommand(logger=mock.Mock()) + + +def _job_error(): + from ckanext.datapusher_plus import utils + + return utils.JobError + + +def test_constructor_checks_the_version_via_capabilities(qsv): + assert qsv.version() == "99.0.0" + + +def test_constructor_rejects_a_too_old_qsv(fake_qsv, monkeypatch): + pytest.importorskip("ckan") + from ckanext.datapusher_plus import qsv_utils + + monkeypatch.setenv("FAKE_QSV_VERSION", "1.0.0") + with mock.patch.object(qsv_utils.conf, "QSV_BIN", fake_qsv), pytest.raises( + _job_error(), match="At least qsv version" + ): + qsv_utils.QSVCommand(logger=mock.Mock()) + + +def test_success_returns_a_completed_process(qsv, fake_qsv): + result = qsv._run_command(["ok"]) + assert isinstance(result, subprocess.CompletedProcess) + assert result.returncode == 0 + assert result.stdout == "out\n" + assert result.stderr == "progress\n" + assert result.args == [str(fake_qsv), "ok"] + + +def test_failure_raises_job_error_with_the_structured_error(qsv): + with pytest.raises(_job_error()) as exc: + qsv._run_command(["fail-json"]) + msg = str(exc.value) + assert msg.startswith("qsv command failed: ") + assert "csv" in msg and "ragged row 3" in msg and "exit 1" in msg + + +def test_failure_raises_job_error_with_text_stderr(qsv): + with pytest.raises(_job_error(), match="plain failure text"): + qsv._run_command(["fail-text"]) + + +def test_failure_with_empty_stderr_still_raises(qsv): + # an earlier DP+ bug: empty stderr meant check=True failures slipped through + with pytest.raises(_job_error(), match="exited with code 1"): + qsv._run_command(["fail-silent"]) + + +def test_exit_255_is_a_failure(qsv): + # qsv-client counts 255 (qsv's "warning") as success; QSVCommand keeps + # subprocess.run(check=True)'s rule that any non-zero exit is a failure + with pytest.raises(_job_error(), match="broken pipe"): + qsv._run_command(["warn"]) + + +def test_failure_with_uses_stdio_returns_the_raw_output(qsv): + out = qsv._run_command(["fail-json"], uses_stdio=True) + assert out["stdout"] == "out-before-failure\n" + assert '"kind": "csv"' in out["stderr"] + + +def test_failure_with_check_false_returns_the_completed_process(qsv): + result = qsv._run_command(["fail-text"], check=False) + assert isinstance(result, subprocess.CompletedProcess) + assert result.returncode == 1 + assert "plain failure text" in result.stderr + + +def test_env_is_merged_into_the_workers_environment(qsv): + # subprocess.run(env=...) used to REPLACE the environment, dropping PATH + seen = json.loads(qsv._run_command(["env"], env={"FOO": "bar"}).stdout) + assert seen["FOO"] == "bar" + assert seen["PATH"] == os.environ["PATH"] + assert seen["QSV_ERROR_FORMAT"] == "json" + + +def test_text_false_returns_bytes(qsv): + result = qsv._run_command(["ok"], text=False) + assert result.stdout == b"out\n" + assert result.stderr == b"progress\n" + + +def test_timeout_raises_job_error_and_kills_qsvs_children(qsv, tmp_path): + pid_file = tmp_path / "child.pid" + started = time.monotonic() + with pytest.raises(_job_error(), match=r"timed out after 1s"): + qsv._run_command(["hang", str(pid_file)], timeout=1) + assert time.monotonic() - started < 30 + + child = int(pid_file.read_text()) + deadline = time.monotonic() + 10 + while _alive(child): + if time.monotonic() > deadline: + os.kill(child, 9) + pytest.fail("qsv's child process survived the timeout") + time.sleep(0.05) + + +def _alive(pid): + """True while ``pid`` runs. A zombie counts as dead: an orphan is only + reaped if PID 1 reaps, and a container's PID 1 may not.""" + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + stat = f"/proc/{pid}/stat" + if os.path.exists(stat): + with open(stat) as f: + return f.read().rsplit(")", 1)[-1].split()[0] != "Z" + state = subprocess.run( + ["ps", "-o", "stat=", "-p", str(pid)], capture_output=True, text=True + ).stdout.strip() + return bool(state) and not state.startswith("Z") From 8d7ec1e8859901a1ea42ef7c49ebd9deff2131de Mon Sep 17 00:00:00 2001 From: Joel Natividad <1980690+jqnatividad@users.noreply.github.com> Date: Thu, 8 Oct 2026 15:43:31 -0400 Subject: [PATCH 2/2] fix(qsv): keep the full command line in qsv failure messages The old JobError text came from CalledProcessError, which included the whole argv. Keep that: operators reading a failed job's log need to see which file and flags were used, not just the subcommand. Co-Authored-By: Claude Opus 5.5 --- ckanext/datapusher_plus/qsv_utils.py | 2 +- tests/test_qsv_command_client.py | 6 ++++-- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/ckanext/datapusher_plus/qsv_utils.py b/ckanext/datapusher_plus/qsv_utils.py index fd27955..55ef853 100644 --- a/ckanext/datapusher_plus/qsv_utils.py +++ b/ckanext/datapusher_plus/qsv_utils.py @@ -129,7 +129,7 @@ def _run_command( err = error_from_run( exit_code=res.exit_code, stderr=res.stderr, args=res.args, command=command ) - error_msg = f"qsv command failed: {err}" + error_msg = f"qsv command failed: {err} [command: {cmdline}]" if not err.structured and res.stderr.strip(): error_msg += f" - {res.stderr}" self.logger.error(error_msg) diff --git a/tests/test_qsv_command_client.py b/tests/test_qsv_command_client.py index a563872..e675055 100644 --- a/tests/test_qsv_command_client.py +++ b/tests/test_qsv_command_client.py @@ -103,12 +103,14 @@ def test_success_returns_a_completed_process(qsv, fake_qsv): assert result.args == [str(fake_qsv), "ok"] -def test_failure_raises_job_error_with_the_structured_error(qsv): +def test_failure_raises_job_error_with_the_structured_error(qsv, fake_qsv): with pytest.raises(_job_error()) as exc: - qsv._run_command(["fail-json"]) + qsv._run_command(["fail-json", "/data/in.csv", "--flag"]) msg = str(exc.value) assert msg.startswith("qsv command failed: ") assert "csv" in msg and "ragged row 3" in msg and "exit 1" in msg + # operators reading the job log need to see what ran + assert f"[command: {fake_qsv} fail-json /data/in.csv --flag]" in msg def test_failure_raises_job_error_with_text_stderr(qsv):