diff --git a/securesystemslib/process.py b/securesystemslib/process.py index 245aa952..26dd8f44 100644 --- a/securesystemslib/process.py +++ b/securesystemslib/process.py @@ -37,19 +37,22 @@ import subprocess import securesystemslib.formats - +import securesystemslib.settings DEVNULL = subprocess.DEVNULL PIPE = subprocess.PIPE -# NOTE: If changed programatically, please do via this process module, e.g. -# securesystemslib.process.SUBPROCESS_TIMEOUT = -from securesystemslib.settings import SUBPROCESS_TIMEOUT - log = logging.getLogger(__name__) +def _default_timeout(): + """Helper to use securesystemslib.settings.SUBPROCESS_TIMEOUT as default + argument, and still be able to modify it after the function definitions are + evaluated. """ + return securesystemslib.settings.SUBPROCESS_TIMEOUT + + -def run(cmd, check=True, timeout=SUBPROCESS_TIMEOUT, **kwargs): +def run(cmd, check=True, timeout=_default_timeout(), **kwargs): """ Provide wrapper for `subprocess.run` (see @@ -126,7 +129,7 @@ def run(cmd, check=True, timeout=SUBPROCESS_TIMEOUT, **kwargs): -def run_duplicate_streams(cmd, timeout=SUBPROCESS_TIMEOUT): +def run_duplicate_streams(cmd, timeout=_default_timeout()): """ Provide a function that executes a command in a subprocess and, upon diff --git a/tests/test_process.py b/tests/test_process.py index 6491e4e2..25d645a7 100644 --- a/tests/test_process.py +++ b/tests/test_process.py @@ -23,6 +23,7 @@ import io import sys import securesystemslib.process +import securesystemslib.settings class Test_Process(unittest.TestCase): @@ -121,6 +122,20 @@ def test_run_duplicate_streams_timeout(self): timeout=-1) + def test__default_timeout(self): + """Test default timeout modification. """ + # Backup timeout and check that it is what's returned by _default_timeout() + timeout_old = securesystemslib.settings.SUBPROCESS_TIMEOUT + self.assertEqual(securesystemslib.process._default_timeout(), timeout_old) + + # Modify timeout and check that _default_timeout() returns the same value + timeout_new = timeout_old + 1 + securesystemslib.settings.SUBPROCESS_TIMEOUT = timeout_new + self.assertEqual(securesystemslib.process._default_timeout(), timeout_new) + + # Restore original timeout + securesystemslib.settings.SUBPROCESS_TIMEOUT = timeout_old + if __name__ == "__main__": unittest.main()