diff --git a/compat/python-vmaf/core/matlab_feature_extractor.py b/compat/python-vmaf/core/matlab_feature_extractor.py index a46a44f89a..f3e8dba792 100644 --- a/compat/python-vmaf/core/matlab_feature_extractor.py +++ b/compat/python-vmaf/core/matlab_feature_extractor.py @@ -1,6 +1,8 @@ # SPDX-License-Identifier: BSD-2-Clause-Patent import os -import subprocess +import subprocess # nosec B404 -- MATLAB argv invocation and timeout diagnostics. +from collections.abc import Iterator +from contextlib import contextmanager import numpy as np @@ -13,6 +15,40 @@ __copyright__ = "Copyright 2016-2020, Netflix, Inc." __license__ = "BSD+Patent" +# Not measured: MATLAB is not installed on the measuring host. One call runs a +# whole-video MATLAB feature script, so the bound leaves room for long clips. +SCRIPT_TIMEOUT_S = 3600 + + +def _matlab_output(output: str | bytes | None) -> str: + """Decode captured diagnostics without displaying a bytes repr or None.""" + return output.decode("utf-8", errors="replace") if isinstance(output, bytes) else output or "" + + +def _matlab_char(value: str) -> str: + """Quote `value` as a MATLAB character vector for a ``-r`` script. + + A single quote ends a MATLAB character vector; two single quotes stand for + one quote inside it (MathWorks, "Text in String and Character Arrays"). + The argv call keeps asset paths away from a shell, but MATLAB parses the + script itself, so an unquoted path holding a quote could end the literal + and add statements. A value with a control character is refused, since a + newline would split the one-line script. + """ + if not value.isprintable(): + raise ValueError(f"MATLAB script argument holds a control character: {value!r}") + return "'" + value.replace("'", "''") + "'" + + +@contextmanager +def _matlab_workspace(workspace: str) -> Iterator[None]: + """Restore the caller's working directory even when MATLAB fails or times out.""" + previous = os.getcwd() + try: + os.chdir(workspace) + yield + finally: + os.chdir(previous) def _run_matlab(matlab_bin: str, matlab_script: str, log_file_path: str | None = None) -> None: @@ -26,7 +62,9 @@ def _run_matlab(matlab_bin: str, matlab_script: str, log_file_path: str | None = ``>> {log_file_path}`` redirect semantics without invoking a shell. Stderr is folded into stdout so the captured/redirected log - matches the prior behaviour for MATLAB error diagnostics. + matches the prior behaviour for MATLAB error diagnostics. A failure + or timeout raises AssertionError chained to the subprocess error, + with the captured output decoded as text. """ argv: list[str] = [ matlab_bin, @@ -45,19 +83,32 @@ def _run_matlab(matlab_bin: str, matlab_script: str, log_file_path: str | None = env["LANG"] = "C" try: if log_file_path is None: - subprocess.check_output(argv, stderr=subprocess.STDOUT, env=env, shell=False) + subprocess.check_output( # noqa: S603 # nosec B603 -- configured MATLAB executable and fixed argv, no shell. + argv, + stderr=subprocess.STDOUT, + env=env, + shell=False, + timeout=SCRIPT_TIMEOUT_S, + ) else: with open(log_file_path, "ab") as log_fh: - subprocess.run( + subprocess.run( # noqa: S603 # nosec B603 -- configured MATLAB executable and fixed argv, no shell. argv, stdout=log_fh, stderr=subprocess.STDOUT, env=env, shell=False, check=True, + timeout=SCRIPT_TIMEOUT_S, ) except subprocess.CalledProcessError as e: - raise AssertionError(f"Process returned {e.returncode}, cmd: {argv}, msg: {str(e.output)}") + raise AssertionError( + f"Process returned {e.returncode}, cmd: {argv}, msg: {_matlab_output(e.output)}" + ) from e + except subprocess.TimeoutExpired as e: + raise AssertionError( + f"Process timed out after {SCRIPT_TIMEOUT_S}s (SCRIPT_TIMEOUT_S), cmd: {argv}, msg: {_matlab_output(e.output)}" + ) from e class MatlabFeatureExtractor(FeatureExtractor): @@ -96,6 +147,7 @@ def _assert_an_asset(cls, asset): ), "STRRED feature extractor only supports yuv420p for now." def _generate_result(self, asset): + """Generate feature scores and restore cwd after MATLAB exits or fails.""" # routine to call the command-line executable and generate quality # scores in the log file. @@ -112,9 +164,9 @@ def _generate_result(self, asset): quality_width, quality_height = asset.quality_width_height matlab_bin = VmafExternalConfig.get_and_assert_matlab() - matlab_script = "run_strred('{ref}', '{dis}', {h}, {w}); exit;".format( - ref=ref_procfile_path, - dis=dis_procfile_path, + matlab_script = "run_strred({ref}, {dis}, {h}, {w}); exit;".format( + ref=_matlab_char(ref_procfile_path), + dis=_matlab_char(dis_procfile_path), w=quality_width, h=quality_height, ) @@ -126,12 +178,11 @@ def _generate_result(self, asset): log_file_path, ) - os.chdir(self.MATLAB_WORKSPACE) - # Argv-list invocation (no shell) — fixes CWE-78 shell injection - # via asset paths. Log redirect is performed by `_run_matlab` - # opening `log_file_path` in append mode. - _run_matlab(matlab_bin, matlab_script, log_file_path=log_file_path) - os.chdir(current_dir) + with _matlab_workspace(self.MATLAB_WORKSPACE): + # Argv-list invocation (no shell) — fixes CWE-78 shell injection + # via asset paths. Log redirect is performed by `_run_matlab` + # opening `log_file_path` in append mode. + _run_matlab(matlab_bin, matlab_script, log_file_path=log_file_path) @classmethod @override(Executor) @@ -199,6 +250,7 @@ def _assert_an_asset(cls, asset): ), "STRRED opt feature extractor only supports yuv420p for now." def _generate_result(self, asset): + """Generate feature scores and restore cwd after MATLAB exits or fails.""" # routine to call the command-line executable and generate quality # scores in the log file. @@ -215,9 +267,9 @@ def _generate_result(self, asset): quality_width, quality_height = asset.quality_width_height matlab_bin = VmafExternalConfig.get_and_assert_matlab() - matlab_script = "run_strred_opt('{ref}', '{dis}', {w}, {h}); exit;".format( - ref=ref_procfile_path, - dis=dis_procfile_path, + matlab_script = "run_strred_opt({ref}, {dis}, {w}, {h}); exit;".format( + ref=_matlab_char(ref_procfile_path), + dis=_matlab_char(dis_procfile_path), w=quality_width, h=quality_height, ) @@ -230,10 +282,9 @@ def _generate_result(self, asset): log_file_path, ) - os.chdir(self.MATLAB_WORKSPACE) - # Argv-list invocation (no shell) — fixes CWE-78 shell injection. - _run_matlab(matlab_bin, matlab_script, log_file_path=log_file_path) - os.chdir(current_dir) + with _matlab_workspace(self.MATLAB_WORKSPACE): + # Argv-list invocation (no shell) — fixes CWE-78 shell injection. + _run_matlab(matlab_bin, matlab_script, log_file_path=log_file_path) @classmethod @override(Executor) @@ -301,6 +352,7 @@ class SpEEDMatlabFeatureExtractor(MatlabFeatureExtractor): MATLAB_WORKSPACE = VmafConfig.root_path("python", "vmaf", "matlab", "SpEED") def _generate_result(self, asset): + """Generate feature scores and restore cwd after MATLAB exits or fails.""" # routine to call the command-line executable and generate quality # scores in the log file. @@ -313,15 +365,13 @@ def _generate_result(self, asset): log_file_path = make_absolute_path(log_file_path, current_dir) quality_width, quality_height = asset.quality_width_height matlab_bin = VmafExternalConfig.get_and_assert_matlab() - matlab_script = ( - "run_speed('{ref}', '{dis}', {w}, {h}, {bands}, '{yuv_type}'); exit;".format( - ref=ref_procfile_path, - dis=dis_procfile_path, - w=quality_width, - h=quality_height, - bands=self.scale_list, - yuv_type=self._get_workfile_yuv_type(asset), - ) + matlab_script = "run_speed({ref}, {dis}, {w}, {h}, {bands}, {yuv_type}); exit;".format( + ref=_matlab_char(ref_procfile_path), + dis=_matlab_char(dis_procfile_path), + w=quality_width, + h=quality_height, + bands=self.scale_list, + yuv_type=_matlab_char(self._get_workfile_yuv_type(asset)), ) if self.logger: self.logger.info( @@ -330,10 +380,9 @@ def _generate_result(self, asset): matlab_script, log_file_path, ) - os.chdir(self.MATLAB_WORKSPACE) - # Argv-list invocation (no shell) — fixes CWE-78 shell injection. - _run_matlab(matlab_bin, matlab_script, log_file_path=log_file_path) - os.chdir(current_dir) + with _matlab_workspace(self.MATLAB_WORKSPACE): + # Argv-list invocation (no shell) — fixes CWE-78 shell injection. + _run_matlab(matlab_bin, matlab_script, log_file_path=log_file_path) @classmethod @override(Executor) @@ -386,13 +435,13 @@ class STMADFeatureExtractor(MatlabFeatureExtractor): # compile necessary functions; need to use mex from within matlab def _custom_init(self): + """Compile MATLAB helpers with exception-safe workspace restoration.""" def run_stmad_mex(matlab_bin: str, mex_script: str) -> None: - current_dir = os.getcwd() + "/" - os.chdir(self.MATLAB_WORKSPACE) - # Argv-list invocation (no shell) — fixes CWE-78 shell injection. - _run_matlab(matlab_bin, mex_script) - os.chdir(current_dir) + """Compile one MEX helper while preserving the caller workspace.""" + with _matlab_workspace(self.MATLAB_WORKSPACE): + # Argv-list invocation (no shell) — fixes CWE-78 shell injection. + _run_matlab(matlab_bin, mex_script) matlab_bin = VmafExternalConfig.get_and_assert_matlab() run_stmad_mex(matlab_bin, "mex ical_std.c; exit;") @@ -406,6 +455,7 @@ def _assert_an_asset(cls, asset): ), "STMAD feature extractor only supports yuv420p for now." def _generate_result(self, asset): + """Generate feature scores and restore cwd after MATLAB exits or fails.""" # routine to call the command-line executable and generate quality # scores in the log file. @@ -422,9 +472,9 @@ def _generate_result(self, asset): quality_width, quality_height = asset.quality_width_height matlab_bin = VmafExternalConfig.get_and_assert_matlab() - matlab_script = "run_stmad('{ref}', '{dis}', {w}, {h}); exit;".format( - ref=ref_procfile_path, - dis=dis_procfile_path, + matlab_script = "run_stmad({ref}, {dis}, {w}, {h}); exit;".format( + ref=_matlab_char(ref_procfile_path), + dis=_matlab_char(dis_procfile_path), w=quality_width, h=quality_height, ) @@ -437,10 +487,9 @@ def _generate_result(self, asset): log_file_path, ) - os.chdir(self.MATLAB_WORKSPACE) - # Argv-list invocation (no shell) — fixes CWE-78 shell injection. - _run_matlab(matlab_bin, matlab_script, log_file_path=log_file_path) - os.chdir(current_dir) + with _matlab_workspace(self.MATLAB_WORKSPACE): + # Argv-list invocation (no shell) — fixes CWE-78 shell injection. + _run_matlab(matlab_bin, matlab_script, log_file_path=log_file_path) @classmethod @override(Executor) @@ -494,6 +543,7 @@ def _assert_an_asset(cls, asset): assert asset.ref_yuv_type == asset.dis_yuv_type def _generate_result(self, asset): + """Generate feature scores and restore cwd after MATLAB exits or fails.""" # routine to call the command-line executable and generate quality # scores in the log file. @@ -510,12 +560,12 @@ def _generate_result(self, asset): quality_width, quality_height = asset.quality_width_height matlab_bin = VmafExternalConfig.get_and_assert_matlab() - matlab_script = "run_icid('{ref}', '{dis}', {h}, {w}, '{yuvtype}'); exit;".format( - ref=ref_workfile_path, - dis=dis_workfile_path, + matlab_script = "run_icid({ref}, {dis}, {h}, {w}, {yuvtype}); exit;".format( + ref=_matlab_char(ref_workfile_path), + dis=_matlab_char(dis_workfile_path), w=quality_width, h=quality_height, - yuvtype=asset.ref_yuv_type, + yuvtype=_matlab_char(asset.ref_yuv_type), ) if self.logger: self.logger.info( @@ -525,10 +575,9 @@ def _generate_result(self, asset): log_file_path, ) - os.chdir(self.MATLAB_WORKSPACE) - # Argv-list invocation (no shell) — fixes CWE-78 shell injection. - _run_matlab(matlab_bin, matlab_script, log_file_path=log_file_path) - os.chdir(current_dir) + with _matlab_workspace(self.MATLAB_WORKSPACE): + # Argv-list invocation (no shell) — fixes CWE-78 shell injection. + _run_matlab(matlab_bin, matlab_script, log_file_path=log_file_path) @classmethod def _post_process_result(cls, result): diff --git a/compat/python-vmaf/tests/test_matlab_run_timeout.py b/compat/python-vmaf/tests/test_matlab_run_timeout.py new file mode 100644 index 0000000000..f385a2431f --- /dev/null +++ b/compat/python-vmaf/tests/test_matlab_run_timeout.py @@ -0,0 +1,193 @@ +# Copyright 2026 Lusoris +# SPDX-License-Identifier: EUPL-1.2 +# +# Unit tests for _run_matlab timeout and error handling in +# compat/python-vmaf/core/matlab_feature_extractor.py. +"""Pytest cases for _run_matlab timeout and process error handling.""" + +from __future__ import annotations + +import subprocess # nosec B404 -- construct subprocess results and exceptions for test doubles. +from pathlib import Path +from typing import Any +from types import SimpleNamespace +from unittest import mock + +import pytest + +from vmaf.core import matlab_feature_extractor +from vmaf.core.matlab_feature_extractor import SCRIPT_TIMEOUT_S, _run_matlab + + +def test_run_matlab_positive_no_log_records_timeout(monkeypatch: pytest.MonkeyPatch) -> None: + """Verify check_output receives the configured timeout when no log path is provided.""" + recorded_kwargs: dict[str, Any] = {} + + def fake_check_output(*args: Any, **kwargs: Any) -> bytes: + """Record check_output kwargs and return empty bytes.""" + recorded_kwargs.update(kwargs) + return b"" + + monkeypatch.setattr(matlab_feature_extractor.subprocess, "check_output", fake_check_output) + _run_matlab("matlab", "test_script") + assert recorded_kwargs.get("timeout") == SCRIPT_TIMEOUT_S + + +def test_run_matlab_positive_with_log_records_timeout( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Verify subprocess.run receives the configured timeout when a log path is provided.""" + recorded_kwargs: dict[str, Any] = {} + + def fake_run(*args: Any, **kwargs: Any) -> subprocess.CompletedProcess[bytes]: + """Record run kwargs and return a successful CompletedProcess.""" + recorded_kwargs.update(kwargs) + return subprocess.CompletedProcess(args[0] if args else [], 0, stdout=b"", stderr=b"") + + monkeypatch.setattr(matlab_feature_extractor.subprocess, "run", fake_run) + _run_matlab("matlab", "test_script", log_file_path=tmp_path / "log") + assert recorded_kwargs.get("timeout") == SCRIPT_TIMEOUT_S + + +def test_run_matlab_negative_called_process_error(monkeypatch: pytest.MonkeyPatch) -> None: + """Verify CalledProcessError converts to AssertionError matching process returncode.""" + argv = ["matlab", "-nodisplay", "-nosplash", "-nodesktop", "-r", "test_script"] + + def fake_check_output(*args: Any, **kwargs: Any) -> bytes: + """Simulate check_output failing with a CalledProcessError.""" + raise subprocess.CalledProcessError(3, argv, output=b"x") + + monkeypatch.setattr(matlab_feature_extractor.subprocess, "check_output", fake_check_output) + with pytest.raises(AssertionError, match=r"Process returned 3") as exc_info: + _run_matlab("matlab", "test_script") + assert isinstance(exc_info.value.__cause__, subprocess.CalledProcessError) + assert "msg: x" in str(exc_info.value) + assert "b'x'" not in str(exc_info.value) + + +def test_run_matlab_negative_called_process_error_with_log( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Verify CalledProcessError with log path converts to AssertionError matching returncode.""" + argv = ["matlab", "-nodisplay", "-nosplash", "-nodesktop", "-r", "test_script"] + + def fake_run(*args: Any, **kwargs: Any) -> subprocess.CompletedProcess[bytes]: + """Simulate run failing with a CalledProcessError.""" + raise subprocess.CalledProcessError(3, argv, output=b"x") + + monkeypatch.setattr(matlab_feature_extractor.subprocess, "run", fake_run) + with pytest.raises(AssertionError, match=r"Process returned 3") as exc_info: + _run_matlab("matlab", "test_script", log_file_path=tmp_path / "log") + assert isinstance(exc_info.value.__cause__, subprocess.CalledProcessError) + assert "msg: x" in str(exc_info.value) + + +def test_run_matlab_boundary_timeout_no_log(monkeypatch: pytest.MonkeyPatch) -> None: + """Verify TimeoutExpired without log file converts to AssertionError with timeout details.""" + argv = ["matlab", "-nodisplay", "-nosplash", "-nodesktop", "-r", "test_script"] + + def fake_check_output(*args: Any, **kwargs: Any) -> bytes: + """Simulate check_output raising a TimeoutExpired exception.""" + raise subprocess.TimeoutExpired(argv, SCRIPT_TIMEOUT_S, output=b"partial") + + monkeypatch.setattr(matlab_feature_extractor.subprocess, "check_output", fake_check_output) + with pytest.raises(AssertionError) as exc_info: + _run_matlab("matlab", "test_script") + msg = str(exc_info.value) + assert "timed out after" in msg + assert str(SCRIPT_TIMEOUT_S) in msg + assert "SCRIPT_TIMEOUT_S" in msg + assert isinstance(exc_info.value.__cause__, subprocess.TimeoutExpired) + assert "msg: partial" in msg + assert "b'partial'" not in msg + + +def test_run_matlab_boundary_timeout_with_log( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Verify TimeoutExpired with log file converts to AssertionError with timeout details.""" + argv = ["matlab", "-nodisplay", "-nosplash", "-nodesktop", "-r", "test_script"] + + def fake_run(*args: Any, **kwargs: Any) -> subprocess.CompletedProcess[bytes]: + """Simulate run raising a TimeoutExpired exception.""" + raise subprocess.TimeoutExpired(argv, SCRIPT_TIMEOUT_S, output=b"partial") + + monkeypatch.setattr(matlab_feature_extractor.subprocess, "run", fake_run) + with pytest.raises(AssertionError) as exc_info: + _run_matlab("matlab", "test_script", log_file_path=tmp_path / "log") + msg = str(exc_info.value) + assert "timed out after" in msg + assert str(SCRIPT_TIMEOUT_S) in msg + assert "SCRIPT_TIMEOUT_S" in msg + assert isinstance(exc_info.value.__cause__, subprocess.TimeoutExpired) + assert "msg: partial" in msg + assert "b'partial'" not in msg + + +@pytest.mark.parametrize("output, expected", [(b"text\xff", "text�"), (None, ""), ("text", "text")]) +def test_matlab_error_output_is_readable(output: str | bytes | None, expected: str) -> None: + """Handle invalid bytes, absent output and decoded strings consistently.""" + assert matlab_feature_extractor._matlab_output(output) == expected + + +@pytest.mark.parametrize( + "extractor_type", + [ + matlab_feature_extractor.StrredFeatureExtractor, + matlab_feature_extractor.StrredOptFeatureExtractor, + matlab_feature_extractor.SpEEDMatlabFeatureExtractor, + matlab_feature_extractor.STMADFeatureExtractor, + matlab_feature_extractor.iCIDFeatureExtractor, + ], +) +def test_generate_result_restores_cwd_on_timeout(tmp_path: Path, extractor_type: type) -> None: + """Every MATLAB extractor must restore cwd after a timeout, with chained diagnostics.""" + previous = Path.cwd() + asset = SimpleNamespace( + ref_procfile_path="ref", + dis_procfile_path="dis", + ref_workfile_path="ref", + dis_workfile_path="dis", + quality_width_height=(576, 324), + ref_yuv_type="yuv420p", + ) + extractor = SimpleNamespace( + MATLAB_WORKSPACE=str(tmp_path), + logger=None, + scale_list=[1], + _get_log_file_path=mock.Mock(return_value=str(tmp_path / "log")), + _get_workfile_yuv_type=mock.Mock(return_value="yuv420p"), + ) + timeout = subprocess.TimeoutExpired(["matlab"], SCRIPT_TIMEOUT_S, output=b"partial\xff") + with ( + mock.patch.object( + matlab_feature_extractor.VmafExternalConfig, + "get_and_assert_matlab", + return_value="matlab", + ), + mock.patch.object(matlab_feature_extractor.subprocess, "run", side_effect=timeout), + ): + with pytest.raises(AssertionError, match="msg: partial�") as exc_info: + extractor_type._generate_result(extractor, asset) + assert exc_info.value.__cause__ is timeout + assert Path.cwd() == previous + + +def test_mex_initialization_restores_cwd_on_timeout(tmp_path: Path) -> None: + """MEX initialization shares exception-safe cwd restoration with feature runs.""" + previous = Path.cwd() + timeout = subprocess.TimeoutExpired(["matlab"], SCRIPT_TIMEOUT_S) + with ( + mock.patch.object( + matlab_feature_extractor.VmafExternalConfig, + "get_and_assert_matlab", + return_value="matlab", + ), + mock.patch.object(matlab_feature_extractor.subprocess, "check_output", side_effect=timeout), + ): + with pytest.raises(AssertionError) as exc_info: + matlab_feature_extractor.STMADFeatureExtractor._custom_init( + SimpleNamespace(MATLAB_WORKSPACE=str(tmp_path)) + ) + assert exc_info.value.__cause__ is timeout + assert Path.cwd() == previous diff --git a/compat/python-vmaf/tests/test_matlab_script_literal.py b/compat/python-vmaf/tests/test_matlab_script_literal.py new file mode 100644 index 0000000000..c62ffc86ac --- /dev/null +++ b/compat/python-vmaf/tests/test_matlab_script_literal.py @@ -0,0 +1,91 @@ +# Copyright 2026 Lusoris +# SPDX-License-Identifier: EUPL-1.2 +"""Asset paths reach MATLAB's ``-r`` script only as quoted character vectors.""" + +from __future__ import annotations + +import re +from pathlib import Path +from types import SimpleNamespace +from unittest import mock + +import pytest + +from vmaf.core import matlab_feature_extractor +from vmaf.core.matlab_feature_extractor import _matlab_char + +# One MATLAB character vector: quote, then any run of non-quotes or doubled quotes, then quote. +CHAR_VECTOR = re.compile(r"'(?:[^']|'')*'") +# What a builder's script must be once each quoted vector is replaced by `S`: +# one call with literal and numeric arguments only, then exit. +ONE_CALL = re.compile(r"^run_[a-z_]+\((?:S|\d+|\[[\d, ]*\])(?:, (?:S|\d+|\[[\d, ]*\]))*\); exit;$") +PAYLOAD = "/clips/x'); system('touch pwned'); %" + +EXTRACTORS = [ + matlab_feature_extractor.StrredFeatureExtractor, + matlab_feature_extractor.StrredOptFeatureExtractor, + matlab_feature_extractor.SpEEDMatlabFeatureExtractor, + matlab_feature_extractor.STMADFeatureExtractor, + matlab_feature_extractor.iCIDFeatureExtractor, +] + + +def test_plain_path_is_quoted_unchanged() -> None: + """Positive: a path without quotes is wrapped in one pair of single quotes.""" + assert _matlab_char("/clips/ref_576x324.yuv") == "'/clips/ref_576x324.yuv'" + + +def test_quote_is_doubled_and_cannot_end_the_vector() -> None: + """Negative: a quote in the value is doubled, so the payload stays one vector.""" + quoted = _matlab_char(PAYLOAD) + assert quoted == "'/clips/x''); system(''touch pwned''); %'" + assert CHAR_VECTOR.fullmatch(quoted) + + +@pytest.mark.parametrize("value", ["/clips/a\nsystem('x')", "/clips/a\rb", "/clips/a\x00b", "a\tb"]) +def test_control_character_is_refused(value: str) -> None: + """Negative: a newline or other control character would split the one-line script.""" + with pytest.raises(ValueError, match="control character"): + _matlab_char(value) + + +@pytest.mark.parametrize( + "value, expected", + [("", "''"), ("'", "''''"), ("''", "''''''"), ("/clips/é ü.yuv", "'/clips/é ü.yuv'")], +) +def test_boundary_values(value: str, expected: str) -> None: + """Boundary: empty value, quotes only, and printable non-ASCII text.""" + assert _matlab_char(value) == expected + assert CHAR_VECTOR.fullmatch(expected) + + +@pytest.mark.parametrize("extractor_type", EXTRACTORS) +def test_every_builder_quotes_the_asset_paths(tmp_path: Path, extractor_type: type) -> None: + """Every MATLAB extractor hands `_run_matlab` one call, whatever the asset path holds.""" + asset = SimpleNamespace( + ref_procfile_path=PAYLOAD, + dis_procfile_path=PAYLOAD, + ref_workfile_path=PAYLOAD, + dis_workfile_path=PAYLOAD, + quality_width_height=(576, 324), + ref_yuv_type="yuv420p", + ) + extractor = SimpleNamespace( + MATLAB_WORKSPACE=str(tmp_path), + logger=None, + scale_list=[1], + _get_log_file_path=mock.Mock(return_value=str(tmp_path / "log")), + _get_workfile_yuv_type=mock.Mock(return_value="yuv420p"), + ) + with ( + mock.patch.object( + matlab_feature_extractor.VmafExternalConfig, + "get_and_assert_matlab", + return_value="matlab", + ), + mock.patch.object(matlab_feature_extractor, "_run_matlab") as run_matlab, + ): + extractor_type._generate_result(extractor, asset) + script = run_matlab.call_args.args[1] + assert _matlab_char(PAYLOAD) in script + assert ONE_CALL.fullmatch(CHAR_VECTOR.sub("S", script)), script diff --git a/compat/python-vmaf/tests/test_misc_extended.py b/compat/python-vmaf/tests/test_misc_extended.py index 7fac498777..9d5976e04c 100644 --- a/compat/python-vmaf/tests/test_misc_extended.py +++ b/compat/python-vmaf/tests/test_misc_extended.py @@ -11,10 +11,12 @@ from __future__ import annotations import os +import subprocess # nosec B404 -- construct subprocess results and exceptions for test doubles. import pytest from vmaf.tools.misc import ( + SUBPROCESS_TIMEOUT_S, check_program_exist, check_scanf_match, cmd_option_exists, @@ -131,6 +133,20 @@ def test_existing_program_ls(self): def test_existing_program_pwd(self): assert check_program_exist("pwd") is True + def test_timeout_means_program_exists(self, monkeypatch): + """A timeout indicates the program started, so it exists.""" + recorded_kwargs = {} + + def stub_call(*args, **kwargs): + """Record keyword arguments and raise TimeoutExpired.""" + recorded_kwargs.update(kwargs) + raise subprocess.TimeoutExpired(cmd="x", timeout=1) + + monkeypatch.setattr("vmaf.tools.misc.subprocess.call", stub_call) + assert check_program_exist("ls") is True + assert recorded_kwargs.get("timeout") == SUBPROCESS_TIMEOUT_S + assert recorded_kwargs.get("stdin") == subprocess.DEVNULL + # --------------------------------------------------------------------------- # check_scanf_match diff --git a/compat/python-vmaf/tools/misc.py b/compat/python-vmaf/tools/misc.py index a5e0b54dfa..48b82aee90 100644 --- a/compat/python-vmaf/tools/misc.py +++ b/compat/python-vmaf/tools/misc.py @@ -4,7 +4,7 @@ import multiprocessing import os import re -import subprocess +import subprocess # nosec B404 -- bounded argv existence check with closed stdin, no shell. import sys import tempfile import unittest @@ -25,6 +25,9 @@ __copyright__ = "Copyright 2016-2020, Netflix, Inc." __license__ = "BSD+Patent" +# Bound = max(60, 10x the slowest measured run): check_program_exist("ls") took +# 0.002 s. The check only needs the program to start. +SUBPROCESS_TIMEOUT_S = 60 def get_stdout_logger(): @@ -411,7 +414,7 @@ def run_group(group): def check_program_exist(program): - """ + """Return whether the program starts, closing stdin and bounding the wait. >>> check_program_exist("xxxafasd34df") False @@ -427,7 +430,15 @@ def check_program_exist(program): """ try: with open(os.devnull, "wb") as devnull_fd: - subprocess.call(program.split(), stdout=devnull_fd) + subprocess.call( # noqa: S603 # nosec B603 -- split argv without a shell; existence check only. + program.split(), + stdin=subprocess.DEVNULL, + stdout=devnull_fd, + timeout=SUBPROCESS_TIMEOUT_S, + ) + return True + except subprocess.TimeoutExpired: + # The program started, so it exists; the timeout only stopped a long run. return True except OSError as e: if e.errno == errno.ENOENT: diff --git a/core/src/sycl/run_captured.py b/core/src/sycl/run_captured.py index 7284126fac..b38a8e985e 100644 --- a/core/src/sycl/run_captured.py +++ b/core/src/sycl/run_captured.py @@ -11,26 +11,72 @@ kept for a failed build and dropped for a successful one. Usage: run_captured.py -- COMMAND [ARGS...] +Exit status: 0 for success; the command's status on ordinary failure; 2 for usage; +124 for timeout; 125 for captured-output limit; 126 for validation or launch failure. """ from __future__ import annotations -import subprocess import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[3])) + +from scripts.lib.safe_subprocess import ( + CommandOutputLimitExceeded, + CommandTimedOut, + CommandValidationError, +) +from scripts.lib.safe_subprocess import run as run_command + +# Bound = 10x the slowest measured run, rounded up: the scratch_check.cpp AOT +# compile for 19 device targets took 10.72 s. +CC_TIMEOUT_S = 120 +# Budget: 1 MiB of diagnostics per AOT target plus 1 MiB for the driver (19 targets). +# This 20 MiB ceiling allows verbose failures without retaining an unbounded compiler log. +MAX_COMPILE_OUTPUT_BYTES = 20 * 1_048_576 + + +def _print_retained_stdout(output: str | bytes | None) -> None: + """Print diagnostic output retained before a timeout or output overflow.""" + if output: + sys.stdout.write(output if isinstance(output, str) else output.decode("utf-8", "replace")) def main(argv: list[str]) -> int: + """Run a command with combined output and a timeout, returning its exit code.""" if argv[:1] != ["--"] or len(argv) == 1: sys.stderr.write("usage: run_captured.py -- COMMAND [ARGS...]\n") return 2 - result = subprocess.run( # noqa: S603 -- the command is the build's own compiler line. - argv[1:], - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - text=True, - errors="replace", - check=False, - ) + try: + result = run_command( + argv[1:], + allowed_executables=[argv[1]], + capture_output=True, + stderr_to_stdout=True, + text=True, + errors="replace", + check=False, + timeout_seconds=CC_TIMEOUT_S, + max_output_bytes=MAX_COMPILE_OUTPUT_BYTES, + ) + except CommandTimedOut as exc: + sys.stderr.write( + f"run_captured: command timed out after {CC_TIMEOUT_S}s (CC_TIMEOUT_S): {argv[1:]}\n" + ) + _print_retained_stdout(exc.stdout) + return 124 + except CommandOutputLimitExceeded as exc: + sys.stderr.write( + "run_captured: command exceeded " + f"{MAX_COMPILE_OUTPUT_BYTES} captured bytes (MAX_COMPILE_OUTPUT_BYTES): {argv[1:]}\n" + ) + _print_retained_stdout(exc.stdout) + return 125 + except (CommandValidationError, OSError) as exc: + reason = str(exc).replace("\r", "\\r").replace("\n", "\\n") + sys.stderr.write(f"run_captured: {reason}\n") + return 126 if result.returncode != 0: sys.stdout.write(result.stdout) return result.returncode diff --git a/core/test/check_exported_symbols.py b/core/test/check_exported_symbols.py index b1e3155000..b436a2b509 100644 --- a/core/test/check_exported_symbols.py +++ b/core/test/check_exported_symbols.py @@ -83,6 +83,8 @@ UNDEFINED_TYPES = {"U", "w", "v"} NM_TIMEOUT = 120 # seconds; HISS-02 bounds every external call +# Bound = max(60, 10x the slowest measured run): c++filt took 0.004 s. +CXXFILT_TIMEOUT_S = 60 @dataclass(frozen=True) @@ -171,12 +173,18 @@ def vmafx_findings(symbols: DynamicSymbols, listed: dict[str, str]) -> list[str] def demangled(names: list[str]) -> dict[str, str]: + """Demangle symbol names using c++filt and return a mapping to demangled names.""" cxxfilt = shutil.which("c++filt") if cxxfilt is None: print("SKIP: c++filt not found") raise SystemExit(77) out = subprocess.run( # noqa: S603 -- resolved c++filt path, names on stdin, no shell - [cxxfilt], input="\n".join(names), check=True, capture_output=True, text=True + [cxxfilt], + input="\n".join(names), + check=True, + capture_output=True, + text=True, + timeout=CXXFILT_TIMEOUT_S, ).stdout.splitlines() # c++filt answers one line per input line; a mismatch is a tool failure. return dict(zip(names, out, strict=True)) diff --git a/core/test/test_cuda_parity_gate_default_run.py b/core/test/test_cuda_parity_gate_default_run.py index 4f6ed8b915..9fa2905a2f 100755 --- a/core/test/test_cuda_parity_gate_default_run.py +++ b/core/test/test_cuda_parity_gate_default_run.py @@ -16,13 +16,26 @@ from __future__ import annotations import json -import os -import subprocess +import signal +import subprocess # nosec B404 -- fixed repository gate requires SIGINT supervision. import sys from pathlib import Path ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from scripts.lib import vmaftest # noqa: E402 - needs the repository root on sys.path + SCRIPT = ROOT / "scripts" / "ci" / "cross_backend_parity_gate.py" +# Not measured: the parity gate needs a free GPU, which the measuring host did not have. +# Set under the 120 s Meson timeout of this test (core/test/meson.build), so this +# message, not Meson's kill, names the stuck run; it covers the lock wait too. +SCRIPT_TIMEOUT_S = 105 +INTERRUPT_GRACE_S = 10 +# After SIGKILL the gate dies at once; this only waits for the pipes to close, which a +# descendant that kept them open would otherwise hold forever. +KILL_GRACE_S = 2 REF = ROOT / "python" / "test" / "resource" / "yuv" / "src01_hrc00_576x324.yuv" DIS = ROOT / "python" / "test" / "resource" / "yuv" / "src01_hrc01_576x324.yuv" @@ -58,19 +71,20 @@ def build_has_cuda(build_dir: Path) -> bool | None: def find_binary() -> Path | None: - """The vmaf CLI of VMAF_BUILD_DIR, of `build/`, or of the working directory.""" - vmaf_bin = os.environ.get("VMAF_BUILD_DIR") - binary = Path(vmaf_bin) / "tools" / "vmaf" if vmaf_bin else ROOT / "build" / "tools" / "vmaf" - if binary.is_file(): - return binary - binary = Path.cwd() / "tools" / "vmaf" - return binary if binary.is_file() else None + """The vmaf CLI under test, from the one resolver every Python test uses. + + scripts/lib/vmaftest.py reads VMAF_BIN, VMAF_BIN_FOR_TESTS, then the repository's + build directories. Those directories are usually CPU builds, which skip_reason() skips: + name a CUDA build with VMAF_BIN. A set variable that names no executable raises + vmaftest.InvalidVmafBinary. + """ + return vmaftest.find(root=ROOT) def skip_reason(binary: Path | None) -> str | None: """Why there is nothing to compare, or None. Decided without the device.""" if binary is None: - return "vmaf binary not found" + return vmaftest.MISSING_MESSAGE if build_has_cuda(binary.parent.parent) is False: return "This build was configured without CUDA" if not REF.is_file() or not DIS.is_file(): @@ -78,16 +92,19 @@ def skip_reason(binary: Path | None) -> str | None: return None -def main() -> int: - binary = find_binary() - reason = skip_reason(binary) - if reason is not None: - sys.stderr.write(f"{reason}; skipping (77)\n") - return 77 +def gate_command(binary: Path, lock_path: Path | None = None) -> list[str]: + """The parity gate command for `binary`, run under the GPU lock at `lock_path`. - cmd = [ + --no-fork: flock execs the gate, so the timeout path signals the gate itself and the + lock frees only when the gate exits; a forked gate would outlive a kill and keep the + lock. `lock_path` defaults to the shared 4090 lock; tests pass a temporary one. + """ + if lock_path is None: + lock_path = Path.home() / ".cache" / "vmafx-locks" / "cuda-4090.lock" + return [ "flock", - str(Path.home() / ".cache" / "vmafx-locks" / "cuda-4090.lock"), + "--no-fork", + str(lock_path), sys.executable, str(SCRIPT), "--vmaf-binary", @@ -105,7 +122,53 @@ def main() -> int: "cuda", ] - proc = subprocess.run(cmd, capture_output=True, text=True, check=False) # noqa: S603 + +def as_text(output: str | bytes | None) -> str: + """Partial output of a TimeoutExpired, which communicate() leaves undecoded.""" + if isinstance(output, bytes): + return output.decode("utf-8", errors="replace") + return output or "" + + +def stop_gate(proc: subprocess.Popen[str]) -> tuple[str, str]: + """SIGINT, then SIGKILL, each wait bounded; return the output collected so far.""" + proc.send_signal(signal.SIGINT) + try: + return proc.communicate(timeout=INTERRUPT_GRACE_S) + except subprocess.TimeoutExpired: + proc.kill() + try: + return proc.communicate(timeout=KILL_GRACE_S) + except subprocess.TimeoutExpired as exc: + sys.stderr.write( + f"A gate descendant still held the output pipes {KILL_GRACE_S}s after the kill" + " (KILL_GRACE_S)\n" + ) + return as_text(exc.stdout), as_text(exc.stderr) + + +def run_gate(cmd: list[str]) -> subprocess.CompletedProcess[str] | None: + """Interrupt a timed-out gate so its asyncio supervisor reaps vmaf before unlocking.""" + with subprocess.Popen( # noqa: S603 # nosec B603 -- fixed argv uses sys.executable and a repository script. + cmd, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True + ) as proc: + try: + stdout, stderr = proc.communicate(timeout=SCRIPT_TIMEOUT_S) + except subprocess.TimeoutExpired: + stdout, stderr = stop_gate(proc) + sys.stdout.write(stdout or "") + sys.stderr.write(stderr or "") + sys.stderr.write( + f"Default parity gate timed out after {SCRIPT_TIMEOUT_S}s (SCRIPT_TIMEOUT_S)\n" + ) + return None + return subprocess.CompletedProcess(cmd, proc.returncode, stdout, stderr) + + +def gate_verdict(proc: subprocess.CompletedProcess[str] | None) -> int: + """The exit code for a finished gate run: 0, 77 to skip, 124 on timeout, else a failure.""" + if proc is None: + return 124 if proc.returncode != 0: if cuda_unavailable(proc.stdout + proc.stderr): sys.stderr.write("CUDA is not available to this binary; skipping (77)\n") @@ -125,5 +188,20 @@ def main() -> int: return 0 +def main() -> int: + """Run the CUDA parity gate under the GPU lock and return its exit code, or 77 to skip.""" + try: + binary = find_binary() + except vmaftest.InvalidVmafBinary as exc: + sys.stderr.write(f"{exc}\n") + return 1 + reason = skip_reason(binary) + if reason is not None or binary is None: + sys.stderr.write(f"{reason}; skipping (77)\n") + return 77 + + return gate_verdict(run_gate(gate_command(binary))) + + if __name__ == "__main__": sys.exit(main()) diff --git a/core/test/test_cuda_parity_gate_skip.py b/core/test/test_cuda_parity_gate_skip.py index eb0f77321e..d0205aea07 100644 --- a/core/test/test_cuda_parity_gate_skip.py +++ b/core/test/test_cuda_parity_gate_skip.py @@ -12,20 +12,40 @@ It also pins that a build directory configured without CUDA is recognised from Meson's own record, which the default-run test reads before it waits for the device lock. + +Last, it pins the timeout of the default run: a hung gate returns 124 with a +message naming the bound, the bound sits under the Meson timeout of the test, +and flock execs the gate so the kill reaches the gate and frees the lock. """ from __future__ import annotations +import contextlib import importlib.util +import io import json +import os import re +import shutil +import signal +import subprocess # nosec B404 -- fixed gate supervision results and timeout test doubles. +import sys import tempfile +import threading +import time import unittest +from collections.abc import Callable from pathlib import Path +from types import ModuleType +from typing import Any +from unittest import mock ROOT = Path(__file__).resolve().parents[2] DEFAULT_RUN = ROOT / "core" / "test" / "test_cuda_parity_gate_default_run.py" +MESON_BUILD = ROOT / "core" / "test" / "meson.build" CLI = ROOT / "core" / "tools" / "vmaf.cpp" +# The real Popen, kept before any test replaces subprocess.Popen with a stand-in. +REAL_POPEN = subprocess.Popen # The refusal in validate_requested_backend(), split over two C string literals. REFUSAL = re.compile( @@ -34,7 +54,8 @@ ) -def load_default_run(): +def load_default_run() -> ModuleType: + """Load the repository parity wrapper for device-free tests.""" spec = importlib.util.spec_from_file_location("parity_gate_default_run", DEFAULT_RUN) if spec is None or spec.loader is None: raise AssertionError(f"cannot load {DEFAULT_RUN}") @@ -71,7 +92,7 @@ def test_a_parity_failure_is_not_a_skip(self) -> None: self.assertFalse(module.cuda_unavailable(cli_refusal("sycl"))) -def build_dir_with(options: str | None) -> tempfile.TemporaryDirectory: +def build_dir_with(options: str | None) -> tempfile.TemporaryDirectory[str]: """A build directory whose Meson option record holds `options`, or none.""" build = tempfile.TemporaryDirectory() if options is not None: @@ -81,10 +102,66 @@ def build_dir_with(options: str | None) -> tempfile.TemporaryDirectory: return build +def make_cli(build: Path) -> Path: + """Create an empty executable `tools/vmaf` under `build` and return its path.""" + binary = build / "tools" / "vmaf" + binary.parent.mkdir(parents=True) + binary.write_text("", encoding="utf-8") + binary.chmod(0o755) + return binary + + +class FindBinaryTest(unittest.TestCase): + """The CLI comes from scripts/lib/vmaftest.py, not from a lookup of this test's own.""" + + def run_with(self, root: Path, env: dict[str, str]) -> tuple[ModuleType, Path | None]: + """find_binary() with `root` as repository root and `env` as the whole environment.""" + module = load_default_run() + with mock.patch.object(module, "ROOT", root), mock.patch.dict(os.environ, env, clear=True): + found: Path | None = module.find_binary() + return module, found + + def test_vmaf_bin_names_the_cli(self) -> None: + """VMAF_BIN wins over a build directory, so a CUDA build can be named.""" + with tempfile.TemporaryDirectory() as root, tempfile.TemporaryDirectory() as cuda: + make_cli(Path(root) / "build") + binary = make_cli(Path(cuda)) + self.assertEqual(self.run_with(Path(root), {"VMAF_BIN": str(binary)})[1], binary) + + def test_build_is_the_fallback(self) -> None: + """Without a variable, the CLI under the repository's `build/` is used.""" + with tempfile.TemporaryDirectory() as root: + binary = make_cli(Path(root) / "build") + self.assertEqual(self.run_with(Path(root), {})[1], binary) + + def test_no_cli_is_a_skip(self) -> None: + """No variable and no build: main() skips with the resolver's message.""" + with tempfile.TemporaryDirectory() as root: + module, found = self.run_with(Path(root), {}) + self.assertIsNone(found) + self.assertEqual(module.skip_reason(None), module.vmaftest.MISSING_MESSAGE) + + def test_invalid_vmaf_bin_is_a_failure(self) -> None: + """A VMAF_BIN naming no executable fails the run, not skips it or falls through.""" + module = load_default_run() + with tempfile.TemporaryDirectory() as root: + make_cli(Path(root) / "build") + missing = str(Path(root) / "missing" / "vmaf") + stderr = io.StringIO() + with ( + mock.patch.object(module, "ROOT", Path(root)), + mock.patch.dict(os.environ, {"VMAF_BIN": missing}, clear=True), + contextlib.redirect_stderr(stderr), + ): + self.assertEqual(module.main(), 1) + self.assertIn(f"VMAF_BIN={missing} does not name an executable file", stderr.getvalue()) + + class BuildHasCudaTest(unittest.TestCase): def has_cuda(self, options: str | None) -> bool | None: with build_dir_with(options) as build: - return load_default_run().build_has_cuda(Path(build)) + has_cuda: bool | None = load_default_run().build_has_cuda(Path(build)) + return has_cuda def test_build_configured_without_cuda_is_recognised(self) -> None: options = json.dumps( @@ -105,6 +182,7 @@ def test_a_build_that_does_not_say_is_not_skipped(self) -> None: self.assertIsNone(self.has_cuda(json.dumps([{"name": "enable_cuda", "value": "auto"}]))) def test_a_build_without_cuda_is_skipped_before_the_device_lock(self) -> None: + """Skip CPU-only builds before constructing or launching the lock command.""" module = load_default_run() options = json.dumps([{"name": "enable_cuda", "value": False}]) with build_dir_with(options) as build: @@ -112,11 +190,319 @@ def test_a_build_without_cuda_is_skipped_before_the_device_lock(self) -> None: binary.parent.mkdir() binary.write_text("", encoding="utf-8") self.assertEqual(module.skip_reason(binary), "This build was configured without CUDA") - self.assertEqual(module.skip_reason(None), "vmaf binary not found") + with ( + mock.patch.object(module, "find_binary", return_value=binary), + mock.patch.object( + module.subprocess, + "Popen", + side_effect=AssertionError("lock acquired before skip"), + ) as popen, + contextlib.redirect_stderr(io.StringIO()), + ): + self.assertEqual(module.main(), 77) + popen.assert_not_called() + self.assertEqual(module.skip_reason(None), module.vmaftest.MISSING_MESSAGE) # main() decides the skip ahead of the command that waits for the lock. source = DEFAULT_RUN.read_text(encoding="utf-8") main = source[source.index("def main() -> int:") :] - self.assertLess(main.index("reason = skip_reason(binary)"), main.index('"flock",')) + self.assertLess( + main.index("reason = skip_reason(binary)"), main.index("gate_command(binary)") + ) + + +def meson_timeout(test_name: str) -> int: + """The `timeout :` that core/test/meson.build gives the test `test_name`.""" + block = re.search( + rf"^test\('{re.escape(test_name)}',(?P.*?)^\)", + MESON_BUILD.read_text(encoding="utf-8"), + re.MULTILINE | re.DOTALL, + ) + if block is None: + raise AssertionError(f"core/test/meson.build no longer registers {test_name}") + timeout = re.search(r"\btimeout\s*:\s*(\d+)", block.group("body")) + if timeout is None: + raise AssertionError(f"{test_name} has no timeout in core/test/meson.build") + return int(timeout.group(1)) + + +class DefaultRunTimeoutTest(unittest.TestCase): + """main() of the default run, with the device-free steps replaced by stubs.""" + + def run_main( + self, outcome: subprocess.CompletedProcess[str] | BaseException + ) -> tuple[ModuleType, int, str, mock.MagicMock]: + """Run main() with `outcome` as the result, or the exception, of the gate run.""" + module = load_default_run() + stderr = io.StringIO() + binary = Path("build") / "tools" / "vmaf" + with contextlib.ExitStack() as stack: + stack.enter_context(mock.patch.object(module, "find_binary", return_value=binary)) + stack.enter_context(mock.patch.object(module, "skip_reason", return_value=None)) + gate = stack.enter_context(mock.patch.object(module.subprocess, "Popen")) + proc = gate.return_value.__enter__.return_value + if isinstance(outcome, BaseException): + proc.communicate.side_effect = [outcome, ("partial stdout", "partial stderr")] + else: + proc.returncode = outcome.returncode + proc.communicate.return_value = (outcome.stdout, outcome.stderr) + stack.enter_context(contextlib.redirect_stdout(io.StringIO())) + stack.enter_context(contextlib.redirect_stderr(stderr)) + rc = module.main() + return module, rc, stderr.getvalue(), gate + + def test_a_clean_run_passes_and_flock_execs_the_gate(self) -> None: + """Retain flock exec semantics and bound communicate on successful runs.""" + cells = "motion cpu ↔ cuda PASS\ncambi cpu ↔ cuda PASS\n" + module, rc, _, gate = self.run_main(subprocess.CompletedProcess([], 0, cells, "")) + self.assertEqual(rc, 0) + # Without --no-fork the timeout kills flock and the forked gate keeps the lock. + self.assertEqual(gate.call_args.args[0][:2], ["flock", "--no-fork"]) + proc = gate.return_value.__enter__.return_value + proc.communicate.assert_called_once_with(timeout=module.SCRIPT_TIMEOUT_S) + + def test_a_hung_gate_returns_124_and_names_the_bound(self) -> None: + """Interrupt a hung gate, retaining diagnostics and the named timeout.""" + module, rc, stderr, gate = self.run_main(subprocess.TimeoutExpired(["flock"], 105)) + self.assertEqual(rc, 124) + self.assertIn(f"timed out after {module.SCRIPT_TIMEOUT_S}s (SCRIPT_TIMEOUT_S)", stderr) + self.assertIn("partial stderr", stderr) + proc = gate.return_value.__enter__.return_value + proc.send_signal.assert_called_once_with(signal.SIGINT) + self.assertEqual(proc.communicate.call_args.kwargs["timeout"], module.INTERRUPT_GRACE_S) + + def test_the_bound_is_under_the_meson_timeout(self) -> None: + """Keep the command timeout and both kill graces below Meson termination.""" + # At or over the Meson timeout, Meson kills the test first and the + # message above never prints. + module = load_default_run() + bound = module.SCRIPT_TIMEOUT_S + module.INTERRUPT_GRACE_S + module.KILL_GRACE_S + self.assertLess(bound, meson_timeout("test_cuda_parity_gate_default_run")) + + def test_partial_output_is_decoded_or_empty(self) -> None: + """TimeoutExpired carries bytes or None; undecodable bytes are replaced, not raised.""" + module = load_default_run() + self.assertEqual(module.as_text(b"partial \xff"), "partial �") + self.assertEqual(module.as_text(None), "") + self.assertEqual(module.as_text(""), "") + self.assertEqual(module.as_text("already text"), "already text") + + +@unittest.skipUnless(os.name == "posix" and shutil.which("flock"), "POSIX flock is unavailable") +class DefaultRunProcessCleanupTest(unittest.TestCase): + """Exercise the real gate supervisor with a fake vmaf, without a GPU.""" + + def test_timeout_reaps_vmaf_before_releasing_the_lock(self) -> None: + """SIGINT must reap the safe_subprocess child before the GPU lock is free.""" + import fcntl # noqa: PLC0415 -- unavailable on Windows; imported only after POSIX skip. + + module = load_default_run() + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + pid_file, lock = root / "pid", root / "lock" + reference, distorted = root / "reference.yuv", root / "distorted.yuv" + reference.write_bytes(b"reference") + distorted.write_bytes(b"distorted") + fake = root / "vmaf" + fake.write_text( + f"#!{sys.executable}\nimport os, time\nfrom pathlib import Path\n" + f"Path({str(pid_file)!r}).write_text(str(os.getpid()))\n" + "print('fake vmaf started', flush=True)\ntime.sleep(600)\n", + encoding="utf-8", + ) + fake.chmod(0o755) + stdout, stderr = io.StringIO(), io.StringIO() + try: + with ( + mock.patch.object(module, "REF", reference), + mock.patch.object(module, "DIS", distorted), + mock.patch.object(module, "find_binary", return_value=fake), + mock.patch.object(module, "skip_reason", return_value=None), + mock.patch.object( + module.subprocess, "Popen", side_effect=popen_when_ready(pid_file) + ), + mock.patch.object(module, "SCRIPT_TIMEOUT_S", 3), + mock.patch.object(module, "INTERRUPT_GRACE_S", 5), + contextlib.redirect_stdout(stdout), + contextlib.redirect_stderr(stderr), + ): + command = module.gate_command(fake, lock_path=lock) + command += ["--features", "motion", "--workdir", str(root / "work")] + with mock.patch.object(module, "gate_command", return_value=command): + self.assertEqual(module.main(), 124) + self.assertIn("timed out after 3s", stderr.getvalue()) + self.assertTrue(pid_file.is_file(), stderr.getvalue()) + pid = int(pid_file.read_text()) + self.assertTrue(wait_until_gone(pid), f"vmaf {pid} survived gate timeout") + with lock.open("a") as descriptor: + fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + finally: + if pid_file.is_file(): + with contextlib.suppress(ProcessLookupError): + os.kill(int(pid_file.read_text()), signal.SIGKILL) + + def test_interrupt_ignoring_gate_is_killed_and_partial_output_is_printed(self) -> None: + """Bound fallback even for a gate ignoring SIGINT, with a watchdog for mutants.""" + module = load_default_run() + watchdogs = [] + ready_at: list[float] = [] + + def arm(proc: subprocess.Popen[str]) -> None: + """Note when the gate is ready; kill it at 3 s if the fallback is removed.""" + ready_at.append(time.monotonic()) + watchdog = threading.Timer(3, proc.kill) + watchdog.start() + watchdogs.append(watchdog) + + with tempfile.TemporaryDirectory() as directory: + ready = Path(directory) / "ready" + script = ( + "import signal, sys, time; signal.signal(signal.SIGINT, signal.SIG_IGN); " + "print('partial out', flush=True); " + "print('partial err', file=sys.stderr, flush=True); " + f"open({str(ready)!r}, 'w').close(); time.sleep(600)" + ) + stdout, stderr = io.StringIO(), io.StringIO() + try: + with ( + mock.patch.object( + module.subprocess, "Popen", side_effect=popen_when_ready(ready, arm) + ), + mock.patch.object(module, "SCRIPT_TIMEOUT_S", 0.5), + mock.patch.object(module, "INTERRUPT_GRACE_S", 0.5), + contextlib.redirect_stdout(stdout), + contextlib.redirect_stderr(stderr), + ): + result = module.run_gate([sys.executable, "-c", script]) + self.assertIsNone(result) + self.assertLess(time.monotonic() - ready_at[0], 2.5) + self.assertIn("partial out", stdout.getvalue()) + self.assertIn("partial err", stderr.getvalue()) + finally: + for watchdog in watchdogs: + watchdog.cancel() + watchdog.join() + + def test_a_descendant_holding_the_pipes_does_not_hang_the_kill(self) -> None: + """After SIGKILL, wait at most KILL_GRACE_S for pipes a descendant kept open.""" + module = load_default_run() + with tempfile.TemporaryDirectory() as directory: + pid_file = Path(directory) / "pid" + script = ( + "import signal, subprocess, sys, time; " + "signal.signal(signal.SIGINT, signal.SIG_IGN); " + "holder = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(600)']," + " start_new_session=True); " + f"open({str(pid_file)!r}, 'w').write(str(holder.pid)); " + "print('partial out', flush=True); time.sleep(600)" + ) + # Without the bound the test would hang; the watchdog closes the pipes 4 s after + # the child is ready, so a removed bound fails the elapsed-time assertion instead. + watchdog = threading.Timer(4, kill_pid_in, args=(pid_file,)) + ready_at: list[float] = [] + + def arm(_proc: subprocess.Popen[str]) -> None: + """Note when the holder exists and start the watchdog from that point.""" + ready_at.append(time.monotonic()) + watchdog.start() + + stdout, stderr = io.StringIO(), io.StringIO() + try: + with ( + mock.patch.object( + module.subprocess, "Popen", side_effect=popen_when_ready(pid_file, arm) + ), + mock.patch.object(module, "SCRIPT_TIMEOUT_S", 0.5), + mock.patch.object(module, "INTERRUPT_GRACE_S", 0.5), + mock.patch.object(module, "KILL_GRACE_S", 0.5), + contextlib.redirect_stdout(stdout), + contextlib.redirect_stderr(stderr), + ): + result = module.run_gate([sys.executable, "-c", script]) + self.assertIsNone(result) + self.assertLess(time.monotonic() - ready_at[0], 3.5) + self.assertIn("partial out", stdout.getvalue()) + self.assertIn("still held the output pipes 0.5s after the kill", stderr.getvalue()) + finally: + watchdog.cancel() + if watchdog.is_alive(): + watchdog.join() + kill_pid_in(pid_file) + + +def kill_pid_in(pid_file: Path) -> None: + """SIGKILL the process whose PID `pid_file` holds, if it was written and still runs.""" + with contextlib.suppress(OSError, ValueError): + os.kill(int(pid_file.read_text(encoding="utf-8")), signal.SIGKILL) + + +def wait_for_file(path: Path, proc: subprocess.Popen[str]) -> bool: + """Poll for at most ten seconds until `path` exists or the child exits.""" + for _ in range(200): + if path.is_file(): + return True + if proc.poll() is not None: + return False + time.sleep(0.05) + return False + + +def stop_and_collect(proc: subprocess.Popen[str]) -> str: + """Stop and reap a child, closing its pipes even if a descendant holds them open.""" + if proc.poll() is None: + proc.kill() + try: + try: + _, stderr = proc.communicate(timeout=5) + except subprocess.TimeoutExpired: + stderr = "" + with contextlib.suppress(subprocess.TimeoutExpired): + proc.wait(timeout=5) + return stderr or "" + finally: + for pipe in (proc.stdin, proc.stdout, proc.stderr): + if pipe is not None: + pipe.close() + + +def popen_when_ready( + ready_file: Path, on_ready: Callable[[subprocess.Popen[str]], None] | None = None +) -> Callable[..., subprocess.Popen[str]]: + """Popen stand-in that returns only once the child has written `ready_file`. + + The timeout under test starts when Popen returns. Waiting for the child's own + signal first means start-up time on a loaded host cannot decide the result. + """ + + def start(*args: Any, **kwargs: Any) -> subprocess.Popen[str]: + """Start the real process, then hold until it reports ready.""" + proc: subprocess.Popen[str] = REAL_POPEN(*args, **kwargs) + if not wait_for_file(ready_file, proc): + stderr_start = stop_and_collect(proc)[:500] + raise AssertionError( + f"the child did not write {ready_file}; exit code {proc.returncode}; " + f"stderr starts with {stderr_start!r}" + ) + if on_ready is not None: + try: + on_ready(proc) + except BaseException: + with contextlib.suppress(BaseException): + stop_and_collect(proc) + raise + return proc + + return start + + +def wait_until_gone(pid: int) -> bool: + """Poll for at most five seconds until the child PID has been reaped.""" + for _ in range(100): + try: + os.kill(pid, 0) + except ProcessLookupError: + return True + time.sleep(0.05) + return False if __name__ == "__main__": diff --git a/core/test/test_gpu_public_header_docs.py b/core/test/test_gpu_public_header_docs.py index 026da2f922..e7b4d902fb 100644 --- a/core/test/test_gpu_public_header_docs.py +++ b/core/test/test_gpu_public_header_docs.py @@ -13,6 +13,8 @@ from pathlib import Path INCLUDE_DIR = Path(__file__).resolve().parents[1] / "include" / "libvmaf" +# Bound = max(60, 10x the slowest measured run): the call took 0.09 s. +SUBPROCESS_TIMEOUT_S = 60 # Between a block and its declaration only a deprecation marker may stand # (VMAF_DEPRECATED, ADR-1852 decision D7). ATTACHED_GAP = re.compile(r'\s*(?:VMAF_DEPRECATED\("[^"]*"\)\s*)?') @@ -166,6 +168,7 @@ def test_doxygen_execution_zero_warnings(self) -> None: cwd=str(self.repo_root), capture_output=True, text=True, + timeout=SUBPROCESS_TIMEOUT_S, ) self.assertEqual(proc.returncode, 0, f"doxygen failed: {proc.stderr}\n{proc.stdout}") if warn_log.exists(): diff --git a/core/test/test_strict_fp_compiler_args.py b/core/test/test_strict_fp_compiler_args.py index 292a528dc7..6ab87b32eb 100644 --- a/core/test/test_strict_fp_compiler_args.py +++ b/core/test/test_strict_fp_compiler_args.py @@ -16,19 +16,50 @@ from __future__ import annotations +import functools import importlib.util import json import os import re import shutil -import subprocess import sys import tempfile import textwrap +import time import unittest from pathlib import Path +from unittest import mock + +# Direct script execution also needs the canonical repository subprocess helper. +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + +from test_cuda_parity_gate_skip import meson_timeout + +from scripts.lib.safe_subprocess import TextCommandResult +from scripts.lib.safe_subprocess import run as run_command + +# One deadline for this Meson test process, shared by all matrix and compiler calls. +TEST_TIMEOUT_S = 80 + + +@functools.cache +def _test_deadline() -> float: + """Start the shared deadline when its first caller needs a command budget.""" + return time.monotonic() + TEST_TIMEOUT_S + + +def remaining_timeout(bound: float) -> float: + """Limit a command to the shared test deadline, refusing work after it expires.""" + remaining = _test_deadline() - time.monotonic() + if remaining <= 0: + raise AssertionError(f"strict-FP test deadline exhausted after {TEST_TIMEOUT_S}s") + return min(bound, remaining) + SOURCE_MESON = Path(__file__).resolve().parents[1] / "src" / "meson.build" +# Bounds = max(60, 10x the slowest measured run): the meson call took 1.33 s, icx 0.09 s. +MESON_TIMEOUT_S = 60 +CC_TIMEOUT_S = 60 TEST_MESON = Path(__file__).resolve().with_name("meson.build") TOOLS_MESON = Path(__file__).resolve().parents[1] / "tools" / "meson.build" METAL_MESON = Path(__file__).resolve().parents[1] / "src" / "metal" / "meson.build" @@ -410,11 +441,14 @@ def _meson_setup(case: unittest.TestCase, name: str, fixture: str) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) (root / "meson.build").write_text(fixture, encoding="utf-8") - result = subprocess.run( # noqa: S603 -- Meson is resolved locally. + result = run_command( [*MESON_COMMAND, "setup", str(root / "build"), str(root)], + allowed_executables=[MESON_COMMAND[0]], capture_output=True, check=False, + text=True, encoding="utf-8", + timeout_seconds=remaining_timeout(MESON_TIMEOUT_S), ) case.assertEqual( result.returncode, @@ -438,6 +472,29 @@ def _target_block(source: str, target: str) -> str: raise AssertionError(f"{target}: unbalanced static_library() call") +class StrictFpDeadlineTest(unittest.TestCase): + """Keep all calls inside the Meson timeout and reject an exhausted budget.""" + + def test_deadline_is_under_meson_timeout(self) -> None: + """Leave time for process-group cleanup before Meson kills the test.""" + self.assertLess(TEST_TIMEOUT_S, meson_timeout("test_strict_fp_compiler_args")) + + def test_remaining_budget_caps_the_call_bound(self) -> None: + """Use the smaller of the call bound and time remaining in the test.""" + deadline = _test_deadline() + with mock.patch.object(time, "monotonic", return_value=deadline - 3): + self.assertEqual(remaining_timeout(60), 3) + self.assertEqual(remaining_timeout(1), 1) + + def test_exhausted_deadline_names_the_bound(self) -> None: + """Refuse subprocess calls at or beyond the shared deadline.""" + deadline = _test_deadline() + for offset in (0, 1): + with mock.patch.object(time, "monotonic", return_value=deadline + offset): + with self.assertRaisesRegex(AssertionError, f"deadline.*{TEST_TIMEOUT_S}s"): + remaining_timeout(60) + + class StrictFpCompilerArgsTest(unittest.TestCase): @unittest.skipUnless(MESON_COMMAND, "Meson is not installed") def test_compiler_matrix_executes_shipped_policy(self) -> None: @@ -850,12 +907,15 @@ def test_icx_strict_policy_draws_no_overriding_option_warning(self) -> None: src.write_text("float f(float a, float b, float c) { return a * b + c; }\n") obj = Path(tmp) / "probe.o" - def compile_with(flags: list[str]) -> subprocess.CompletedProcess[str]: - return subprocess.run( # noqa: S603 -- icx is resolved on PATH above. + def compile_with(flags: list[str]) -> TextCommandResult: + """Compile the C probe source with icx using the given compiler flags.""" + return run_command( [icx, "-O2", "-Werror", *flags, "-c", str(src), "-o", str(obj)], + allowed_executables=[icx], capture_output=True, text=True, check=False, + timeout_seconds=remaining_timeout(CC_TIMEOUT_S), ) ok = compile_with(strict) @@ -865,11 +925,13 @@ def compile_with(flags: list[str]) -> subprocess.CompletedProcess[str]: self.assertNotEqual(bad.returncode, 0) self.assertIn("-Woverriding-option", bad.stderr) # No contraction: a * b + c is two instructions with the strict spelling. - disasm = subprocess.run( # noqa: S603 -- icx is resolved on PATH above. + disasm = run_command( [icx, "-O2", "-mavx2", "-mfma", *strict, "-S", "-o", "-", str(src)], + allowed_executables=[icx], capture_output=True, text=True, check=False, + timeout_seconds=remaining_timeout(CC_TIMEOUT_S), ) self.assertEqual(disasm.returncode, 0, disasm.stderr) self.assertNotIn("vfmadd", disasm.stdout) diff --git a/core/test/test_vmafx_python_binding.py b/core/test/test_vmafx_python_binding.py index 6625fe262c..7c46e592a5 100644 --- a/core/test/test_vmafx_python_binding.py +++ b/core/test/test_vmafx_python_binding.py @@ -29,6 +29,8 @@ from vmafx import _api as vmafx REPO_ROOT = Path(__file__).resolve().parents[2] +# Bound = max(60, 10x the slowest measured run): the vmaf call took 0.17 s cold. +VMAF_TIMEOUT_S = 60 WRAPPED_CLASSES = ( vmafx.Library, vmafx.Context, @@ -620,6 +622,7 @@ def test_missing_input_fails_when_the_pair_is_required(self) -> None: self._require_inputs(absent) def test_netflix_576x324_matches_c_api_float_hex(self) -> None: + """The Python binding pooled float hex matches the C API contract output.""" lib_path = Path(os.environ["VMAFX_LIBRARY"]).resolve() contract_bin = lib_path.parent.parent / "test" / "vmafx_score_contract" self._require_inputs((self.REF_YUV, self.DIST_YUV, self.MODEL_JSON, contract_bin)) @@ -638,6 +641,7 @@ def test_netflix_576x324_matches_c_api_float_hex(self) -> None: capture_output=True, text=True, check=True, + timeout=VMAF_TIMEOUT_S, ) c_payload = json.loads(proc.stdout) w, h = 576, 324 diff --git a/mcp-server/vmaf-mcp/tests/test_import_graph.py b/mcp-server/vmaf-mcp/tests/test_import_graph.py index f70ba29239..877500ad09 100644 --- a/mcp-server/vmaf-mcp/tests/test_import_graph.py +++ b/mcp-server/vmaf-mcp/tests/test_import_graph.py @@ -14,6 +14,9 @@ import pytest +# Bound = max(60, 10x the slowest measured run): each call took 0.58 s or less. +SCRIPT_TIMEOUT_S = 60 + def _module_name(source: Path) -> str: return "vmaf_mcp" if source.stem == "__init__" else f"vmaf_mcp.{source.stem}" @@ -99,6 +102,7 @@ def test_http_transport_import_isolated_from_server_bootstrap() -> None: capture_output=True, text=True, env=env, + timeout=SCRIPT_TIMEOUT_S, ) assert completed.returncode == 0, completed.stderr @@ -140,5 +144,6 @@ def dumps_strict(self, data): capture_output=True, text=True, env=env, + timeout=SCRIPT_TIMEOUT_S, ) assert completed.returncode == 0, completed.stderr diff --git a/mcp-server/vmaf-mcp/tests/test_pytest_pythonpath.py b/mcp-server/vmaf-mcp/tests/test_pytest_pythonpath.py index 2a600b3572..e0f8f18a3c 100644 --- a/mcp-server/vmaf-mcp/tests/test_pytest_pythonpath.py +++ b/mcp-server/vmaf-mcp/tests/test_pytest_pythonpath.py @@ -9,12 +9,14 @@ from __future__ import annotations import os -import subprocess +import subprocess # nosec B404 -- fixed isolated interpreter import checks. import sys import tomllib from pathlib import Path _MCP_ROOT = Path(__file__).resolve().parents[1] +# Bound = max(60, 10x the slowest measured run): each call took 0.58 s or less. +SCRIPT_TIMEOUT_S = 60 def test_vmaf_mcp_importable_without_editable_install() -> None: @@ -55,7 +57,13 @@ def test_pythonpath_red_demonstration(tmp_path: Path) -> None: "-c", "import vmaf_mcp", ] - proc_fail = subprocess.run(cmd_fail, env=child_env, capture_output=True, text=True) + proc_fail = subprocess.run( # noqa: S603, RUF100 # nosec B603 -- fixed sys.executable argv checks local package imports. + cmd_fail, + env=child_env, + capture_output=True, + text=True, + timeout=SCRIPT_TIMEOUT_S, + ) assert proc_fail.returncode != 0 assert "ModuleNotFoundError" in proc_fail.stderr assert "vmaf_mcp" in proc_fail.stderr @@ -76,5 +84,11 @@ def test_pythonpath_red_demonstration(tmp_path: Path) -> None: "assert bool(vmaf_mcp.__version__)" ), ] - proc_pass = subprocess.run(cmd_pass, env=child_env, capture_output=True, text=True) + proc_pass = subprocess.run( # noqa: S603, RUF100 # nosec B603 -- fixed sys.executable argv checks local package imports. + cmd_pass, + env=child_env, + capture_output=True, + text=True, + timeout=SCRIPT_TIMEOUT_S, + ) assert proc_pass.returncode == 0, proc_pass.stderr diff --git a/python/test/command_line_test.py b/python/test/command_line_test.py index 1f9bf29a21..b0ae88b9b7 100644 --- a/python/test/command_line_test.py +++ b/python/test/command_line_test.py @@ -1,15 +1,39 @@ # SPDX-License-Identifier: BSD-2-Clause-Patent import os -import subprocess +import shlex +import sys import tempfile import unittest +from pathlib import Path from vmaf import ExternalProgram, run_process from vmaf.config import VmafConfig from vmaf.tools.misc import MyTestCase +_REPO_ROOT = Path(__file__).resolve().parents[2] +if str(_REPO_ROOT) not in sys.path: + sys.path.insert(0, str(_REPO_ROOT)) + +from scripts.lib.safe_subprocess import run as run_command # noqa: E402 + __copyright__ = "Copyright 2016-2020, Netflix, Inc." __license__ = "BSD+Patent" +# Bound = 10x the slowest measured run, rounded up: the slowest call took 28.61 s cold. +VMAF_TIMEOUT_S = 300 + + +def _run_quiet(cmd): + """Run a test command line without a shell, discarding its output like `>/dev/null 2>&1`.""" + argv = shlex.split(cmd) + with open(os.devnull, "w", encoding="utf-8") as devnull: + return run_command( + argv, + allowed_executables=[argv[0]], + stdout=devnull, + stderr_to_stdout=True, + check=False, + timeout_seconds=VMAF_TIMEOUT_S, + ).returncode class RunProcessTest(MyTestCase): @@ -153,6 +177,7 @@ def test_run_vmaf_ci(self): self.assertEqual(ret, 0) def test_run_vmaf_both_local_explain_and_ci(self): + """Passing both --local-explain and --ci returns exit code 2.""" exe = VmafConfig.root_path("python", "vmaf", "script", "run_vmaf.py") line = ( "yuv420p 576 324 {root}/python/test/resource/yuv/src01_hrc00_576x324.yuv " @@ -160,13 +185,8 @@ def test_run_vmaf_both_local_explain_and_ci(self): root=VmafConfig.root_path() ) ) - cmd = "{exe} {line} --local-explain --ci >/dev/null 2>&1".format(line=line, exe=exe) - # `cmd` is built from VmafConfig.root_path() + hardcoded test-fixture - # paths under python/test/resource/yuv/. No attacker-controlled string - # flows in. shell=True is needed for `>/dev/null 2>&1` redirection / - # CLI argument quoting. See Research-0090, F13–F17. - # nosemgrep: python.lang.security.audit.subprocess-shell-true.subprocess-shell-true - ret = subprocess.call(cmd, shell=True) + cmd = "{exe} {line} --local-explain --ci".format(line=line, exe=exe) + ret = _run_quiet(cmd) self.assertEqual(ret, 2) def test_run_psnr(self): @@ -216,6 +236,7 @@ def tearDown(self) -> None: super().tearDown() def test_run_vmafexec(self): + """Run vmafexec and assert successful execution with expected PSNR output.""" exe = ExternalProgram.vmafexec cmd = ( "{exe} --reference {ref} --distorted {dis} --width 576 --height 324 --pixel_format 420 --bitdepth 8 --xml --feature psnr " @@ -227,12 +248,7 @@ def test_run_vmafexec(self): output=self.output_file_path, ) ) - # `cmd` is built from VmafConfig.root_path() + hardcoded test-fixture - # paths under python/test/resource/yuv/. No attacker-controlled string - # flows in. shell=True is needed for `>/dev/null 2>&1` redirection / - # CLI argument quoting. See Research-0090, F13–F17. - # nosemgrep: python.lang.security.audit.subprocess-shell-true.subprocess-shell-true - ret = subprocess.call(cmd, shell=True) + ret = _run_quiet(cmd) self.assertEqual(ret, self.RC_SUCCESS) with open(self.output_file_path, "rt") as fo: fc = fo.read() @@ -242,6 +258,7 @@ def test_run_vmafexec(self): ) def test_run_vmafexec_with_frame_skipping(self): + """Run vmafexec with equal frame skipping and assert expected PSNR output.""" exe = ExternalProgram.vmafexec cmd = ( "{exe} --reference {ref} --distorted {dis} --width 576 --height 324 --pixel_format 420 --bitdepth 8 --xml --feature psnr " @@ -253,12 +270,7 @@ def test_run_vmafexec_with_frame_skipping(self): output=self.output_file_path, ) ) - # `cmd` is built from VmafConfig.root_path() + hardcoded test-fixture - # paths under python/test/resource/yuv/. No attacker-controlled string - # flows in. shell=True is needed for `>/dev/null 2>&1` redirection / - # CLI argument quoting. See Research-0090, F13–F17. - # nosemgrep: python.lang.security.audit.subprocess-shell-true.subprocess-shell-true - ret = subprocess.call(cmd, shell=True) + ret = _run_quiet(cmd) self.assertEqual(ret, self.RC_SUCCESS) with open(self.output_file_path, "rt") as fo: fc = fo.read() @@ -268,6 +280,7 @@ def test_run_vmafexec_with_frame_skipping(self): ) def test_run_vmafexec_with_frame_skipping_unequal(self): + """Run vmafexec with unequal frame skipping and assert expected PSNR output.""" exe = ExternalProgram.vmafexec cmd = ( "{exe} --reference {ref} --distorted {dis} --width 576 --height 324 --pixel_format 420 --bitdepth 8 --xml --feature psnr " @@ -279,12 +292,7 @@ def test_run_vmafexec_with_frame_skipping_unequal(self): output=self.output_file_path, ) ) - # `cmd` is built from VmafConfig.root_path() + hardcoded test-fixture - # paths under python/test/resource/yuv/. No attacker-controlled string - # flows in. shell=True is needed for `>/dev/null 2>&1` redirection / - # CLI argument quoting. See Research-0090, F13–F17. - # nosemgrep: python.lang.security.audit.subprocess-shell-true.subprocess-shell-true - ret = subprocess.call(cmd, shell=True) + ret = _run_quiet(cmd) self.assertEqual(ret, self.RC_SUCCESS) with open(self.output_file_path, "rt") as fo: fc = fo.read() diff --git a/python/test/tools_test.py b/python/test/tools_test.py index 819011c44a..0e24554053 100644 --- a/python/test/tools_test.py +++ b/python/test/tools_test.py @@ -1,14 +1,23 @@ # SPDX-License-Identifier: BSD-2-Clause-Patent import os -import subprocess import sys import unittest +from pathlib import Path from test.testutil import set_default_576_324_videos_for_testing from vmaf.config import VmafConfig from vmaf.core.quality_runner import PsnrQualityRunner from vmaf.tools.misc import MyTestCase, QualityRunnerTestMixin, import_python_file +_REPO_ROOT = Path(__file__).resolve().parents[2] +if str(_REPO_ROOT) not in sys.path: + sys.path.insert(0, str(_REPO_ROOT)) + +from scripts.lib.safe_subprocess import run as run_command # noqa: E402 + +# Bound = max(60, 10x the slowest measured run): the script took 0.95 s. +SCRIPT_TIMEOUT_S = 60 + class QualityRunnerTestMixinTest(MyTestCase, QualityRunnerTestMixin): @@ -37,19 +46,22 @@ class MiscTest(unittest.TestCase): @staticmethod def _run_python_with_warnings_as_errors(source): + """Execute Python source code in a subprocess with warnings treated as errors.""" env = os.environ.copy() python_path = VmafConfig.root_path("python") if env.get("PYTHONPATH"): python_path = os.pathsep.join((python_path, env["PYTHONPATH"])) env["PYTHONPATH"] = python_path - return subprocess.run( + return run_command( [sys.executable, "-W", "error", "-c", source], + allowed_executables=[sys.executable], cwd=VmafConfig.root_path(), env=env, capture_output=True, text=True, check=False, + timeout_seconds=SCRIPT_TIMEOUT_S, ) def test_sigproc_import_is_warning_free(self): diff --git a/python/test/vmaf_v1_quality_runner_test.py b/python/test/vmaf_v1_quality_runner_test.py index 5b9e3b6401..f5c89fb51d 100644 --- a/python/test/vmaf_v1_quality_runner_test.py +++ b/python/test/vmaf_v1_quality_runner_test.py @@ -2,8 +2,9 @@ from __future__ import absolute_import import os -import subprocess +import sys import unittest +from pathlib import Path from test.testutil import ( set_default_576_324_hdr_videos_for_testing, set_default_576_324_videos_for_testing, @@ -15,6 +16,12 @@ from vmaf.core.quality_runner import VmafexecQualityRunner from vmaf.tools.misc import MyTestCase +_REPO_ROOT = Path(__file__).resolve().parents[2] +if str(_REPO_ROOT) not in sys.path: + sys.path.insert(0, str(_REPO_ROOT)) + +from scripts.lib.safe_subprocess import run as run_command # noqa: E402 + __copyright__ = "Copyright 2016-2020, Netflix, Inc." __license__ = "BSD+Patent" @@ -22,9 +29,12 @@ # Set only on the CI leg built with -Denable_zimg=false: there the HDR cases assert the # refusal that names zimg. A build with zimg fails that assertion. EXPECT_NO_ZIMG = os.environ.get("VMAF_EXPECT_NO_ZIMG") == "1" +# Bound = 10x the slowest measured run, rounded up: the HDR probe took 11.52 s cold. +VMAF_TIMEOUT_S = 120 def hdr_conversion_available(): + """Return whether HDR conversion and zimg support are available in vmafexec.""" # The HDR model needs a vmaf built with zimg (picture conversion); a build # without it exits non-zero naming it. ref_path, _, _, _ = set_default_576_324_hdr_videos_for_testing() @@ -59,7 +69,15 @@ def hdr_conversion_available(): "--color_matrix_" + suffix, "bt2020nc", ] - proc = subprocess.run(cmd, capture_output=True, text=True) + # A timeout raises: returning False here would pass the no-zimg leg silently. + proc = run_command( + cmd, + allowed_executables=[cmd[0]], + capture_output=True, + text=True, + check=False, + timeout_seconds=VMAF_TIMEOUT_S, + ) return not ( proc.returncode != 0 and ( diff --git a/scripts/ci/tests/test_run_captured.py b/scripts/ci/tests/test_run_captured.py index 3202db8e93..898acfff59 100644 --- a/scripts/ci/tests/test_run_captured.py +++ b/scripts/ci/tests/test_run_captured.py @@ -5,30 +5,52 @@ from __future__ import annotations -import subprocess +import contextlib +import importlib.util +import io +import subprocess # nosec B404 -- launch the fixed repository script in integration tests. import sys import unittest from pathlib import Path +from types import ModuleType +from unittest import mock SCRIPT = Path(__file__).resolve().parents[3] / "core/src/sycl/run_captured.py" PYTHON = sys.executable SCRIPT_TIMEOUT_S = 60 +def load_run_captured_module() -> ModuleType: + """Load core/src/sycl/run_captured.py dynamically via its file location.""" + spec = importlib.util.spec_from_file_location("run_captured", SCRIPT) + if spec is None or spec.loader is None: + raise ImportError(f"Cannot load module spec from {SCRIPT}.") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + def run(*args: str) -> subprocess.CompletedProcess[str]: """Execute the run_captured script with arguments and return the completed process.""" - return ( - subprocess.run( # noqa: S603 -- the interpreter and the script are this repository's own. - [PYTHON, str(SCRIPT), *args], - capture_output=True, - text=True, - check=False, - timeout=SCRIPT_TIMEOUT_S, - ) + return subprocess.run( # noqa: S603 # nosec B603 -- fixed interpreter and repository script. + [PYTHON, str(SCRIPT), *args], + capture_output=True, + text=True, + check=False, + timeout=SCRIPT_TIMEOUT_S, ) class RunCapturedTests(unittest.TestCase): + def assert_run_options(self, mock_run: mock.Mock, mod: ModuleType, cmd: list[str]) -> None: + """Check the bounded compile options passed by every in-process path.""" + options = mock_run.call_args.kwargs + self.assertEqual(mock_run.call_args.args, (cmd,)) + self.assertEqual(options["allowed_executables"], [cmd[0]]) + self.assertIs(options["stderr_to_stdout"], True) + self.assertEqual(options["max_output_bytes"], mod.MAX_COMPILE_OUTPUT_BYTES) + self.assertEqual(options["timeout_seconds"], mod.CC_TIMEOUT_S) + def test_success_prints_nothing(self) -> None: noisy = "import sys; print('warning: spilled'); print('more', file=sys.stderr)" result = run("--", PYTHON, "-c", noisy) @@ -46,6 +68,115 @@ def test_missing_separator_or_command_is_a_usage_error(self) -> None: with self.subTest(args=args): self.assertEqual(run(*args).returncode, 2) + def test_in_process_positive_records_timeout(self) -> None: + """Verify successful in-process run returns zero and records the configured timeout.""" + mod = load_run_captured_module() + fake_result = subprocess.CompletedProcess(["fake", "cmd"], 0, stdout="", stderr="") + with mock.patch.object(mod, "run_command", return_value=fake_result) as mock_run: + stdout_buf = io.StringIO() + stderr_buf = io.StringIO() + with contextlib.redirect_stdout(stdout_buf), contextlib.redirect_stderr(stderr_buf): + rc = mod.main(["--", "fake", "cmd"]) + self.assertEqual(rc, 0) + self.assert_run_options(mock_run, mod, ["fake", "cmd"]) + + def test_in_process_timeout_str_boundary(self) -> None: + """Verify string timeout output returns code 124 and echoes stdout and stderr.""" + mod = load_run_captured_module() + cmd = ["fake", "cmd"] + timeout_exc = mod.CommandTimedOut(tuple(cmd), mod.CC_TIMEOUT_S, "partial", None) + with mock.patch.object(mod, "run_command", side_effect=timeout_exc) as mock_run: + stdout_buf = io.StringIO() + stderr_buf = io.StringIO() + with contextlib.redirect_stdout(stdout_buf), contextlib.redirect_stderr(stderr_buf): + rc = mod.main(["--", *cmd]) + self.assertEqual(rc, 124) + self.assertIn("timed out after", stderr_buf.getvalue()) + self.assertIn(str(mod.CC_TIMEOUT_S), stderr_buf.getvalue()) + self.assertIn("CC_TIMEOUT_S", stderr_buf.getvalue()) + self.assertIn("partial", stdout_buf.getvalue()) + self.assert_run_options(mock_run, mod, cmd) + + def test_in_process_timeout_bytes_boundary(self) -> None: + """Verify byte timeout output decodes without error and returns code 124.""" + mod = load_run_captured_module() + cmd = ["fake", "cmd"] + timeout_exc = mod.CommandTimedOut(tuple(cmd), mod.CC_TIMEOUT_S, b"partial\xff", None) + with mock.patch.object(mod, "run_command", side_effect=timeout_exc) as mock_run: + stdout_buf = io.StringIO() + stderr_buf = io.StringIO() + with contextlib.redirect_stdout(stdout_buf), contextlib.redirect_stderr(stderr_buf): + rc = mod.main(["--", *cmd]) + self.assertEqual(rc, 124) + self.assertIn("partial", stdout_buf.getvalue()) + self.assert_run_options(mock_run, mod, cmd) + + def test_in_process_timeout_negative_empty_output(self) -> None: + """Verify timeout with None output returns code 124 and empty stdout.""" + mod = load_run_captured_module() + cmd = ["fake", "cmd"] + timeout_exc = mod.CommandTimedOut(tuple(cmd), mod.CC_TIMEOUT_S, None, None) + with mock.patch.object(mod, "run_command", side_effect=timeout_exc) as mock_run: + stdout_buf = io.StringIO() + stderr_buf = io.StringIO() + with contextlib.redirect_stdout(stdout_buf), contextlib.redirect_stderr(stderr_buf): + rc = mod.main(["--", *cmd]) + self.assertEqual(rc, 124) + self.assertEqual(stdout_buf.getvalue(), "") + self.assert_run_options(mock_run, mod, cmd) + + def test_in_process_output_limit_keeps_retained_stdout(self) -> None: + """Report the compile limit and retain diagnostics before returning code 125.""" + mod = load_run_captured_module() + cmd = ["fake", "cmd"] + limit_exc = mod.CommandOutputLimitExceeded( + tuple(cmd), mod.MAX_COMPILE_OUTPUT_BYTES, b"partial\xff", None + ) + with mock.patch.object(mod, "run_command", side_effect=limit_exc) as mock_run: + stdout_buf = io.StringIO() + stderr_buf = io.StringIO() + with contextlib.redirect_stdout(stdout_buf), contextlib.redirect_stderr(stderr_buf): + rc = mod.main(["--", *cmd]) + self.assertEqual(rc, 125) + self.assertEqual(stdout_buf.getvalue(), "partial\ufffd") + self.assertEqual( + stderr_buf.getvalue(), + "run_captured: command exceeded " + f"{mod.MAX_COMPILE_OUTPUT_BYTES} captured bytes (MAX_COMPILE_OUTPUT_BYTES): {cmd}\n", + ) + self.assert_run_options(mock_run, mod, cmd) + + def test_in_process_validation_failure_has_no_traceback(self) -> None: + """Show a missing executable as one diagnostic line and return code 126.""" + mod = load_run_captured_module() + cmd = ["missing", "cmd"] + reason = "executable is unavailable: missing" + with mock.patch.object( + mod, "run_command", side_effect=mod.CommandValidationError(reason) + ) as mock_run: + stdout_buf = io.StringIO() + stderr_buf = io.StringIO() + with contextlib.redirect_stdout(stdout_buf), contextlib.redirect_stderr(stderr_buf): + rc = mod.main(["--", *cmd]) + self.assertEqual(rc, 126) + self.assertEqual(stdout_buf.getvalue(), "") + self.assertEqual(stderr_buf.getvalue(), f"run_captured: {reason}\n") + self.assert_run_options(mock_run, mod, cmd) + + def test_in_process_launch_failure_has_no_traceback(self) -> None: + """Show an operating-system launch failure as one diagnostic line.""" + mod = load_run_captured_module() + reason = "launch failed" + with mock.patch.object(mod, "run_command", side_effect=OSError(reason)) as mock_run: + stdout_buf = io.StringIO() + stderr_buf = io.StringIO() + with contextlib.redirect_stdout(stdout_buf), contextlib.redirect_stderr(stderr_buf): + rc = mod.main(["--", "fake"]) + self.assertEqual(rc, 126) + self.assertEqual(stdout_buf.getvalue(), "") + self.assertEqual(stderr_buf.getvalue(), f"run_captured: {reason}\n") + self.assert_run_options(mock_run, mod, ["fake"]) + if __name__ == "__main__": unittest.main()