run.py 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405
  1. #!/usr/bin/python2 -O
  2. # vim: fileencoding=utf-8
  3. #
  4. # The Qubes OS Project, https://www.qubes-os.org/
  5. #
  6. # Copyright (C) 2014-2015 Joanna Rutkowska <joanna@invisiblethingslab.com>
  7. # Copyright (C) 2014-2015 Wojtek Porczyk <woju@invisiblethingslab.com>
  8. #
  9. # This program is free software; you can redistribute it and/or modify
  10. # it under the terms of the GNU General Public License as published by
  11. # the Free Software Foundation; either version 2 of the License, or
  12. # (at your option) any later version.
  13. #
  14. # This program is distributed in the hope that it will be useful,
  15. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  16. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  17. # GNU General Public License for more details.
  18. #
  19. # You should have received a copy of the GNU General Public License along
  20. # with this program; if not, write to the Free Software Foundation, Inc.,
  21. # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  22. #
  23. import argparse
  24. import curses
  25. import logging
  26. import logging.handlers
  27. import os
  28. import socket
  29. import subprocess
  30. import sys
  31. import unittest
  32. import unittest.signals
  33. import qubes.tests
  34. class CursesColor(dict):
  35. def __init__(self):
  36. super(CursesColor, self).__init__()
  37. try:
  38. curses.setupterm()
  39. except curses.error:
  40. return
  41. # pylint: disable=bad-whitespace
  42. self['black'] = curses.tparm(curses.tigetstr('setaf'), 0)
  43. self['red'] = curses.tparm(curses.tigetstr('setaf'), 1)
  44. self['green'] = curses.tparm(curses.tigetstr('setaf'), 2)
  45. self['yellow'] = curses.tparm(curses.tigetstr('setaf'), 3)
  46. self['blue'] = curses.tparm(curses.tigetstr('setaf'), 4)
  47. self['magenta'] = curses.tparm(curses.tigetstr('setaf'), 5)
  48. self['cyan'] = curses.tparm(curses.tigetstr('setaf'), 6)
  49. self['white'] = curses.tparm(curses.tigetstr('setaf'), 7)
  50. self['bold'] = curses.tigetstr('bold')
  51. self['normal'] = curses.tigetstr('sgr0')
  52. def __missing__(self, key):
  53. # pylint: disable=unused-argument,no-self-use
  54. return ''
  55. class QubesTestResult(unittest.TestResult):
  56. '''A test result class that can print colourful text results to a stream.
  57. Used by TextTestRunner. This is a lightly rewritten unittest.TextTestResult.
  58. '''
  59. separator1 = unittest.TextTestResult.separator1
  60. separator2 = unittest.TextTestResult.separator2
  61. def __init__(self, stream, descriptions, verbosity):
  62. super(QubesTestResult, self).__init__(stream, descriptions, verbosity)
  63. self.stream = stream
  64. self.showAll = verbosity > 1 # pylint: disable=invalid-name
  65. self.dots = verbosity == 1
  66. self.descriptions = descriptions
  67. self.color = CursesColor()
  68. self.hostname = socket.gethostname()
  69. self.log = logging.getLogger('qubes.tests')
  70. def _fmtexc(self, err):
  71. if str(err[1]):
  72. return '{color[bold]}{}:{color[normal]} {!s}'.format(
  73. err[0].__name__, err[1], color=self.color)
  74. else:
  75. return '{color[bold]}{}{color[normal]}'.format(
  76. err[0].__name__, color=self.color)
  77. def get_log(self, test):
  78. try:
  79. return test.log
  80. except AttributeError:
  81. return self.log
  82. def getDescription(self, test): # pylint: disable=invalid-name
  83. teststr = str(test).split('/')
  84. for i in range(-2, 0):
  85. try:
  86. fullname = teststr[i].split('_', 2)
  87. except IndexError:
  88. continue
  89. fullname[-1] = '{color[bold]}{}{color[normal]}'.format(
  90. fullname[-1], color=self.color)
  91. teststr[i] = '_'.join(fullname)
  92. teststr = '/'.join(teststr)
  93. doc_first_line = test.shortDescription()
  94. if self.descriptions and doc_first_line:
  95. return '\n'.join((teststr, ' {}'.format(
  96. doc_first_line, color=self.color)))
  97. else:
  98. return teststr
  99. def startTest(self, test): # pylint: disable=invalid-name
  100. super(QubesTestResult, self).startTest(test)
  101. self.get_log(test).critical('started')
  102. if self.showAll:
  103. if not qubes.tests.in_git:
  104. self.stream.write('{}: '.format(self.hostname))
  105. self.stream.write(self.getDescription(test))
  106. self.stream.write(' ... ')
  107. self.stream.flush()
  108. def addSuccess(self, test): # pylint: disable=invalid-name
  109. super(QubesTestResult, self).addSuccess(test)
  110. self.get_log(test).warning('ok')
  111. if self.showAll:
  112. self.stream.writeln('{color[green]}ok{color[normal]}'.format(
  113. color=self.color))
  114. elif self.dots:
  115. self.stream.write('.')
  116. self.stream.flush()
  117. def addError(self, test, err): # pylint: disable=invalid-name
  118. super(QubesTestResult, self).addError(test, err)
  119. self.get_log(test).critical(
  120. 'ERROR ({err[0].__name__}: {err[1]!r})'.format(err=err))
  121. if self.showAll:
  122. self.stream.writeln(
  123. '{color[red]}{color[bold]}ERROR{color[normal]} ({})'.format(
  124. self._fmtexc(err), color=self.color))
  125. elif self.dots:
  126. self.stream.write(
  127. '{color[red]}{color[bold]}E{color[normal]}'.format(
  128. color=self.color))
  129. self.stream.flush()
  130. def addFailure(self, test, err): # pylint: disable=invalid-name
  131. super(QubesTestResult, self).addFailure(test, err)
  132. self.get_log(test).error(
  133. 'FAIL ({err[0].__name__}: {err[1]!r})'.format(err=err))
  134. if self.showAll:
  135. self.stream.writeln('{color[red]}FAIL{color[normal]}'.format(
  136. color=self.color))
  137. elif self.dots:
  138. self.stream.write('{color[red]}F{color[normal]}'.format(
  139. color=self.color))
  140. self.stream.flush()
  141. def addSkip(self, test, reason): # pylint: disable=invalid-name
  142. super(QubesTestResult, self).addSkip(test, reason)
  143. self.get_log(test).warning('skipped ({})'.format(reason))
  144. if self.showAll:
  145. self.stream.writeln(
  146. '{color[cyan]}skipped{color[normal]} ({})'.format(
  147. reason, color=self.color))
  148. elif self.dots:
  149. self.stream.write('{color[cyan]}s{color[normal]}'.format(
  150. color=self.color))
  151. self.stream.flush()
  152. def addExpectedFailure(self, test, err): # pylint: disable=invalid-name
  153. super(QubesTestResult, self).addExpectedFailure(test, err)
  154. self.get_log(test).warning('expected failure')
  155. if self.showAll:
  156. self.stream.writeln(
  157. '{color[yellow]}expected failure{color[normal]}'.format(
  158. color=self.color))
  159. elif self.dots:
  160. self.stream.write('{color[yellow]}x{color[normal]}'.format(
  161. color=self.color))
  162. self.stream.flush()
  163. def addUnexpectedSuccess(self, test): # pylint: disable=invalid-name
  164. super(QubesTestResult, self).addUnexpectedSuccess(test)
  165. self.get_log(test).error('unexpected success')
  166. if self.showAll:
  167. self.stream.writeln(
  168. '{color[yellow]}{color[bold]}unexpected success'
  169. '{color[normal]}'.format(color=self.color))
  170. elif self.dots:
  171. self.stream.write(
  172. '{color[yellow]}{color[bold]}u{color[normal]}'.format(
  173. color=self.color))
  174. self.stream.flush()
  175. def printErrors(self): # pylint: disable=invalid-name
  176. if self.dots or self.showAll:
  177. self.stream.writeln()
  178. self.printErrorList(
  179. '{color[red]}{color[bold]}ERROR{color[normal]}'.format(
  180. color=self.color),
  181. self.errors)
  182. self.printErrorList(
  183. '{color[red]}FAIL{color[normal]}'.format(color=self.color),
  184. self.failures)
  185. def printErrorList(self, flavour, errors): # pylint: disable=invalid-name
  186. for test, err in errors:
  187. self.stream.writeln(self.separator1)
  188. self.stream.writeln('%s: %s' % (flavour, self.getDescription(test)))
  189. self.stream.writeln(self.separator2)
  190. self.stream.writeln('%s' % err)
  191. class QubesDNCTestResult(QubesTestResult):
  192. do_not_clean = True
  193. def demo(verbosity=2):
  194. class TC_00_Demo(qubes.tests.QubesTestCase):
  195. '''Demo class'''
  196. # pylint: disable=no-self-use
  197. def test_0_success(self):
  198. '''Demo test (success)'''
  199. pass
  200. def test_1_error(self):
  201. '''Demo test (error)'''
  202. raise Exception()
  203. def test_2_failure(self):
  204. '''Demo test (failure)'''
  205. self.fail('boo')
  206. def test_3_skip(self):
  207. '''Demo test (skipped by call to self.skipTest())'''
  208. self.skipTest('skip')
  209. @unittest.skip(None)
  210. def test_4_skip_decorator(self):
  211. '''Demo test (skipped by decorator)'''
  212. pass
  213. @unittest.expectedFailure
  214. def test_5_expected_failure(self):
  215. '''Demo test (expected failure)'''
  216. self.fail()
  217. @unittest.expectedFailure
  218. def test_6_unexpected_success(self):
  219. '''Demo test (unexpected success)'''
  220. pass
  221. suite = unittest.TestLoader().loadTestsFromTestCase(TC_00_Demo)
  222. runner = unittest.TextTestRunner(stream=sys.stdout, verbosity=verbosity)
  223. runner.resultclass = QubesTestResult
  224. return runner.run(suite).wasSuccessful()
  225. parser = argparse.ArgumentParser(
  226. epilog='''When running only specific tests, write their names like in log,
  227. in format: MODULE+"/"+CLASS+"/"+FUNCTION. MODULE should omit initial
  228. "qubes.tests.". Example: basic/TC_00_Basic/test_000_create''')
  229. parser.add_argument('--verbose', '-v',
  230. action='count',
  231. help='increase console verbosity level')
  232. parser.add_argument('--quiet', '-q',
  233. action='count',
  234. help='decrease console verbosity level')
  235. parser.add_argument('--list', '-l',
  236. action='store_true', dest='list',
  237. help='list all available tests and exit')
  238. parser.add_argument('--failfast', '-f',
  239. action='store_true', dest='failfast',
  240. help='stop on the first fail, error or unexpected success')
  241. parser.add_argument('--no-failfast',
  242. action='store_false', dest='failfast',
  243. help='disable --failfast')
  244. parser.add_argument('--do-not-clean', '--dnc', '-D',
  245. action='store_true', dest='do_not_clean',
  246. help='do not execute tearDown on failed tests. Implies --failfast.')
  247. parser.add_argument('--do-clean', '-C',
  248. action='store_false', dest='do_not_clean',
  249. help='do execute tearDown even on failed tests.')
  250. # pylint: disable=protected-access
  251. parser.add_argument('--loglevel', '-L', metavar='LEVEL',
  252. action='store', choices=tuple(k
  253. for k in sorted(logging._levelNames.keys(),
  254. key=lambda x: logging._levelNames[x])
  255. if isinstance(k, basestring)),
  256. help='logging level for file and syslog forwarding '
  257. '(one of: %(choices)s; default: %(default)s)')
  258. # pylint: enable=protected-access
  259. parser.add_argument('--logfile', '-o', metavar='FILE',
  260. action='store',
  261. help='if set, test run will be also logged to file')
  262. parser.add_argument('--syslog',
  263. action='store_true', dest='syslog',
  264. help='reenable logging to syslog')
  265. parser.add_argument('--no-syslog',
  266. action='store_false', dest='syslog',
  267. help='disable logging to syslog')
  268. parser.add_argument('--kmsg', '--very-brave-or-very-stupid',
  269. action='store_true', dest='kmsg',
  270. help='log most important things to kernel ring-buffer')
  271. parser.add_argument('--no-kmsg', '--i-am-smarter-than-kay-sievers',
  272. action='store_false', dest='kmsg',
  273. help='do not abuse kernel ring-buffer')
  274. parser.add_argument('names', metavar='TESTNAME',
  275. action='store', nargs='*',
  276. help='list of tests to run named like in description '
  277. '(default: run all tests)')
  278. parser.set_defaults(
  279. failfast=False,
  280. loglevel='DEBUG',
  281. logfile=None,
  282. syslog=True,
  283. kmsg=False,
  284. verbose=2,
  285. quiet=0)
  286. def list_test_cases(suite):
  287. for test in suite:
  288. if isinstance(test, unittest.TestSuite):
  289. #yield from
  290. for i in list_test_cases(test):
  291. yield i
  292. else:
  293. yield test
  294. def main():
  295. args = parser.parse_args()
  296. suite = unittest.TestSuite()
  297. loader = unittest.TestLoader()
  298. if args.names:
  299. alltests = loader.loadTestsFromName('qubes.tests')
  300. for name in args.names:
  301. suite.addTests(
  302. [test for test in list_test_cases(alltests)
  303. if (str(test)).startswith(name)])
  304. else:
  305. suite.addTests(loader.loadTestsFromName('qubes.tests'))
  306. if args.list:
  307. for test in list_test_cases(suite):
  308. print(str(test)) # pylint: disable=superfluous-parens
  309. return True
  310. if args.do_not_clean:
  311. args.failfast = True
  312. logging.root.setLevel(args.loglevel)
  313. if args.logfile is not None:
  314. ha_file = logging.FileHandler(
  315. os.path.join(os.environ['HOME'], args.logfile))
  316. ha_file.setFormatter(
  317. logging.Formatter('%(asctime)s %(name)s[%(process)d]: %(message)s'))
  318. logging.root.addHandler(ha_file)
  319. if args.syslog:
  320. ha_syslog = logging.handlers.SysLogHandler('/dev/log')
  321. ha_syslog.setFormatter(
  322. logging.Formatter('%(name)s[%(process)d]: %(message)s'))
  323. logging.root.addHandler(ha_syslog)
  324. if args.kmsg:
  325. try:
  326. subprocess.check_call(('sudo', 'chmod', '666', '/dev/kmsg'))
  327. except subprocess.CalledProcessError:
  328. parser.error('could not chmod /dev/kmsg')
  329. else:
  330. ha_kmsg = logging.FileHandler('/dev/kmsg', 'w')
  331. ha_kmsg.setFormatter(
  332. logging.Formatter('%(name)s[%(process)d]: %(message)s'))
  333. ha_kmsg.setLevel(logging.CRITICAL)
  334. logging.root.addHandler(ha_kmsg)
  335. runner = unittest.TextTestRunner(stream=sys.stdout,
  336. verbosity=(args.verbose-args.quiet),
  337. failfast=args.failfast)
  338. unittest.signals.installHandler()
  339. runner.resultclass = QubesDNCTestResult \
  340. if args.do_not_clean else QubesTestResult
  341. return runner.run(suite).wasSuccessful()
  342. if __name__ == '__main__':
  343. sys.exit(not main())