#!/usr/bin/env python # coding: utf-8 # Copyright (c) 2020 Huawei Technologies Co., Ltd. # oec-hardware is licensed under the Mulan PSL v2. # You can use this software according to the terms and conditions of the Mulan PSL v2. # You may obtain a copy of Mulan PSL v2 at: # http://license.coscl.org.cn/MulanPSL2 # THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR # PURPOSE. # See the Mulan PSL v2 for more details. # Create: 2020-04-01 import sys import re import subprocess class Command: def __init__(self, command): """ Creates a Command object that wraps the shell command """ self.command = command self.origin_output = None self.output = None self.errors = None self.returncode = 0 self.pipe = None self.regex = None self.single_line = True self.regex_group = None def _run(self): if sys.version_info.major < 3: self.pipe = subprocess.Popen(self.command, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) else: self.pipe = subprocess.Popen(self.command, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding='utf8') (output, errors) = self.pipe.communicate() if output: #Strip new line character/s if any from the end of output string output = output.rstrip('\n') self.origin_output = output self.output = output.splitlines() if errors: self.errors = errors.splitlines() self.returncode = self.pipe.returncode def start(self): if sys.version_info.major < 3: self.pipe = subprocess.Popen(self.command, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) else: self.pipe = subprocess.Popen(self.command, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding='utf8') def run(self, ignore_errors=False): """ run the command ignore_errors: do not raise exceptions """ self._run() if not ignore_errors: if self.returncode != 0: self.print_output() self.print_errors() raise CertCommandError(self, "returned %d" % self.returncode) if self.errors and len(self.errors) > 0: self.print_errors() # raise CertCommandError(self, "has output on stderr") def run_quiet(self): self._run() if self.returncode != 0: raise CertCommandError(self, "returned %d" % self.returncode) def echo(self, ignore_errors=False): self.run(ignore_errors) self.print_output() return def print_output(self): """ 结果显示 :return: """ if self.output: for line in self.output: sys.stdout.write(line) sys.stdout.write("\n") sys.stdout.flush() def print_errors(self): """ 页面显示错误信息 :return: """ if self.errors: for line in self.errors: sys.stderr.write(line) sys.stderr.write("\n") sys.stderr.flush() def pid(self): """ 获取管道pid值 :return: """ if self.pipe: return self.pipe.pid def readline(self): """ 按行读取输出信息 :return """ if self.pipe: return self.pipe.stdout.readline() def read(self): """ 执行命令，并读取结果 :return: """ self.pipe = subprocess.Popen(self.command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) if self.pipe: return self.pipe.stdout.read().decode('utf-8', 'ignore').rstrip() def poll(self): if self.pipe: return self.pipe.poll() def _get_str(self, regex=None, regex_group=None, single_line=True, return_list=False): self.regex = regex self.single_line = single_line self.regex_group = regex_group self._run() if self.single_line: if self.output and len(self.output) > 1: raise CertCommandError(self, "Found %u lines of output, expected 1" % len(self.output)) if self.output: line = self.output[0].strip() if not self.regex: return line # otherwise, try the regex pattern = re.compile(self.regex) match = pattern.match(line) if match: if self.regex_group: return match.group(self.regex_group) # otherwise, no group, return the whole line return line # no regex match try a grep-style match if not self.regex_group: match = pattern.search(line) if match: return match.group() # otherwise raise CertCommandError(self, "no match for regular expression %s" % self.regex) #otherwise, multi-line or single-line regex if not self.regex: raise CertCommandError(self, "no regular expression set for multi-line command") pattern = re.compile(self.regex) result = None if return_list: result = list() if self.output: for line in self.output: if self.regex_group: match = pattern.match(line) if match: if self.regex_group: if return_list: result.append(match.group(self.regex_group)) else: return match.group(self.regex_group) else: # otherwise, return the matching line match = pattern.search(line) if match: if return_list: result.append(match.group()) else: return match.group() if result: return result raise CertCommandError(self, "no match for regular expression %s" % self.regex) def get_str(self, regex=None, regex_group=None, single_line=True, return_list=False, ignore_errors=False): result = self._get_str(regex, regex_group, single_line, return_list) if not ignore_errors: if self.returncode != 0: self.print_output() self.print_errors() raise CertCommandError(self, "returned %d" % self.returncode) # if self.errors and len(self.errors) > 0: # raise CertCommandError(self, "has output on stderr") return result class CertCommandError(Exception): def __init__(self, command, message): super(CertCommandError, self).__init__() self.message = message self.command = command self.__message = None def __str__(self): return "\"%s\" %s" % (self.command.command, self.message) def _get_message(self): return self.__message def _set_message(self, value): self.__message = value message = property(_get_message, _set_message)