Merge pull request #202 from lukpueh/add-default-kwarg-helper

Add default timeout setting helper to process module

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 = <seconds>
-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):
   """
   <Purpose>
     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()):
   """
   <Purpose>
     Provide a function that executes a command in a subprocess and, upon

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()