1#!/usr/bin/env python3

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

3#

4# graphics_stress_test.py

5#

6# This file is part of Checkbox.

7#

8# Copyright 2012 Canonical Ltd.

9#

10# Authors: Alberto Milone <alberto.milone@canonical.com>

11#

12# Checkbox is free software: you can redistribute it and/or modify

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

14# as published by the Free Software Foundation.

15

16#

17# Checkbox is distributed in the hope that it will be useful,

18# but WITHOUT ANY WARRANTY; without even the implied warranty of

19# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

20# GNU General Public License for more details.

21#

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

23# along with Checkbox. If not, see <http://www.gnu.org/licenses/>.

24

25 import errno

26 import logging

27 import os

28 import re

29 import sys

30 import tempfile

31 import time

32

33 from argparse import ArgumentParser

34 from subprocess import call, Popen, PIPE

35 from checkbox_support.contrib import xrandr

36

37

38 class VtWrapper(object):

39 """docstring for VtWrapper"""

40

41 def __init__(self):

42 self.x_vt = self._get_x_vt()

43

44 def _get_x_vt(self):

45 """Get the vt where X lives"""

46 vt = 0

47 proc = Popen(["ps", "aux"], stdout=PIPE, universal_newlines=True)

48 proc_output = proc.communicate()[0].split("\n")

49 proc_line = re.compile(r".*tty(\d+).+/usr/bin/X.*")

50 for line in proc_output:

51 match = proc_line.match(line)

52 if match:

53 vt = match.group(1).strip().lower()

54 return int(vt)

55

56 def set_vt(self, vt):

57 retcode = call(["chvt", "%d" % vt])

58 return retcode

59

60

61 class SuspendWrapper(object):

62 def __init__(self):

63 pass

64

65 def can_we_sleep(self, mode):

66 """

67 Test to see if S3 state is available to us. /proc/acpi/* is old

68 and will be deprecated, using /sys/power to maintine usefulness for

69 future kernels.

70

71 """

72 states_fh = open("/sys/power/state", "r")

73 try:

74 states = states_fh.read().split()

75 finally:

76 states_fh.close()

77

78 if mode in states:

79 return True

80 else:

81 return False

82

83 def get_current_time(self):

84 cur_time = 0

85 time_fh = open("/sys/class/rtc/rtc0/since_epoch", "r")

86 try:

87 cur_time = int(time_fh.read())

88 finally:

89 time_fh.close()

90 return cur_time

91

92 def set_wake_time(self, time):

93 """

94 Get the current epoch time from /sys/class/rtc/rtc0/since_epoch

95 then add time and write our new wake_alarm time to

96 /sys/class/rtc/rtc0/wakealarm.

97

98 The math could probably be done better but this method avoids having to

99 worry about whether or not we're using UTC or local time for both the

100 hardware and system clocks.

101

102 """

103 cur_time = self.get_current_time()

104 logging.debug("Current epoch time: %s" % cur_time)

105

106 wakealarm_fh = open("/sys/class/rtc/rtc0/wakealarm", "w")

107

108 try:

109 wakealarm_fh.write("0\n")

110 wakealarm_fh.flush()

111

112 wakealarm_fh.write("+%s\n" % time)

113 wakealarm_fh.flush()

114 finally:

115 wakealarm_fh.close()

116

117 logging.debug("Wake alarm in %s seconds" % time)

118

119 def do_suspend(self, mode):

120 """

121 Suspend the system and hope it wakes up.

122 Previously tried writing new state to /sys/power/state but that

123 seems to put the system into an uncrecoverable S3 state. So far,

124 pm-suspend seems to be the most reliable way to go.

125

126 """

127 if mode == "mem":

128 status = call("/usr/sbin/pm-suspend")

129 elif mode == "disk":

130 status = call("/usr/sbin/pm-hibernate")

131 else:

132 logging.debug("Unknown sleep state passed")

133 status == 1

134

135 return status

136

137

138 class RotationWrapper(object):

139

140 def __init__(self):

141 self._rotations = {

142 "normal": xrandr.RR_ROTATE_0,

143 "right": xrandr.RR_ROTATE_90,

144 "inverted": xrandr.RR_ROTATE_180,

145 "left": xrandr.RR_ROTATE_270,

146 }

147

148 def _rotate_screen(self, rotation):

149 # Refresh the screen. Required by NVIDIA

150 screen = xrandr.get_current_screen()

151 screen.set_rotation(rotation)

152 return screen.apply_config()

153

154 def do_rotation_cycle(self):

155 """Cycle through all possible rotations"""

156 rots_statuses = {}

157

158 for rot in self._rotations:

159 try:

160 status = self._rotate_screen(self._rotations[rot])

161 except (xrandr.RRError, xrandr.UnsupportedRRError) as err:

162 status = 1

163 error = err

164 else:

165 error = "N/A"

166 # Collect the status and the error message

167 rots_statuses[rot] = (status, error)

168 time.sleep(4)

169

170 # Try to set the screen back to normal

171 try:

172 self._rotate_screen(xrandr.RR_ROTATE_0)

173 except (xrandr.RRError, xrandr.UnsupportedRRError) as error:

174 print(error)

175

176 result = 0

177 for elem in rots_statuses:

178 status = rots_statuses.get(elem)[0]

179 error = rots_statuses.get(elem)[1]

180 if status != 0:

181 logging.error(

182 'Error: rotation "%s" failed with status %d: %s.'

183 % (elem, status, error)

184 )

185 result = 1

186 return result

187

188

189 class RenderCheckWrapper(object):

190 """A simple class to run the rendercheck suites"""

191

192 def __init__(self, temp_dir=None):

193 self._temp_dir = temp_dir

194

195 def _print_test_info(self, suites="all", iteration=1, show_errors=False):

196 """Print the output of the test suite"""

197

198 main_command = "rendercheck"

199 passed = 0

200 total = 0

201

202 if self._temp_dir:

203 # Use the specified path

204 temp_file = tempfile.NamedTemporaryFile(

205 dir=self._temp_dir, delete=False

206 )

207 else:

208 # Use /tmp

209 temp_file = tempfile.NamedTemporaryFile(delete=False)

210

211 if suites == all:

212 full_command = [main_command, "-f", "a8r8g8b8"]

213 else:

214 full_command = [main_command, "-t", suites, "-f", "a8r8g8b8"]

215

216 try:

217 # Let's dump the output into file as it can be very large

218 # and we don't want to store it in memory

219 process = Popen(

220 full_command, stdout=temp_file, universal_newlines=True

221 )

222 except OSError as exc:

223 if exc.errno == errno.ENOENT:

224 logging.error(

225 "Error: please make sure that rendercheck " "is installed."

226 )

227 exit(1)

228 else:

229 raise

230

231 exit_code = process.wait()

232

233 temp_file.close()

234

235 # Read values from the file

236 errors = re.compile(r".*test error.*")

237 results = re.compile(r"(.+) tests passed of (.+) total.*")

238

239 first_error = True

240 with open(temp_file.name) as temp_handle:

241 for line in temp_handle:

242 match_output = results.match(line)

243 match_errors = errors.match(line)

244 if match_output:

245 passed = int(match_output.group(1).strip())

246 total = int(match_output.group(2).strip())

247 logging.info("Results:")

248 logging.info(

249 " %d tests passed out of %d." % (passed, total)

250 )

251 if show_errors and match_errors:

252 error = match_errors.group(0).strip()

253 if first_error:

254 logging.debug(

255 "Rendercheck %s suite errors "

256 "from iteration %d:" % (suites, iteration)

257 )

258 first_error = False

259 logging.debug(" %s" % error)

260

261 # Remove the file

262 os.unlink(temp_file.name)

263

264 return (exit_code, passed, total)

265

266 def run_test(self, suites=[], iterations=1, show_errors=False):

267 exit_status = 0

268 for suite in suites:

269 for it in range(iterations):

270 logging.info(

271 "Iteration %d of Rendercheck %s suite..." % (it + 1, suite)

272 )

273 (status, passed, total) = self._print_test_info(

274 suites=suite, iteration=it + 1, show_errors=show_errors

275 )

276 if status != 0:

277 # Make sure to catch a non-zero exit status

278 logging.info(

279 "Iteration %d of Rendercheck %s suite "

280 "exited with status %d." % (it + 1, suite, status)

281 )

282 exit_status = status

283 it += 1

284

285 # exit with 1 if passed < total

286 if passed < total:

287 if exit_status == 0:

288 exit_status = 1

289 return exit_status

290

291 def get_suites_list(self):

292 """Return a list of the available test suites"""

293 try:

294 process = Popen(

295 ["rendercheck", "--help"],

296 stdout=PIPE,

297 stderr=PIPE,

298 universal_newlines=True,

299 )

300 except OSError as exc:

301 if exc.errno == errno.ENOENT:

302 logging.error(

303 "Error: please make sure that rendercheck " "is installed."

304 )

305 exit(1)

306 else:

307 raise

308

309 proc = process.communicate()[1].split("\n")

310 found = False

311 tests_pattern = re.compile(r".*Available tests: *(.+).*")

312 temp_line = ""

313 tests = []

314 for line in proc:

315 if found:

316 temp_line += line

317 match = tests_pattern.match(line)

318 if match:

319 first_line = match.group(1).strip().lower()

320 found = True

321 temp_line += first_line

322 for elem in temp_line.split(","):

323 test = elem.strip()

324 if elem:

325 tests.append(test)

326 return tests

327

328

329 def main():

330 # Make sure that we have root privileges

331 if os.geteuid() != 0:

332 print("Error: please run this program as root", file=sys.stderr)

333 exit(1)

334

335 usage = "Usage: %prog [OPTIONS]"

336 parser = ArgumentParser(usage)

337 parser.add_argument(

338 "-i",

339 "--iterations",

340 type=int,

341 default=10,

342 help="The number of times to run the test. \

343 Default is 10",

344 )

345 parser.add_argument(

346 "-d",

347 "--debug",

348 action="store_true",

349 help="Choose this to add verbose output \

350 for debug purposes",

351 )

352 parser.add_argument(

353 "-b",

354 "--blacklist",

355 nargs="+",

356 help="Name(s) of rendercheck test(s) to blacklist.",

357 )

358 parser.add_argument(

359 "-o",

360 "--output",

361 default="",

362 help="The path to the log which will be dumped. \

363 Default is stdout",

364 )

365 parser.add_argument(

366 "-tp",

367 "--temp",

368 default="",

369 help="The path where to store temporary files. \

370 Default is /tmp",

371 )

372 args = parser.parse_args()

373

374 # Set up logging to console

375 format = "%(message)s"

376

377 console_handler = logging.StreamHandler()

378 console_handler.setFormatter(logging.Formatter(format))

379

380 # Set up the overall logger

381 logger = logging.getLogger()

382 # This is necessary to ensure debug messages are passed through the logger

383 # to the handler

384 logger.setLevel(logging.DEBUG)

385

386 # This is what happens when -d and/or -o are passed:

387 # -o -> stdout (info) - log (info)

388 # -d -> only stdout (info and debug) - no log

389 # -d -o -> stdout (info) - log (info and debug)

390

391 # Write to a log

392 if args.output:

393 # Write INFO to stdout

394 console_handler.setLevel(logging.INFO)

395 logger.addHandler(console_handler)

396 # Specify a log file

397 logfile = args.output

398 logfile_handler = logging.FileHandler(logfile)

399 if args.debug:

400 # Write INFO and DEBUG to a log

401 logfile_handler.setLevel(logging.DEBUG)

402 else:

403 # Write INFO to a log

404 logfile_handler.setLevel(logging.INFO)

405

406 logfile_handler.setFormatter(logging.Formatter(format))

407 logger.addHandler(logfile_handler)

408

409 # Write only to stdout

410 else:

411 if args.debug:

412 # Write INFO and DEBUG to stdout

413 console_handler.setLevel(logging.DEBUG)

414 logger.addHandler(console_handler)

415 else:

416 # Write INFO to stdout

417 console_handler.setLevel(logging.INFO)

418 logger.addHandler(console_handler)

419

420 status = 0

421

422 rendercheck = RenderCheckWrapper(args.temp)

423 tests = rendercheck.get_suites_list()

424 for test in args.blacklist:

425 if test in tests:

426 tests.remove(test)

427

428 # Switch between the tty where X lives and tty10

429 vt_wrap = VtWrapper()

430 target_vt = 10

431 if vt_wrap.x_vt != target_vt:

432 logging.info("== Vt switch test ==")

433 for it in range(args.iterations):

434 logging.info("Iteration %d...", it)

435 retcode = vt_wrap.set_vt(target_vt)

436 if retcode != 0:

437 logging.error(

438 "Error: switching to tty%d failed with code %d "

439 "on iteration %d" % (target_vt, retcode, it)

440 )

441 status = 1

442 else:

443 logging.info("Switching to tty%d: passed" % (target_vt))

444 time.sleep(2)

445 retcode = vt_wrap.set_vt(vt_wrap.x_vt)

446 if retcode != 0:

447 logging.error(

448 "Error: switching to tty%d failed with code %d "

449 "on iteration %d" % (vt_wrap.x_vt, retcode, it)

450 )

451 else:

452 logging.info("Switching to tty%d: passed" % (vt_wrap.x_vt))

453 status = 1

454 else:

455 logging.error("Error: please run X on a tty other than 10")

456

457 # Call sleep x times

458 logging.info("== Sleep test ==")

459 sleep_test = SuspendWrapper()

460 sleep_mode = "mem"

461 # See if we can sleep

462 if sleep_test.can_we_sleep(sleep_mode):

463 for it in range(args.iterations):

464 # Run the test

465 logging.info("Iteration %d...", it + 1)

466 # Set wake time

467 sleep_test.set_wake_time(20)

468 # Suspend to RAM

469 if sleep_test.do_suspend(sleep_mode) == 0:

470 logging.info("Passed")

471 else:

472 logging.error("Failed")

473 status = 1

474 else:

475 # Skip the test

476 logging.info("Skipped (the system does not seem to support S3")

477

478 # Rotate the screen x times

479 # The app already rotates the screen 5 times

480 logging.info("== Rotation test ==")

481 rotation_test = RotationWrapper()

482

483 for it in range(args.iterations):

484 logging.info("Iteration %d...", it + 1)

485 if rotation_test.do_rotation_cycle() == 0:

486 logging.info("Passed")

487 else:

488 logging.error("Failed")

489 status = 1

490

491 # Call rendercheck x times

492 logging.info("== Rendercheck test ==")

493 if rendercheck.run_test(tests, args.iterations, args.debug) == 0:

494 logging.info("Passed")

495 else:

496 logging.error("Failed")

497 status = 1

498

499 return status

500

501

502 if __name__ == "__main__":

503 exit(main())