1#!/usr/bin/env python3

2"""

3 Copyright (C) 2020 Canonical Ltd.

4

5 Authors

6 Adrian Lane <adrian.lane@canonical.com>

7 Jeff Lane <jeff@ubuntu.com>

8 This program is free software: you can redistribute it and/or modify

9 it under the terms of the GNU General Public License version 3,

10 as published by the Free Software Foundation.

11

12 This program is distributed in the hope that it will be useful,

13 but WITHOUT ANY WARRANTY; without even the implied warranty of

14 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

15 GNU General Public License for more details.

16

17 You should have received a copy of the GNU General Public License

18 along with this program. If not, see <http://www.gnu.org/licenses/>.

19

20 Tests IPMI subsystem on SUT.

21"""

22

23 import re

24 import shutil

25 import sys

26 import platform

27 import argparse

28 import logging

29 from subprocess import Popen, check_call, PIPE, TimeoutExpired, SubprocessError

30

31

32 class FreeIpmiTest:

33

34 def __init__(self):

35 def get_path(binary):

36 """Get absolute path of FreeIPMI/nix binary,

37 warn upon failure.

38 """

39 path_full = shutil.which(binary)

40 if path_full:

41 return path_full

42 else:

43 logging.info("* Unable to stat absolute path for %s!" % binary)

44 return binary

45

46 # paths to load_kernel_module() binaries

47 self._path_lsmod = get_path("lsmod")

48 self._path_modprobe = get_path("modprobe")

49 # ipmi kernel modules to load/verify

50 self._kernel_modules = (

51 "ipmi_si",

52 "ipmi_devintf",

53 "ipmi_powernv",

54 "ipmi_ssif",

55 "ipmi_msghandler",

56 )

57 # method subprocess commands (FreeIPMI)

58 self._cmd_ipmi_chassis = [get_path("ipmi-chassis"), "--get-status"]

59 self._cmd_ipmi_channel = [

60 get_path("ipmi-config"),

61 "--checkout",

62 "-S",

63 "User1",

64 "--lan-channel-number",

65 ]

66 self._cmd_get_bmc_info = [get_path("bmc-info")]

67 self._cmd_ipmi_locate = [get_path("ipmi-locate")]

68 # min. ipmi version to pass

69 self._ipmi_ver = 2.0

70 # subprocess call timeout (s)

71 self._subproc_timeout = 15

72 # raised subproc exceptions to handle

73 # (decoupled from self._process_exc())

74 self._sub_process_excs = (

75 TimeoutExpired,

76 SubprocessError,

77 OSError,

78 TypeError,

79 FileNotFoundError,

80 )

81

82 def _subproc_logging(self, cmd):

83 """Subprocess stdin/stderr handling."""

84 process = Popen(cmd, stdout=PIPE, stderr=PIPE, universal_newlines=True)

85 output, error = process.communicate(timeout=self._subproc_timeout)

86 logging.debug("## Debug Output: ##")

87 if output:

88 logging.debug(" [Stdout]\n")

89 logging.debug("%s\n" % output)

90 if error:

91 logging.debug(" [Stderr]\n")

92 logging.debug("%s\n" % error)

93 logging.debug("## End Debug Output ##\n")

94 # ignore stderr

95 return output

96

97 def _process_exc(self, exc, test_method):

98 """Allows for bundling of exception handling for all

99 methods within this class.

100 """

101 if type(exc) is TimeoutExpired:

102 logging.info(

103 "* Timeout calling %s! (%ss)\n"

104 % (test_method, self._subproc_timeout)

105 )

106 elif type(exc) is FileNotFoundError:

107 logging.info(

108 "* Error calling %s! Check cmds/paths.\n" % test_method

109 )

110 else:

111 logging.info("* Error calling %s!\n" % test_method)

112

113 def _modprobe_hlpr(self, module):

114 """load_kernel_mods() helper function to call modprobe."""

115 try:

116 check_call(

117 [self._path_modprobe, module],

118 stderr=PIPE,

119 timeout=self._subproc_timeout,

120 )

121 except self._sub_process_excs:

122 logging.info("* Unable to load module %s!" % module)

123 logging.info(" **********************************************")

124 logging.info(" Warning: proceeding, but in-band IPMI may fail")

125 logging.info(" **********************************************")

126 else:

127 logging.info("- Successfully loaded module %s" % module)

128

129 def load_kernel_mods(self):

130 """Check (and load) kernel modules."""

131 logging.info("-----------------------")

132 logging.info("Verifying kernel modules:")

133 try:

134 output = self._subproc_logging(self._path_lsmod)

135 for module in self._kernel_modules:

136 if module in output:

137 logging.info("- %s already loaded" % module)

138 else:

139 if (

140 module == "ipmi_powernv"

141 and platform.machine() != "ppc64le"

142 ):

143 logging.info(

144 " * Skipping module %s, incorrect "

145 "system architecture" % module

146 )

147 else:

148 self._modprobe_hlpr(module)

149 logging.info("")

150 except self._sub_process_excs as exc:

151 self._process_exc(exc, self.load_kernel_mods.__qualname__)

152 return False

153

154 def get_impi_chassis(self):

155 """Get ipmi chassis data, pass if called w/o error."""

156 logging.info("-----------------------")

157 logging.info("Fetching chassis status:")

158 try:

159 self._subproc_logging(self._cmd_ipmi_chassis)

160 except self._sub_process_excs as exc:

161 self._process_exc(exc, self.get_impi_chassis.__qualname__)

162 return False

163 else:

164 logging.info("- Fetched chassis status!\n")

165 return True

166

167 def get_pwr_status(self):

168 """Get power status via ipmi chassis data,

169 pass if called w/o error & system power field present.

170 """

171 logging.info("-----------------------")

172 logging.info("Fetching power status:")

173 regex = re.compile(r"^System Power")

174 try:

175 output = self._subproc_logging(self._cmd_ipmi_chassis)

176 for line in output.rstrip().split("\n"):

177 if re.search(regex, line):

178 logging.info("- Fetched power status!\n")

179 return True

180 logging.info("* Unable to retrieve power status via IPMI.\n")

181 return False

182 except self._sub_process_excs as exc:

183 self._process_exc(exc, self.get_pwr_status.__qualname__)

184 return False

185

186 def get_bmc_info(self):

187 """Call bmc-info, pass if called w/o error."""

188 logging.info("-----------------------")

189 logging.info("Fetching BMC information:")

190 try:

191 self._subproc_logging(self._cmd_get_bmc_info)

192 except self._sub_process_excs as exc:

193 self._process_exc(exc, self.get_bmc_info.__qualname__)

194 return False

195 else:

196 logging.info("- Fetched BMC information!\n")

197 return True

198

199 def _ipmi_version_hlpr(self):

200 """Ipmi version discovery loop."""

201 regex = re.compile(r"^IPMI Version")

202 output = self._subproc_logging(self._cmd_get_bmc_info)

203 for line in output.rstrip().split("\n"):

204 if re.search(regex, line):

205 version = (line.split(":"))[1].strip()

206 return float(version)

207

208 def chk_ipmi_version(self):

209 """Fetch ipmi version via bmc-info sdout,

210 pass if ipmi version >= self._ipmi_ver.

211 """

212 logging.info("-----------------------")

213 logging.info("Validating IPMI version:")

214 try:

215 version = self._ipmi_version_hlpr()

216 logging.info("- IPMI version: %.1f" % version)

217 except self._sub_process_excs as exc:

218 self._process_exc(exc, self.chk_ipmi_version.__qualname__)

219 return False

220 else:

221 if version < float(self._ipmi_ver):

222 logging.info("* IPMI version below %d!\n" % self._ipmi_ver)

223 return False

224 else:

225 logging.info(" IPMI version compliant!\n")

226 return True

227

228 def _ipmi_channel_hlpr(self, i, regex, channel):

229 """get_ipmi_channel discovery loop."""

230 cmd = self._cmd_ipmi_channel

231 if len(cmd) > 5:

232 cmd.pop(-1)

233 cmd.append(str(i))

234 output = self._subproc_logging(cmd)

235 for line in output.rstrip().split("\n"):

236 if re.search(regex, line):

237 channel.append(i)

238 return channel

239

240 def get_ipmi_channel(self):

241 """Get ipmi channel(s) in use,

242 pass if user data returns after calling ipmi-config.

243 """

244 logging.info("-----------------------")

245 logging.info("Fetching IPMI channels:")

246 # support multiple channels

247 channel = []

248 regex = re.compile(r"Section User")

249 # test channels 0 - 15

250 for i in range(16):

251 # channel 12, 13 are invalid channels, skip

252 if i == (12 | 13):

253 continue

254 try:

255 self._ipmi_channel_hlpr(i, regex, channel)

256 except self._sub_process_excs as exc:

257 self._process_exc(exc, self.get_ipmi_channel.__qualname__)

258 return False

259 if channel:

260 logging.info("- Found %d channel(s)!" % len(channel))

261 logging.info(" IPMI channel(s): %s\n" % channel)

262 return True

263 else:

264 logging.info("* Unable to fetch IPMI channel!\n")

265 return False

266

267 def run_test(self):

268 """Initialize kernel modules, run ipmi tests."""

269 # load/val kernel modules

270 self.load_kernel_mods()

271 results = [

272 self.get_impi_chassis(),

273 self.get_pwr_status(),

274 self.get_bmc_info(),

275 self.chk_ipmi_version(),

276 self.get_ipmi_channel(),

277 ]

278 return results

279

280

281 def main():

282 # instantiate argparse as parser

283 parser = argparse.ArgumentParser()

284 parser.add_argument(

285 "-d",

286 "--debug",

287 action="store_true",

288 help="debug/verbose output (stdout/stderr)",

289 )

290 parser.add_argument(

291 "-q", "--quiet", action="store_true", help="suppress output"

292 )

293 args = parser.parse_args()

294 # init logging

295 if not args.quiet or args.debug:

296 logger = logging.getLogger()

297 logger.setLevel(logging.DEBUG)

298 if not args.quiet:

299 console_handler = logging.StreamHandler()

300 console_handler.setLevel(logging.INFO)

301 logger.addHandler(console_handler)

302 if args.debug:

303 console_handler.setLevel(logging.DEBUG)

304

305 print("## Running IPMI Tests ##")

306 # instantiate FreeIpmiTest as f_ipmi_test

307 f_ipmi_test = FreeIpmiTest()

308 results = f_ipmi_test.run_test()

309 results_dict = {

310 "Chassis": results[0],

311 "Power": results[1],

312 "BMC": results[2],

313 "Version": results[3],

314 "Channel": results[4],

315 }

316 # tally results

317 if sum(results) < len(results):

318 # transpose readable values into results_dict

319 for test, result in results_dict.items():

320 if result:

321 results_dict[test] = "Pass"

322 else:

323 results_dict[test] = "Fail"

324 print("-----------------------")

325 print("## IPMI tests failed! ##")

326 print("## %r ##" % results_dict)

327 return 1

328 else:

329 print("-----------------------")

330 print("## IPMI tests passed! ##")

331 return 0

332

333

334 if __name__ == "__main__":

335 sys.exit(main())