Refactor to oop, add argparsing & logging (debug/quiet). Implemeneted feedback from Jeff Lane.

1 files changed, 250 insertions, 247 deletions

diff --git a/bin/ipmi_test.py b/bin/ipmi_test.py
index 8e14b998..9de4b5d8 100755
--- a/bin/ipmi_test.py
+++ b/bin/ipmi_test.py
@@ -1,277 +1,280 @@
 #!/usr/bin/env python3
-# Copyright 2020 Canonical Ltd.
-# All rights reserved.
-#
-# Written by:
-# Adrian Lane <adrian.lane@canonical.com>
+"""
+Copyright (C) 2020 Canonical Ltd.
 
-import subprocess
-import re
-import os
-import time
+Authors
+ Adrian Lane <adrian.lane@canonical.com>
 
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU General Public License version 3,
+as published by the Free Software Foundation.
 
-def main():
- # globals
- # using relative paths
- start_time = time.clock()
- path_lsmod = 'lsmod'
- path_modprobe = 'modprobe'
- kernel_modules = (
- 'ipmi_si',
- 'ipmi_devintf',
- 'ipmi_powernv',
- 'ipmi_ssif',
- 'ipmi_msghandler')
- path_ipmi_chassis = 'ipmi-chassis'
- path_ipmi_config = 'ipmi-config'
- path_bmc_info = 'bmc-info'
- path_ipmi_locate = 'ipmi-locate'
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+GNU General Public License for more details.
 
- print ('Running IPMI tests...')
- # check kernel modules
- kernel_mods(path_lsmod, path_modprobe, kernel_modules)
- # tally results
- results = []
- results.append(impi_chassis(path_ipmi_chassis))
- results.append(pwr_status(path_ipmi_chassis))
- results.append(ipmi_channel(path_ipmi_config))
- results.append(bmc_info(path_bmc_info))
- results.append(ipmi_version(path_bmc_info))
- results.append(ipmi_locate(path_ipmi_locate))
+You should have received a copy of the GNU General Public License
+along with this program. If not, see <http://www.gnu.org/licenses/>.
 
- if sum(results) > 0:
- end_time = time.clock()
- total_time = "{0:.4f}".format(end_time - start_time)
- print ('-----------------------')
- print ('## IPMI tests failed! ##')
- print (
- f'## Chassis: {results[0]} Power: {results[1]} '
- f'Channel: {results[2]} BMC: {results[3]} '
- f'IPMI Version: {results[4]} IPMI Locate: {results[5]} ##')
- print (f'## Total time: {total_time}s ##')
- else:
- end_time = time.clock()
- total_time = "{0:.4f}".format(end_time - start_time)
- print ('-----------------------')
- print ('## IPMI tests passed! ##')
- print (f'## Total time: {total_time}s ##')
+Tests IPMI subsystem on SUT.
+"""
 
+import re
+import os
+import shutil
+import sys
+import argparse
+import logging
+from subprocess import (
+ Popen,
+ check_call,
+ PIPE,
+ SubprocessError,
+ TimeoutExpired,
+ SubprocessError)
 
-def kernel_mods(path_lsmod, path_modprobe, kernel_modules):
- print('-----------------------')
- print ('Verifying kernel modules:')
- try:
- process = subprocess.Popen(
- [path_lsmod],
- stdout=subprocess.PIPE, stderr=subprocess.PIPE)
- output, error = process.communicate(timeout=15)
- for module in kernel_modules:
- if module in output.decode('utf-8'):
- print (f'- {module} already loaded.')
- else:
- try:
- subprocess.check_call(
- [path_modprobe, module],
- stderr=subprocess.PIPE, timeout=15)
- print (f'- Successfully loaded module {module}')
- except subprocess.TimeoutExpired:
- print (f'Timeout ({e.timeout}s) calling modprobe!')
- except subprocess.CalledProcessError:
- print (' *******************************************')
- print (f' WARNING: Unable to load module {module}')
- print (' Continuing run, but in-band IPMI may fail')
- print (' *******************************************')
- except EnvironmentError:
- print ('Unable to invoke modprobe!\n')
- print ('')
- except subprocess.TimeoutExpired as e:
- print (f'Timeout ({e.timeout}s) calling lsmod!')
- except subprocess.SubprocessError:
- # fail if true?
- print ('Error calling lsmod!\n')
- except EnvironmentError:
- # fail if true?
- print ('Unable to invoke lsmod!\n')
 
+class IpmiTest(object):
 
-def impi_chassis(path_ipmi_chassis):
- print('-----------------------')
- print('Fetching chassis status:')
- start_time = time.clock()
- try:
- fnull = open(os.devnull, 'w')
- subprocess.check_call(
- [path_ipmi_chassis, '--get-status'],
- stdout=fnull, stderr=subprocess.PIPE, timeout=15)
- end_time = time.clock()
- total_time = "{0:.4f}".format(end_time - start_time)
- print('Successfully fetched chassis status!')
- print (f'(took {total_time}s)\n')
- return 0
- except subprocess.TimeoutExpired as e:
- print (f'Timeout ({e.timeout}s) fetching chassis status!\n')
- return 1
- except subprocess.CalledProcessError:
- print ('Error calling ipmi_chassis() subprocess!\n')
- return 1
- except EnvironmentError:
- print ('Unable to invoke ipmi-chassis!\n')
- return 1
+ def __init__(self):
+ self.path_lsmod = shutil.which('lsmod')
+ self.path_modprobe = shutil.which('modprobe')
+ self.kernel_modules = (
+ 'ipmi_si',
+ 'ipmi_devintf',
+ 'ipmi_powernv',
+ 'ipmi_ssif',
+ 'ipmi_msghandler')
+ self.path_ipmi_chassis = shutil.which('ipmi-chassis')
+ self.path_ipmi_config = shutil.which('ipmi-config')
+ self.path_bmc_info = shutil.which('bmc-info')
+ self.path_ipmi_locate = shutil.which('ipmi-locate')
+ # min. ipmi version to pass
+ self.ipmi_ver = 2.0
+ # subprocess call timeout (s)
+ self.subproc_timeout = 10
 
+ def run_test(self):
+ # load/val kernel modules
+ self.kernel_mods()
+ # tally results
+ results = [self.impi_chassis(),
+ self.pwr_status(),
+ self.ipmi_channel(),
+ self.bmc_info(),
+ self.ipmi_version(),
+ self.ipmi_locate()]
+ return results
 
-def pwr_status(path_ipmi_chassis):
- print('-----------------------')
- print('Fetching power status:')
- start_time = time.clock()
- regex = re.compile('^System Power')
- try:
- process = subprocess.Popen(
- [path_ipmi_chassis, '--get-status'],
- stdout=subprocess.PIPE, stderr=subprocess.PIPE)
- output, error = process.communicate(timeout=15)
- output = output.decode('utf-8')
- for line in output.rstrip().split('\n'):
- if re.search(regex, line):
- end_time = time.clock()
- total_time = "{0:.4f}".format(end_time - start_time)
- print ('Successfully fetched power status!')
- print (f'(took {total_time}s)\n')
- return 0
+ def subproc_logging(self, cmd):
+ process = Popen(
+ cmd, stdout=PIPE, stderr=PIPE, universal_newlines=True)
+ output, error = process.communicate(timeout=self.subproc_timeout)
+ logging.debug('\n## Debug Output: ##')
+ if len(output) > 0:
+ # padding
+ logging.debug(' [Stdout]\n')
+ logging.debug(f'{output}\n')
+ if len(error) > 0:
+ # padding
+ logging.debug(' [Stderr]\n')
+ logging.debug(f'{error}\n')
+ logging.debug('## End Debug Output ##\n')
+ return output
+
+ def proc_ex(self, subtest):
+ if TimeoutExpired:
+ logging.info(
+ f'Timeout calling {subtest}!'
+ f' ({self.subproc_timeout}s)\n')
 else:
- print('Unable to retrieve power status via IPMI.\n')
+ logging.info(f'Error calling {subtest}!\n')
+
+ def modprobe_hlpr(self, module):
+ cmd = [self.path_modprobe, module]
+ try:
+ check_call(
+ [self.path_modprobe, module],
+ stderr=PIPE, timeout=self.subproc_timeout)
+ except (TimeoutExpired, SubprocessError, OSError):
+ logging.info(f'* Unable to load module {module}!')
+ logging.info(' **********************************************')
+ logging.info(f' Warning: proceeding, but in-band IPMI may fail')
+ logging.info(' **********************************************')
+ else:
+ logging.info(f'- Successfully loaded module {module}')
+
+ def kernel_mods(self):
+ logging.info('-----------------------')
+ logging.info('Verifying kernel modules:')
+ cmd = [self.path_lsmod]
+ try:
+ output = self.subproc_logging(cmd)
+ for module in self.kernel_modules:
+ if module in output:
+ logging.info(f'- {module} already loaded')
+ else:
+ self.modprobe_hlpr(module)
+ logging.info('')
+ except (TimeoutExpired, SubprocessError, OSError):
+ self.proc_ex('lsmod')
+
+ def impi_chassis(self):
+ logging.info('-----------------------')
+ logging.info('Fetching chassis status:')
+ cmd = [self.path_ipmi_chassis, '--get-status']
+ try:
+ self.subproc_logging(cmd)
+ except (TimeoutExpired, SubprocessError, OSError):
+ self.proc_ex('ipmi_chassis()')
 return 1
- except subprocess.TimeoutExpired as e:
- print (f'Timeout ({e.timeout}s) fetching power status!\n')
- return 1
- except subprocess.SubprocessError:
- print ('Error calling pwr_status() subprocess!\n')
- return 1
- except EnvironmentError:
- print ('Unable to invoke ipmi-chassis!\n')
- return 1
+ else:
+ logging.info('Fetched chassis status!\n')
+ return 0
 
+ def pwr_status(self):
+ logging.info('-----------------------')
+ logging.info('Fetching power status:')
+ regex = re.compile('^System Power')
+ cmd = [self.path_ipmi_chassis, '--get-status']
+ try:
+ output = self.subproc_logging(cmd)
+ for line in output.rstrip().split('\n'):
+ if re.search(regex, line):
+ logging.info('Fetched power status!\n')
+ return 0
+ else:
+ logging.info('Unable to retrieve power status via IPMI.\n')
+ return 1
+ except (TimeoutExpired, SubprocessError, OSError):
+ self.proc_ex('pwr_status()')
+ return 1
 
-def ipmi_channel(path_ipmi_config):
- print('-----------------------')
- print('Fetching IPMI channel:')
- start_time = time.clock()
- regex = re.compile('Section User')
- matches = 0
- channel = []
- try:
+ def ipmi_channel(self):
+ logging.info('-----------------------')
+ logging.info('Fetching IPMI channel:')
+ regex = re.compile('Section User')
+ matches = 0
+ # support multiple channels
+ channel = []
 # test channels 0 - 15
- for i in range(15):
- process = subprocess.Popen(
- [path_ipmi_config, '--checkout', '--lan-channel-number', str(i)],
- stdout=subprocess.PIPE, stderr=subprocess.PIPE)
- output, error = process.communicate(timeout=15)
- if re.search(regex, output.decode('utf-8')):
- matches += 1
- channel.append(i)
- if matches > 0:
- end_time = time.clock()
- total_time = "{0:.4f}".format(end_time - start_time)
- print ('IPMI Channel(s):', channel)
- print (f'(took {total_time}s)\n')
- return 0
+ try:
+ for i in range(15):
+ cmd = [self.path_ipmi_config, '--checkout',
+ '--lan-channel-number', str(i)]
+ output = self.subproc_logging(cmd)
+ for line in output.rstrip().split('\n'):
+ if re.search(regex, line):
+ matches += 1
+ channel.append(i)
+ break
+ except (TimeoutExpired, SubprocessError, OSError):
+ self.proc_ex('ipmi_channel()')
+ return 1
+ else:
+ if matches > 0:
+ logging.info(f'IPMI Channel(s): {channel}\n')
+ return 0
+ else:
+ logging.info('Unable to fetch IPMI channel!\n')
+ return 1
+
+ def bmc_info(self):
+ logging.info('-----------------------')
+ logging.info('Fetching BMC information:')
+ cmd = [self.path_bmc_info]
+ try:
+ self.subproc_logging(cmd)
+ except (SubprocessError, OSError, TimeoutExpired):
+ self.proc_ex('bmc_info()')
+ return 1
 else:
- print ('Unable to fetch IPMI channel!')
+ logging.info('Fetched BMC information!\n')
+ return 0
+
+ def ipmi_version(self):
+ logging.info('-----------------------')
+ logging.info('Testing IPMI version:')
+ cmd = [self.path_bmc_info]
+ try:
+ output = self.subproc_logging(cmd)
+ # Prefer .index() over .find() for exception handling
+ res_index = output.index('IPMI Version')
+ version = output[(res_index + 24):(res_index + 27)]
+ logging.info(f'IPMI Version: {version}\n')
+ if float(version) < self.ipmi_ver:
+ logging.info(f'IPMI Version below {self.ipmi_ver}!\n')
+ return 1
+ else:
+ return 0
+ except (TimeoutExpired, SubprocessError, OSError,):
+ self.proc_ex('ipmi_version()')
 return 1
- except subprocess.TimeoutExpired as e:
- print (f'Timeout ({e.timeout}s) fetching IPMI channel!\n')
- return 1
- except subprocess.SubprocessError:
- print ('Error calling ipmi_channel() subprocess!\n')
- return 1
- except EnvironmentError:
- print ('Unable to invoke ipmi-config!\n')
- return 1
 
+ def ipmi_locate(self):
+ logging.info('-----------------------')
+ logging.info('Testing ipmi-locate:')
+ regex = re.compile('driver:')
+ cmd = [self.path_ipmi_locate]
+ try:
+ output = self.subproc_logging(cmd)
+ if re.search(regex, output):
+ logging.info('Located IPMI driver!\n')
+ return 0
+ else:
+ logging.info('Unable to locate IPMI driver!\n')
+ return 1
+ except (TimeoutExpired, SubprocessError, OSError):
+ self.proc_ex('ipmi_locate()')
+ return 1
 
-def bmc_info(path_bmc_info):
- print('-----------------------')
- print('Fetching BMC information:')
- start_time = time.clock()
- try:
- fnull = open(os.devnull, 'w')
- subprocess.check_call(
- [path_bmc_info],
- stdout=fnull, stderr=subprocess.PIPE, timeout=15)
- end_time = time.clock()
- total_time = "{0:.4f}".format(end_time - start_time)
- print('Successfully fetched chassis status!')
- print (f'(took {total_time}s)\n')
- return 0
- except subprocess.TimeoutExpired as e:
- print (f'Timeout ({e.timeout}s) fetching BMC information!\n')
- return 1
- except subprocess.CalledProcessError:
- print ('Error calling bmc-info() subprocess!\n')
- return 1
- except EnvironmentError:
- print ('Unable to invoke bmc-info!\n')
- return 1
 
+def main():
+ # instantiate argparse as parser
+ parser = argparse.ArgumentParser()
+ parser.add_argument("-d", "--debug", action="store_true",
+ help="Debug/verbose output (stdout/stderr)")
+ parser.add_argument("-q", "--quiet", action="store_true",
+ help="Suppress output.")
+ args = parser.parse_args()
 
-def ipmi_version(path_bmc_info):
- print('-----------------------')
- print('Testing IPMI version:')
- start_time = time.clock()
- try:
- process = subprocess.Popen(
- [path_bmc_info],
- stdout=subprocess.PIPE, stderr=subprocess.PIPE)
- output, error = process.communicate(timeout=15)
- output = output.decode('utf-8')
- # Prefer .index() over .find() for exception handling
- res_index = output.index('IPMI Version')
- version = output[(res_index + 24):(res_index + 27)]
- print ('IPMI Version:', version)
- if float(version) < 2.0:
- print ('IPMI Version below 2.0!')
- return 1
- else:
- end_time = time.clock()
- total_time = "{0:.4f}".format(end_time - start_time)
- print (f'(took {total_time}s)\n')
- return 0
- except subprocess.TimeoutExpired as e:
- print (f'Timeout ({e.timeout}s) fetching IPMI version!\n')
- return 1
- except subprocess.SubprocessError:
- print ('Error calling ipmi_version() subprocess!\n')
- return 1
- except EnvironmentError:
- print ('Unable to invoke bmc-info!\n')
- return 1
+ # Set up the logging system
+ if not args.quiet or args.debug:
+ logger = logging.getLogger()
+ logger.setLevel(logging.DEBUG)
+ format = ''
+ date_format = '%Y-%m-%d %H:%M:%S'
 
+ # If we DO want console output
+ if not args.quiet:
+ console_handler = logging.StreamHandler()
+ console_handler.setFormatter(
+ logging.Formatter(format, date_format))
+ console_handler.setLevel(logging.INFO)
+ logger.addHandler(console_handler)
 
-def ipmi_locate(path_ipmi_locate):
- print('-----------------------')
- print('Testing ipmi-locate:')
- start_time = time.clock()
- try:
- fnull = open(os.devnull, 'w')
- subprocess.check_call(
- [path_ipmi_locate],
- stdout=fnull, stderr=subprocess.PIPE, timeout=15)
- end_time = time.clock()
- total_time = "{0:.4f}".format(end_time - start_time)
- print('Successfully called ipmi-locate!')
- print (f'(took {total_time}s)\n')
- return 0
- except subprocess.TimeoutExpired as e:
- print (f'Timeout ({e.timeout}s) testing impmi-locate!\n')
- return 1
- except subprocess.CalledProcessError:
- print ('Error calling impi_locate() subprocess!\n')
- return 1
- except EnvironmentError:
- print ('Unable to invoke ipmi-locate!\n')
+ if args.debug:
+ console_handler.setLevel(logging.DEBUG)
+
+ # instantiate IpmiTest as results for post-processing
+ ipmi_test = IpmiTest()
+ results = ipmi_test.run_test()
+ # tally results
+ if sum(results) > 0:
+ print ('-----------------------')
+ print ('## IPMI tests failed! ##')
+ print (
+ f'## Chassis: {results[0]} Power: {results[1]} ',
+ f'Channel: {results[2]} BMC: {results[3]} ',
+ f'IPMI Version: {results[4]} IPMI Locate: {results[5]} ##')
 return 1
+ else:
+ print ('-----------------------')
+ print ('## IPMI tests passed! ##')
+ return 0
 
 
 # call main()
 if __name__ == '__main__':
- main()
+ sys.exit(main())