Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/ut/src/asys/common/cmd_run.py: 94%

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -*- 

3# ---------------------------------------------------------------------------- 

4# Copyright (c) 2025 Huawei Technologies Co., Ltd. 

5# 

6# Licensed under the Apache License, Version 2.0 (the "License"); 

7# you may not use this file except in compliance with the License. 

8# You may obtain a copy of the License at 

9# 

10# http://www.apache.org/licenses/LICENSE-2.0 

11# 

12# Unless required by applicable law or agreed to in writing, software 

13# distributed under the License is distributed on an "AS IS" BASIS, 

14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

15# See the License for the specific language governing permissions and 

16# limitations under the License. 

17# ---------------------------------------------------------------------------- 

18 

19import os 

20import platform 

21import subprocess 

22import sys 

23 

24from common.log import log_debug 

25 

26__all__ = ["run_command", "run_cmd_output", "check_command", "run_linux_cmd", "popen_run_cmd", "real_time_output",] 

27 

28 

29def get_os_type(): 

30 return platform.system() 

31 

32 

33def check_command(command): 

34 os_type = get_os_type() 

35 if os_type == "Windows": 

36 cmd = f"where {command}" 

37 elif os_type == "Linux": 

38 cmd = f"which {command}" 

39 else: 

40 log_debug("Unsupported operating system.") 

41 return False 

42 ret = subprocess.run(cmd, shell=True, capture_output=True, text=True) 

43 return ret.returncode == 0 

44 

45 

46def run_linux_cmd(cmd, cmp_str="") -> bool: 

47 if not isinstance(cmd, str): 

48 return False 

49 ret = subprocess.run(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) 

50 if cmp_str: 

51 return ret.stdout.strip().decode() == cmp_str 

52 if ret.returncode == 0: 

53 return True 

54 return False 

55 

56 

57def run_command(command) -> str: 

58 ret = subprocess.run(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding='utf-8', 

59 env=os.environ) 

60 if ret.returncode == 0: 

61 if ret.stderr != "": 

62 return 'NONE' 

63 return ret.stdout.strip() 

64 else: 

65 ret_err = ret.stderr 

66 log_debug('Run command: {0} failed, ret_code={1}, ret_err={2}'.format(command, ret.returncode, ret_err)) 

67 if 'not found' in ret_err: 

68 return 'NONE' 

69 return ret.stderr.strip().replace('\n', " ") 

70 

71 

72def run_cmd_output(command) -> [bool, str]: 

73 ret = subprocess.run(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding='utf-8', 

74 env=os.environ) 

75 if ret.returncode == 0: 

76 return True, ret.stdout 

77 else: 

78 ret_err = ret.stderr 

79 log_debug('Run command: {0} failed, ret_code={1}, ret_err={2}'.format(command, ret.returncode, ret_err)) 

80 return False, ret.stderr 

81 

82 

83def real_time_output(command, output=True) -> bool: 

84 process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, bufsize=1, 

85 universal_newlines=True, env=os.environ) 

86 if output: 

87 for line in process.stdout: 

88 sys.stdout.write(line) 

89 sys.stdout.flush() 

90 process.wait() 

91 return process.returncode == 0 

92 

93 

94class _IgnoreStderr: 

95 def __init__(self): 

96 self.null_fd = os.open(os.devnull, os.O_RDWR) 

97 self.save_fd = os.dup(2) 

98 

99 def __enter__(self): 

100 os.dup2(self.null_fd, 2) 

101 

102 def __exit__(self, *_): 

103 os.dup2(self.save_fd, 2) 

104 os.close(self.null_fd) 

105 

106 

107def popen_run_cmd(command): 

108 """ 

109 use the os.popen to run the command 

110 """ 

111 with _IgnoreStderr(): 

112 cmd = os.popen(command) 

113 ret = cmd.read() 

114 cmd.close() 

115 

116 return ret 

117