241 lines
8.3 KiB
Python
241 lines
8.3 KiB
Python
#!/usr/bin/python2 -O
|
|
# vim: fileencoding=utf-8
|
|
|
|
#
|
|
# The Qubes OS Project, https://www.qubes-os.org/
|
|
#
|
|
# Copyright (C) 2014-2015 Joanna Rutkowska <joanna@invisiblethingslab.com>
|
|
# Copyright (C) 2014-2015 Wojtek Porczyk <woju@invisiblethingslab.com>
|
|
#
|
|
# This program is free software; you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation; either version 2 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License along
|
|
# with this program; if not, write to the Free Software Foundation, Inc.,
|
|
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
|
|
#
|
|
|
|
import collections
|
|
import os
|
|
import subprocess
|
|
import unittest
|
|
|
|
import lxml.etree
|
|
|
|
import qubes.config
|
|
import qubes.events
|
|
|
|
|
|
#: :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.config.defaults['libvirt_uri']).close()
|
|
in_dom0 = True
|
|
del libvirt
|
|
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 TestEmitter(qubes.events.Emitter):
|
|
'''Dummy event emitter which records events fired on it.
|
|
|
|
Events are counted in :py:attr:`fired_events` attribute, which is
|
|
:py:class:`collections.Counter` instance. For each event, ``(event, args,
|
|
kwargs)`` object is counted. *event* is event name (a string), *args* is
|
|
tuple with positional arguments and *kwargs* is sorted tuple of items from
|
|
keyword arguments.
|
|
|
|
>>> emitter = TestEmitter()
|
|
>>> emitter.fired_events
|
|
Counter()
|
|
>>> emitter.fire_event('event', 1, 2, 3, spam='eggs', foo='bar')
|
|
>>> emitter.fired_events
|
|
Counter({('event', (1, 2, 3), (('foo', 'bar'), ('spam', 'eggs'))): 1})
|
|
'''
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super(TestEmitter, self).__init__(*args, **kwargs)
|
|
|
|
#: :py:class:`collections.Counter` instance
|
|
self.fired_events = collections.Counter()
|
|
|
|
def fire_event(self, event, *args, **kwargs):
|
|
super(TestEmitter, self).fire_event(event, *args, **kwargs)
|
|
self.fired_events[(event, args, tuple(sorted(kwargs.items())))] += 1
|
|
|
|
def fire_event_pre(self, event, *args, **kwargs):
|
|
super(TestEmitter, self).fire_event_pre(event, *args, **kwargs)
|
|
self.fired_events[(event, args, tuple(sorted(kwargs.items())))] += 1
|
|
|
|
|
|
class QubesTestCase(unittest.TestCase):
|
|
'''Base class for Qubes unit tests.
|
|
'''
|
|
|
|
def __str__(self):
|
|
return '{}/{}/{}'.format(
|
|
'.'.join(self.__class__.__module__.split('.')[2:]),
|
|
self.__class__.__name__,
|
|
self._testMethodName)
|
|
|
|
|
|
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))
|
|
|
|
|
|
def assertEventFired(self, emitter, event, args=None, kwargs=None):
|
|
'''Check whether event was fired on given emitter and fail if it did
|
|
not.
|
|
|
|
:param emitter: emitter which is being checked
|
|
:type emitter: :py:class:`TestEmitter`
|
|
:param str event: event identifier
|
|
:param list args: when given, all items must appear in args passed to \
|
|
an event
|
|
:param list kwargs: when given, all items must appear in kwargs passed \
|
|
to an event
|
|
''' # pylint: disable=invalid-name
|
|
|
|
for ev, ev_args, ev_kwargs in emitter.fired_events:
|
|
if ev != event:
|
|
continue
|
|
if args is not None and any(i not in ev_args for i in args):
|
|
continue
|
|
if kwargs is not None and any(i not in ev_kwargs for i in kwargs):
|
|
continue
|
|
|
|
return
|
|
|
|
self.fail('event {!r} did not fire on {!r}'.format(event, emitter))
|
|
|
|
|
|
def assertEventNotFired(self, emitter, event, args=None, kwargs=None):
|
|
'''Check whether event was fired on given emitter. Fail if it did.
|
|
|
|
:param emitter: emitter which is being checked
|
|
:type emitter: :py:class:`TestEmitter`
|
|
:param str event: event identifier
|
|
:param list args: when given, all items must appear in args passed to \
|
|
an event
|
|
:param list kwargs: when given, all items must appear in kwargs passed \
|
|
to an event
|
|
''' # pylint: disable=invalid-name
|
|
|
|
for ev, ev_args, ev_kwargs in emitter.fired_events:
|
|
if ev != event:
|
|
continue
|
|
if args is not None and any(i not in ev_args for i in args):
|
|
continue
|
|
if kwargs is not None and any(i not in ev_kwargs for i in kwargs):
|
|
continue
|
|
|
|
self.fail('event {!r} did fire on {!r}'.format(event, emitter))
|
|
|
|
return
|
|
|
|
|
|
def assertXMLIsValid(self, xml, file=None, schema=None):
|
|
'''Check whether given XML fulfills Relax NG schema.
|
|
|
|
Schema can be given in a couple of ways:
|
|
|
|
- As separate file. This is most common, and also the only way to
|
|
handle file inclusion. Call with file name as second argument.
|
|
|
|
- As string containing actual schema. Put that string in *schema*
|
|
keyword argument.
|
|
|
|
:param lxml.etree._Element xml: XML element instance to check
|
|
:param str file: filename of Relax NG schema
|
|
:param str schema: optional explicit schema string
|
|
''' # pylint: disable=invalid-name,redefined-builtin
|
|
|
|
if schema is not None and file is None:
|
|
relaxng = schema
|
|
if isinstance(relaxng, str):
|
|
relaxng = lxml.etree.XML(relaxng)
|
|
# pylint: disable=protected-access
|
|
if isinstance(relaxng, lxml.etree._Element):
|
|
relaxng = lxml.etree.RelaxNG(relaxng)
|
|
|
|
elif file is not None and schema is None:
|
|
if not os.path.isabs(file):
|
|
basedirs = ['/usr/share/doc/qubes/relaxng']
|
|
if in_git:
|
|
basedirs.insert(0, os.path.join(in_git, 'relaxng'))
|
|
for basedir in basedirs:
|
|
abspath = os.path.join(basedir, file)
|
|
if os.path.exists(abspath):
|
|
file = abspath
|
|
break
|
|
relaxng = lxml.etree.RelaxNG(file=file)
|
|
|
|
else:
|
|
raise TypeError("There should be excactly one of 'file' and "
|
|
"'schema' arguments specified.")
|
|
|
|
# We have to be extra careful here in case someone messed up with
|
|
# self.failureException. It should by default be AssertionError, just
|
|
# what is spewed by RelaxNG(), but who knows what might happen.
|
|
try:
|
|
relaxng.assert_(xml)
|
|
except self.failureException:
|
|
raise
|
|
except AssertionError as e:
|
|
self.fail(str(e))
|