Coverage for biobb_common/biobb_common/command_wrapper/cmd_wrapper.py: 66%

1# -*- coding: utf-8 -*- 

2"""Python wrapper for command line 

3""" 

4import os 

5import subprocess 

6from biobb_common.tools import file_utils as fu 

7from typing import Optional, Union 

8import logging 

9from pathlib import Path 

10 

11 

12class CmdWrapper: 

13 """Command line wrapper using subprocess library 

14 """ 

15 

16 def __init__(self, cmd: list[str], shell_path: Union[str, Path] = os.getenv('SHELL', '/bin/sh'), out_log: Optional[logging.Logger] = None, err_log: Optional[logging.Logger] = None, 

17 global_log: Optional[logging.Logger] = None, env: Optional[dict] = None, timeout: Optional[int] = None) -> None: 

18 

19 self.cmd = cmd 

20 self.shell_path = shell_path 

21 self.out_log = out_log 

22 self.err_log = err_log 

23 self.global_log = global_log 

24 self.env = env 

25 self.timeout = timeout 

26 

27 def log_output(self, exit_code: str, command: str, out: Optional[bytes] = None, err: Optional[bytes] = None, timeout: Optional[str] = None, 

28 out_log: Optional[logging.Logger] = None, err_log: Optional[logging.Logger] = None, global_log: Optional[logging.Logger] = None) -> None: 

29 

30 timeout_str = '' 

31 if timeout: 

32 timeout_str = f"Timeout: {timeout} seconds expired, killing process\n" 

33 command_str = f"Executing: {command[0:80]}..." 

34 exit_code_str = f"Exit code: {exit_code}" 

35 if out_log: 

36 out_log.info(command_str) 

37 out_log.info(exit_code_str) 

38 if timeout_str: 

39 out_log.info(timeout_str) 

40 if out: 

41 out_log.info(out.decode("utf-8")) 

42 else: 

43 print(command_str) 

44 print(exit_code_str) 

45 if timeout_str: 

46 print(timeout_str) 

47 print("") 

48 if err_log and err: 

49 err_log.info(err.decode("utf-8")) 

50 

51 if global_log: 

52 global_log.info(f"{fu.get_logs_prefix()}{command_str}") 

53 global_log.info(f"{fu.get_logs_prefix()}{exit_code_str}") 

54 if timeout_str: 

55 global_log.info(f"{fu.get_logs_prefix()}{timeout_str}") 

56 

57 def launch(self) -> int: 

58 cmd = " ".join(self.cmd) 

59 if self.out_log is None: 

60 print('') 

61 print("cmd_wrapper commnand print: " + cmd) 

62 else: 

63 self.out_log.info(cmd + '\n') 

64 

65 new_env = {**os.environ.copy(), **self.env} if self.env else os.environ.copy() 

66 process = subprocess.Popen(cmd, 

67 stdout=subprocess.PIPE, 

68 stderr=subprocess.PIPE, 

69 shell=True, 

70 executable=self.shell_path, 

71 env=new_env) 

72 try: 

73 out, err = process.communicate(timeout=self.timeout) 

74 except subprocess.TimeoutExpired: 

75 process.kill() 

76 out, err = process.communicate() 

77 process.returncode = 1 

78 self.log_output(exit_code=str(process.returncode), command=" ".join(self.cmd), out=out, err=err, timeout=str(self.timeout), out_log=self.out_log, err_log=self.err_log, global_log=self.global_log) 

79 return process.returncode 

80 

81 process.wait() 

82 self.log_output(exit_code=str(process.returncode), command=" ".join(self.cmd), out=out, err=err, out_log=self.out_log, err_log=self.err_log, global_log=self.global_log) 

83 return process.returncode