(cmd,
log_file_path: str,
timeout: int = 3600,
env=None,
should_print=True)
| 4 | |
| 5 | |
| 6 | def execute_command_with_logging(cmd, |
| 7 | log_file_path: str, |
| 8 | timeout: int = 3600, |
| 9 | env=None, |
| 10 | should_print=True) -> tuple[bool, str]: |
| 11 | if env is None: |
| 12 | env = os.environ.copy() |
| 13 | |
| 14 | if os.path.isfile(log_file_path): |
| 15 | write_type = 'a' |
| 16 | else: |
| 17 | write_type = 'w' |
| 18 | try: |
| 19 | result = True |
| 20 | with open(log_file_path, write_type, encoding='utf-8') as log_file: |
| 21 | start_msg = f'execute command: {cmd}\n' |
| 22 | print(start_msg, end='') |
| 23 | log_file.write(start_msg) |
| 24 | log_file.flush() |
| 25 | |
| 26 | process = subprocess.run(cmd, |
| 27 | shell=True, |
| 28 | text=True, |
| 29 | encoding='utf-8', |
| 30 | errors='replace', |
| 31 | stdout=subprocess.PIPE, |
| 32 | stderr=subprocess.STDOUT, |
| 33 | env=env, |
| 34 | bufsize=1, |
| 35 | timeout=timeout, |
| 36 | start_new_session=True) |
| 37 | |
| 38 | if process.stdout: |
| 39 | if should_print: |
| 40 | print(process.stdout, end='') |
| 41 | log_file.write(process.stdout) |
| 42 | |
| 43 | if process.returncode == 0: |
| 44 | result_msg = 'execute command success!\n' |
| 45 | else: |
| 46 | result = False |
| 47 | result_msg = f'execute command fail: {process.returncode}\n' |
| 48 | |
| 49 | log_file.write(result_msg) |
| 50 | |
| 51 | return result, result_msg.strip() |
| 52 | |
| 53 | except Exception as e: |
| 54 | error_msg = f'execute command fail exception: {str(e)}\n' |
| 55 | print(error_msg, file=sys.stderr, end='') |
| 56 | |
| 57 | with open(log_file_path, 'a', encoding='utf-8') as log_file: |
| 58 | log_file.write(error_msg) |
| 59 | |
| 60 | return False, error_msg.strip() |
no test coverage detected