1#!/usr/bin/env python3

2 import os

3 import sys

4 import re

5 import difflib

6 import logging

7 from argparse import ArgumentParser

8

9# Script return codes

10 SUCCESS = 0

11 NOT_MATCH = 1

12 NOT_FINISHED = 2

13 NOT_FOUND = 3

14

15

16 def main():

17 args = parse_args()

18

19 if not os.path.isfile(args.input_log_filename):

20 sys.stderr.write(

21 "Log file {0!r} not found\n".format(args.input_log_filename)

22 )

23 sys.exit(NOT_FOUND)

24

25 LoggingConfiguration.set(args.log_level, args.output_log_filename)

26 parser = Parser(args.input_log_filename)

27 results = parser.parse()

28

29 if not compare_results(results):

30 sys.exit(NOT_MATCH)

31

32 sys.exit(SUCCESS)

33

34

35 class Parser(object):

36 """

37 Reboot test log file parser

38 """

39

40 is_logging_line = re.compile(

41 r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3}"

42 ).search

43 is_getting_info_line = re.compile(

44 r"Gathering hardware information...$"

45 ).search

46 is_executing_line = re.compile(r"Executing: '(?P<command>.*)'...$").search

47 is_output_line = re.compile(r"Output:$").search

48 is_field_line = re.compile(

49 r"^- (?P<field>returncode|stdout|stderr):$"

50 ).match

51 is_test_complete_line = re.compile(r"test complete$").search

52

53 def __init__(self, filename):

54 self.filename = filename

55

56 def parse(self):

57 """

58 Parse log file and return results

59 """

60 with open(self.filename) as f:

61 results = self._parse_file(LineIterator(f))

62 return results

63

64 def _parse_file(self, iterator):

65 """

66 Parse all lines in iterator and return results

67 """

68 results = []

69 result = {}

70

71 for line in iterator:

72 if self.is_getting_info_line(line):

73 if result:

74 # Add last result to list of results

75 results.append(result)

76

77 # Initialize for a new iteration results

78 result = {}

79

80 match = self.is_executing_line(line)

81 if match:

82 command = match.group("command")

83 command_output = self._parse_command_output(iterator)

84

85 if command_output is not None:

86 result[command] = command_output

87 else:

88 if result:

89 # Add last result to list of results

90 results.append(result)

91

92 if not self.is_test_complete_line(line):

93 sys.stderr.write("Test didn't finish properly according to logs\n")

94 sys.exit(NOT_FINISHED)

95

96 return results

97

98 def _parse_command_output(self, iterator):

99 """

100 Parse one command output

101 """

102 command_output = None

103

104 # Skip all lines until command output is found

105 for line in iterator:

106 if self.is_output_line(line):

107 command_output = {}

108 break

109 if self.is_executing_line(line) or self.is_getting_info_line(line):

110 # Skip commands with no output

111 iterator.unnext(line)

112 return None

113

114 # Parse command output message

115 for line in iterator:

116 match = self.is_field_line(line)

117 if match:

118 field = match.group("field")

119 value = self._parse_command_output_field(iterator)

120 command_output[field] = value

121 # Exit when all command output fields

122 # have been gathered

123 else:

124 iterator.unnext(line)

125 break

126

127 return command_output

128

129 def _parse_command_output_field(self, iterator):

130 """

131 Parse one command output field

132 """

133 # Accummulate as many lines as needed

134 # for the field value

135 value = []

136 for line in iterator:

137 if self.is_logging_line(line) or self.is_field_line(line):

138 iterator.unnext(line)

139 break

140

141 value.append(line)

142

143 value = "".join(value)

144 return value

145

146

147 class LineIterator:

148 """

149 Iterator wrapper to make it possible

150 to push back lines that shouldn't have been consumed

151 """

152

153 def __init__(self, iterator):

154 self.iterator = iterator

155 self.buffer = []

156

157 def __iter__(self):

158 return self

159

160 def __next__(self):

161 if self.buffer:

162 return self.buffer.pop()

163

164 return next(self.iterator)

165

166 def unnext(self, line):

167 self.buffer.append(line)

168

169

170 class LoggingConfiguration(object):

171 @classmethod

172 def set(cls, log_level, log_filename):

173 """

174 Configure a rotating file logger

175 """

176 logger = logging.getLogger()

177 logger.setLevel(logging.DEBUG)

178

179 # Log to sys.stderr using log level passed through command line

180 if log_level != logging.NOTSET:

181 log_handler = logging.StreamHandler()

182 formatter = logging.Formatter("%(levelname)-8s %(message)s")

183 log_handler.setFormatter(formatter)

184 log_handler.setLevel(log_level)

185 logger.addHandler(log_handler)

186

187 # Log to rotating file using DEBUG log level

188 log_handler = logging.FileHandler(log_filename, mode="w")

189 formatter = logging.Formatter(

190 "%(asctime)s %(levelname)-8s " "%(message)s"

191 )

192 log_handler.setFormatter(formatter)

193 log_handler.setLevel(logging.DEBUG)

194 logger.addHandler(log_handler)

195

196

197 def compare_results(results):

198 """

199 Compare results using first one as a baseline

200 """

201 baseline = results[0]

202

203 success = True

204 for index, result in enumerate(results[1:]):

205 for command in baseline.keys():

206 baseline_output = baseline[command]

207 result_output = result[command]

208

209 error_messages = []

210 fields = set(baseline_output.keys()) | set(result_output.keys())

211 for field in fields:

212 baseline_field = baseline_output.get(field, "")

213 result_field = result_output.get(field, "")

214

215 if baseline_field != result_field:

216 differ = difflib.Differ()

217

218 message = [

219 "** {field!r} field doesn't match:".format(field=field)

220 ]

221 comparison = differ.compare(

222 baseline_field.splitlines(), result_field.splitlines()

223 )

224 message.extend(list(comparison))

225 error_messages.append("\n".join(message))

226

227 if not error_messages:

228 logging.debug(

229 "[Iteration {0}] {1}...\t[OK]".format(index + 1, command)

230 )

231 else:

232 success = False

233 if command.startswith("fwts"):

234 logging.error(

235 "[Iteration {0}] {1}...\t[FAIL]".format(

236 index + 1, command

237 )

238 )

239 else:

240 logging.error(

241 "[Iteration {0}] {1}...\t[FAIL]\n".format(

242 index + 1, command

243 )

244 )

245 for message in error_messages:

246 logging.error(message)

247

248 return success

249

250

251 def parse_args():

252 """

253 Parse command-line arguments

254 """

255 parser = ArgumentParser(

256 description=("Check power management " "test case results")

257 )

258 parser.add_argument(

259 "input_log_filename",

260 metavar="log_filename",

261 help=("Path to the input log file " "on which to perform the check"),

262 )

263 parser.add_argument(

264 "output_log_filename",

265 metavar="log_filename",

266 help=("Path to the output log file " "for the results of the check"),

267 )

268 log_levels = ["notset", "debug", "info", "warning", "error", "critical"]

269 parser.add_argument(

270 "--log-level",

271 dest="log_level",

272 default="info",

273 choices=log_levels,

274 help=(

275 "Log level. "

276 "One of {0} or {1} (%(default)s by default)".format(

277 ", ".join(log_levels[:-1]), log_levels[-1]

278 )

279 ),

280 )

281 args = parser.parse_args()

282 args.log_level = getattr(logging, args.log_level.upper())

283

284 return args

285

286

287 if __name__ == "__main__":

288 main()