Coverage for hiperta_stream/utils/subprocess_utils.py: 33%

1import logging 

2import subprocess as sp 

3from subprocess import CalledProcessError 

4 

5 

6def subprocess_run_and_raise_exception_on_error( 

7 sp_cmd, success_log_string=None, failure_log_string=None, error_level=logging.CRITICAL, log_level=logging.INFO 

8): 

9 """Run a subprocess, and print its output in case of error. 

10 

11 Parameters 

12 ---------- 

13 sp_cmd : list of str 

14 The subprocess command. 

15 success_log_string : str, optional 

16 String to log when the subprocess succeeds. Defaults value is None, nothing is logged. 

17 failure_log_string : str, optional 

18 String to log when the subprocess fails. Defaults value is None, nothing is logged. 

19 

20 Raises 

21 ------ 

22 ChildProcessError 

23 If the subprocess returns a non-exit code. 

24 """ 

25 

26 try: 

27 completed_process = sp.run(sp_cmd, capture_output=True, text=True, check=True) 

28 if success_log_string is not None: 

29 logging.log(log_level, success_log_string) 

30 except CalledProcessError as error: 

31 logging.log( 

32 error_level, 

33 failure_log_string 

34 + "\nSubprocess stdout: " 

35 + str(error.stdout) 

36 + "\nSubprocess stderr: " 

37 + str(error.stderr), 

38 exc_info=True, 

39 ) 

40 raise 

41 return completed_process