#!/usr/bin/python2 -O # vim: fileencoding=utf-8 # # The Qubes OS Project, https://www.qubes-os.org/ # # Copyright (C) 2014-2015 Joanna Rutkowska # Copyright (C) 2014-2015 Wojtek Porczyk # # 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=[], kwargs=[]): '''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 any(i not in ev_args for i in args): continue if 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=[], kwargs=[]): '''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 any(i not in ev_args for i in args): continue if 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))