Qubes/core-admin
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

tests: add logging

Browse Source 
Test suite now logs to three places:
 * syslog
 * ${HOME}/qubes-tests.log
 * /dev/kmsg (only start and errors; this is handy while debugging Oopses by
   correlating them with tests)
This commit is contained in:
Wojtek Porczyk 2015-02-11 14:26:29 +01:00
parent b6530200c9
commit d81ae01d48
9 changed files with 174 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

147
tests/__init__.py
View File

@ -24,17 +24,65 @@
#
import multiprocessing
import logging
import os
import shutil
import subprocess
import unittest
import libvirt
import lxml.etree
import qubes.backup
import qubes.qubes
VMPREFIX = 'test-'
#: :py:obj:`True` if running in dom0, :py:obj:`False` otherwise
in_dom0 = False
#: :py:obj:`False` if outside of git repo,
#: path to root of the directory otherwise
in_git = False
try:
import libvirt
libvirt.openReadOnly(qubes.qubes.defaults['libvirt_uri']).close()
in_dom0 = True
except libvirt.libvirtError:
pass
try:
in_git = subprocess.check_output(
['git', 'rev-parse', '--show-toplevel']).strip()
except subprocess.CalledProcessError:
# git returned nonzero, we are outside git repo
pass
except OSError:
# command not found; let's assume we're outside
pass
def skipUnlessDom0(test_item):
'''Decorator that skips test outside dom0.
Some tests (especially integration tests) have to be run in more or less
working dom0. This is checked by connecting to libvirt.
''' # pylint: disable=invalid-name
return unittest.skipUnless(in_dom0, 'outside dom0')(test_item)
def skipUnlessGit(test_item):
'''Decorator that skips test outside git repo.
There are very few tests that an be run only in git. One example is
correctness of example code that won't get included in RPM.
''' # pylint: disable=invalid-name
return unittest.skipUnless(in_git, 'outside git tree')(test_item)
class _AssertNotRaisesContext(object):
"""A context manager used to implement TestCase.assertNotRaises methods.
@ -67,6 +115,70 @@ class _AssertNotRaisesContext(object):
self.exception = exc_value # store for later retrieval
class QubesTestCase(unittest.TestCase):
'''Base class for Qubes unit tests.
'''
def __init__(self, *args, **kwargs):
super(QubesTestCase, self).__init__(*args, **kwargs)
self.log = logging.getLogger('{}.{}.{}'.format(
self.__class__.__module__,
self.__class__.__name__,
self._testMethodName))
def __str__(self):
return '{}/{}/{}'.format(
'.'.join(self.__class__.__module__.split('.')[2:]),
self.__class__.__name__,
self._testMethodName)
def assertNotRaises(self, excClass, callableObj=None, *args, **kwargs):
"""Fail if an exception of class excClass is raised
by callableObj when invoked with arguments args and keyword
arguments kwargs. If a different type of exception is
raised, it will not be caught, and the test case will be
deemed to have suffered an error, exactly as for an
unexpected exception.
If called with callableObj omitted or None, will return a
context object used like this::
with self.assertRaises(SomeException):
do_something()
The context manager keeps a reference to the exception as
the 'exception' attribute. This allows you to inspect the
exception after the assertion::
with self.assertRaises(SomeException) as cm:
do_something()
the_exception = cm.exception
self.assertEqual(the_exception.error_code, 3)
"""
context = _AssertNotRaisesContext(excClass, self)
if callableObj is None:
return context
with context:
callableObj(*args, **kwargs)
def assertXMLEqual(self, xml1, xml2):
'''Check for equality of two XML objects.
:param xml1: first element
:param xml2: second element
:type xml1: :py:class:`lxml.etree._Element`
:type xml2: :py:class:`lxml.etree._Element`
''' # pylint: disable=invalid-name
self.assertEqual(xml1.tag, xml2.tag)
self.assertEqual(xml1.text, xml2.text)
self.assertItemsEqual(xml1.keys(), xml2.keys())
for key in xml1.keys():
self.assertEqual(xml1.get(key), xml2.get(key))
class SystemTestsMixin(object):
def setUp(self):
@ -101,36 +213,6 @@ class SystemTestsMixin(object):
self.conn.close()
def assertNotRaises(self, excClass, callableObj=None, *args, **kwargs):
"""Fail if an exception of class excClass is raised
by callableObj when invoked with arguments args and keyword
arguments kwargs. If a different type of exception is
raised, it will not be caught, and the test case will be
deemed to have suffered an error, exactly as for an
unexpected exception.
If called with callableObj omitted or None, will return a
context object used like this::
with self.assertRaises(SomeException):
do_something()
The context manager keeps a reference to the exception as
the 'exception' attribute. This allows you to inspect the
exception after the assertion::
with self.assertRaises(SomeException) as cm:
do_something()
the_exception = cm.exception
self.assertEqual(the_exception.error_code, 3)
"""
context = _AssertNotRaisesContext(excClass, self)
if callableObj is None:
return context
with context:
callableObj(*args, **kwargs)
def make_vm_name(self, name):
return VMPREFIX + name
@ -382,6 +464,9 @@ class BackupTestsMixin(SystemTestsMixin):
def load_tests(loader, tests, pattern):
# discard any tests from this module, because it hosts base classes
tests = unittest.TestSuite()
for modname in (
'qubes.tests.basic',
'qubes.tests.network',

2
tests/backup.py
View File

@ -31,7 +31,7 @@ import sys
import qubes.tests
class TC_00_Backup(qubes.tests.BackupTestsMixin, unittest.TestCase):
class TC_00_Backup(qubes.tests.BackupTestsMixin, qubes.tests.QubesTestCase):
def test_000_basic_backup(self):
vms = self.create_backup_vms()
self.make_backup(vms)

2
tests/backupcompatibility.py
View File

@ -107,7 +107,7 @@ QUBESXML_R1 = '''<?xml version='1.0' encoding='UTF-8'?>
'''
class TC_00_BackupCompatibility(qubes.tests.BackupTestsMixin, unittest.TestCase):
class TC_00_BackupCompatibility(qubes.tests.BackupTestsMixin, qubes.tests.QubesTestCase):
def create_whitelisted_appmenus(self, filename):
f = open(filename, "w")
f.write("gnome-terminal.desktop\n")

4
tests/basic.py
View File

@ -34,7 +34,7 @@ from qubes.qubes import QubesVmCollection, QubesException, system_path
import qubes.qubes
import qubes.tests
class TC_00_Basic(qubes.tests.SystemTestsMixin, unittest.TestCase):
class TC_00_Basic(qubes.tests.SystemTestsMixin, qubes.tests.QubesTestCase):
def test_000_create(self):
vmname = self.make_vm_name('appvm')
vm = self.qc.add_new_vm('QubesAppVm',
@ -49,7 +49,7 @@ class TC_00_Basic(qubes.tests.SystemTestsMixin, unittest.TestCase):
vm.verify_files()
class TC_01_Properties(qubes.tests.SystemTestsMixin, unittest.TestCase):
class TC_01_Properties(qubes.tests.SystemTestsMixin, qubes.tests.QubesTestCase):
def setUp(self):
super(TC_01_Properties, self).setUp()
self.vmname = self.make_vm_name('appvm')

4
tests/dom0_update.py
View File

@ -27,12 +27,14 @@ import tempfile
import unittest
from qubes.qubes import QubesVmCollection
import qubes.qubes
VM_PREFIX = "test-"
@unittest.skipUnless(os.path.exists('/usr/bin/rpmsign') and
os.path.exists('/usr/bin/rpmbuild'),
'rpm-sign and/or rpm-buid not installed')
class TC_00_Dom0Upgrade(unittest.TestCase):
class TC_00_Dom0Upgrade(qubes.tests.QubesTestCase):
cleanup_paths = []
pkg_name = 'qubes-test-pkg'
dom0_update_common_opts = ['--disablerepo=*', '--enablerepo=test',

2
tests/network.py
View File

@ -34,7 +34,7 @@ from qubes.qubes import QubesVmCollection, defaults
import qubes.tests
class VmNetworkingTests(qubes.tests.SystemTestsMixin, unittest.TestCase):
class VmNetworkingTests(qubes.tests.SystemTestsMixin, qubes.tests.QubesTestCase):
test_ip = '192.168.123.45'
test_name = 'test.example.com'

2
tests/regressions.py
View File

@ -29,7 +29,7 @@ import unittest
import qubes.qubes
import qubes.tests
class TC_00_Regressions(qubes.tests.SystemTestsMixin, unittest.TestCase):
class TC_00_Regressions(qubes.tests.SystemTestsMixin, qubes.tests.QubesTestCase):
# Bug: #906
def test_000_bug_906_db_locking(self):
def create_vm(vmname):

56
tests/run.py
View File

@ -24,7 +24,11 @@
import curses
import importlib
import logging
import logging.handlers
import os
import socket
import subprocess
import sys
import unittest
import unittest.signals
@ -57,7 +61,7 @@ class CursesColor(dict):
return ''
class CursesTestResult(unittest.TestResult):
class QubesTestResult(unittest.TestResult):
'''A test result class that can print colourful text results to a stream.
Used by TextTestRunner. This is a lightly rewritten unittest.TextTestResult.
@ -67,7 +71,7 @@ class CursesTestResult(unittest.TestResult):
separator2 = unittest.TextTestResult.separator2
def __init__(self, stream, descriptions, verbosity):
super(CursesTestResult, self).__init__(stream, descriptions, verbosity)
super(QubesTestResult, self).__init__(stream, descriptions, verbosity)
self.stream = stream
self.showAll = verbosity > 1 # pylint: disable=invalid-name
self.dots = verbosity == 1
@ -76,6 +80,9 @@ class CursesTestResult(unittest.TestResult):
self.color = CursesColor()
self.hostname = socket.gethostname()
self.log = logging.getLogger('qubes.tests')
def _fmtexc(self, err):
if str(err[1]):
return '{color[bold]}{}:{color[normal]} {!s}'.format(
@ -104,7 +111,8 @@ class CursesTestResult(unittest.TestResult):
return teststr
def startTest(self, test): # pylint: disable=invalid-name
super(CursesTestResult, self).startTest(test)
super(QubesTestResult, self).startTest(test)
test.log.critical('started')
if self.showAll:
# if not qubes.tests.in_git:
self.stream.write('{}: '.format(self.hostname))
@ -113,7 +121,8 @@ class CursesTestResult(unittest.TestResult):
self.stream.flush()
def addSuccess(self, test): # pylint: disable=invalid-name
super(CursesTestResult, self).addSuccess(test)
super(QubesTestResult, self).addSuccess(test)
test.log.warning('ok')
if self.showAll:
self.stream.writeln('{color[green]}ok{color[normal]}'.format(
color=self.color))
@ -122,7 +131,8 @@ class CursesTestResult(unittest.TestResult):
self.stream.flush()
def addError(self, test, err): # pylint: disable=invalid-name
super(CursesTestResult, self).addError(test, err)
super(QubesTestResult, self).addError(test, err)
test.log.critical('ERROR ({err[0].__name__}: {err[1]!r})'.format(err=err))
if self.showAll:
self.stream.writeln(
'{color[red]}{color[bold]}ERROR{color[normal]} ({})'.format(
@ -134,7 +144,8 @@ class CursesTestResult(unittest.TestResult):
self.stream.flush()
def addFailure(self, test, err): # pylint: disable=invalid-name
super(CursesTestResult, self).addFailure(test, err)
super(QubesTestResult, self).addFailure(test, err)
test.log.error('FAIL ({err[0]!s}: {err[1]!r})'.format(err=err))
if self.showAll:
self.stream.writeln('{color[red]}FAIL{color[normal]}'.format(
color=self.color))
@ -144,7 +155,8 @@ class CursesTestResult(unittest.TestResult):
self.stream.flush()
def addSkip(self, test, reason): # pylint: disable=invalid-name
super(CursesTestResult, self).addSkip(test, reason)
super(QubesTestResult, self).addSkip(test, reason)
test.log.warning('skipped ({})'.format(reason))
if self.showAll:
self.stream.writeln(
'{color[cyan]}skipped{color[normal]} ({})'.format(
@ -155,7 +167,8 @@ class CursesTestResult(unittest.TestResult):
self.stream.flush()
def addExpectedFailure(self, test, err): # pylint: disable=invalid-name
super(CursesTestResult, self).addExpectedFailure(test, err)
super(QubesTestResult, self).addExpectedFailure(test, err)
test.log.warning('expected failure')
if self.showAll:
self.stream.writeln(
'{color[yellow]}expected failure{color[normal]}'.format(
@ -166,7 +179,8 @@ class CursesTestResult(unittest.TestResult):
self.stream.flush()
def addUnexpectedSuccess(self, test): # pylint: disable=invalid-name
super(CursesTestResult, self).addUnexpectedSuccess(test)
super(QubesTestResult, self).addUnexpectedSuccess(test)
test.log.error('unexpected success')
if self.showAll:
self.stream.writeln(
'{color[yellow]}{color[bold]}unexpected success'
@ -197,14 +211,36 @@ class CursesTestResult(unittest.TestResult):
def main():
ha_file = logging.FileHandler(
os.path.join(os.environ['HOME'], 'qubes-tests.log'))
ha_file.setFormatter(
logging.Formatter('%(asctime)s %(name)s[%(process)d]: %(message)s'))
logging.root.addHandler(ha_file)
ha_syslog = logging.handlers.SysLogHandler('/dev/log')
ha_syslog.setFormatter(
logging.Formatter('%(name)s[%(process)d]: %(message)s'))
try:
subprocess.check_call(('sudo', 'chmod', '666', '/dev/kmsg'))
except subprocess.CalledProcessError:
pass
else:
ha_kmsg = logging.FileHandler('/dev/kmsg', 'w')
ha_kmsg.setFormatter(
logging.Formatter('%(name)s[%(process)d]: %(message)s'))
ha_kmsg.setLevel(logging.CRITICAL)
logging.root.addHandler(ha_kmsg)
suite = unittest.TestSuite()
loader = unittest.TestLoader()
suite.addTests(loader.loadTestsFromName('qubes.tests'))
runner = unittest.TextTestRunner(stream=sys.stdout, verbosity=2)
unittest.signals.installHandler()
runner.resultclass = CursesTestResult
runner.resultclass = QubesTestResult
return runner.run(suite).wasSuccessful()
if __name__ == '__main__':
sys.exit(not main())

6
tests/vm_qrexec_gui.py
View File

@ -35,7 +35,7 @@ import qubes.tests
TEST_DATA = "0123456789" * 1024
class TC_00_AppVM(qubes.tests.SystemTestsMixin, unittest.TestCase):
class TC_00_AppVM(qubes.tests.SystemTestsMixin, qubes.tests.QubesTestCase):
def setUp(self):
super(TC_00_AppVM, self).setUp()
self.testvm1 = self.qc.add_new_vm("QubesAppVm",
@ -327,7 +327,7 @@ class TC_00_AppVM(qubes.tests.SystemTestsMixin, unittest.TestCase):
self.assertEqual(retcode, 0, "file differs")
class TC_10_HVM(qubes.tests.SystemTestsMixin, unittest.TestCase):
class TC_10_HVM(qubes.tests.SystemTestsMixin, qubes.tests.QubesTestCase):
# TODO: test with some OS inside
# TODO: windows tools tests
@ -379,7 +379,7 @@ class TC_10_HVM(qubes.tests.SystemTestsMixin, unittest.TestCase):
self.assertRaises(QubesException, self.templatevm.start)
class TC_20_DispVM(qubes.tests.SystemTestsMixin, unittest.TestCase):
class TC_20_DispVM(qubes.tests.SystemTestsMixin, qubes.tests.QubesTestCase):
def test_000_prepare_dvm(self):
self.qc.unlock_db()
retcode = subprocess.call(['/usr/bin/qvm-create-default-dvm',