__init__.py 8.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240
  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 collections
  24. import os
  25. import subprocess
  26. import unittest
  27. import lxml.etree
  28. import qubes.config
  29. import qubes.events
  30. #: :py:obj:`True` if running in dom0, :py:obj:`False` otherwise
  31. in_dom0 = False
  32. #: :py:obj:`False` if outside of git repo,
  33. #: path to root of the directory otherwise
  34. in_git = False
  35. try:
  36. import libvirt
  37. libvirt.openReadOnly(qubes.config.defaults['libvirt_uri']).close()
  38. in_dom0 = True
  39. del libvirt
  40. except libvirt.libvirtError:
  41. pass
  42. try:
  43. in_git = subprocess.check_output(
  44. ['git', 'rev-parse', '--show-toplevel']).strip()
  45. except subprocess.CalledProcessError:
  46. # git returned nonzero, we are outside git repo
  47. pass
  48. except OSError:
  49. # command not found; let's assume we're outside
  50. pass
  51. def skipUnlessDom0(test_item):
  52. '''Decorator that skips test outside dom0.
  53. Some tests (especially integration tests) have to be run in more or less
  54. working dom0. This is checked by connecting to libvirt.
  55. ''' # pylint: disable=invalid-name
  56. return unittest.skipUnless(in_dom0, 'outside dom0')(test_item)
  57. def skipUnlessGit(test_item):
  58. '''Decorator that skips test outside git repo.
  59. There are very few tests that an be run only in git. One example is
  60. correctness of example code that won't get included in RPM.
  61. ''' # pylint: disable=invalid-name
  62. return unittest.skipUnless(in_git, 'outside git tree')(test_item)
  63. class TestEmitter(qubes.events.Emitter):
  64. '''Dummy event emitter which records events fired on it.
  65. Events are counted in :py:attr:`fired_events` attribute, which is
  66. :py:class:`collections.Counter` instance. For each event, ``(event, args,
  67. kwargs)`` object is counted. *event* is event name (a string), *args* is
  68. tuple with positional arguments and *kwargs* is sorted tuple of items from
  69. keyword arguments.
  70. >>> emitter = TestEmitter()
  71. >>> emitter.fired_events
  72. Counter()
  73. >>> emitter.fire_event('event', 1, 2, 3, spam='eggs', foo='bar')
  74. >>> emitter.fired_events
  75. Counter({('event', (1, 2, 3), (('foo', 'bar'), ('spam', 'eggs'))): 1})
  76. '''
  77. def __init__(self, *args, **kwargs):
  78. super(TestEmitter, self).__init__(*args, **kwargs)
  79. #: :py:class:`collections.Counter` instance
  80. self.fired_events = collections.Counter()
  81. def fire_event(self, event, *args, **kwargs):
  82. super(TestEmitter, self).fire_event(event, *args, **kwargs)
  83. self.fired_events[(event, args, tuple(sorted(kwargs.items())))] += 1
  84. def fire_event_pre(self, event, *args, **kwargs):
  85. super(TestEmitter, self).fire_event_pre(event, *args, **kwargs)
  86. self.fired_events[(event, args, tuple(sorted(kwargs.items())))] += 1
  87. class QubesTestCase(unittest.TestCase):
  88. '''Base class for Qubes unit tests.
  89. '''
  90. def __str__(self):
  91. return '{}/{}/{}'.format(
  92. '.'.join(self.__class__.__module__.split('.')[2:]),
  93. self.__class__.__name__,
  94. self._testMethodName)
  95. def assertXMLEqual(self, xml1, xml2):
  96. '''Check for equality of two XML objects.
  97. :param xml1: first element
  98. :param xml2: second element
  99. :type xml1: :py:class:`lxml.etree._Element`
  100. :type xml2: :py:class:`lxml.etree._Element`
  101. ''' # pylint: disable=invalid-name
  102. self.assertEqual(xml1.tag, xml2.tag)
  103. self.assertEqual(xml1.text, xml2.text)
  104. self.assertItemsEqual(xml1.keys(), xml2.keys())
  105. for key in xml1.keys():
  106. self.assertEqual(xml1.get(key), xml2.get(key))
  107. def assertEventFired(self, emitter, event, args=None, kwargs=None):
  108. '''Check whether event was fired on given emitter and fail if it did
  109. not.
  110. :param emitter: emitter which is being checked
  111. :type emitter: :py:class:`TestEmitter`
  112. :param str event: event identifier
  113. :param list args: when given, all items must appear in args passed to \
  114. an event
  115. :param list kwargs: when given, all items must appear in kwargs passed \
  116. to an event
  117. ''' # pylint: disable=invalid-name
  118. for ev, ev_args, ev_kwargs in emitter.fired_events:
  119. if ev != event:
  120. continue
  121. if args is not None and any(i not in ev_args for i in args):
  122. continue
  123. if kwargs is not None and any(i not in ev_kwargs for i in kwargs):
  124. continue
  125. return
  126. self.fail('event {!r} did not fire on {!r}'.format(event, emitter))
  127. def assertEventNotFired(self, emitter, event, args=None, kwargs=None):
  128. '''Check whether event was fired on given emitter. Fail if it did.
  129. :param emitter: emitter which is being checked
  130. :type emitter: :py:class:`TestEmitter`
  131. :param str event: event identifier
  132. :param list args: when given, all items must appear in args passed to \
  133. an event
  134. :param list kwargs: when given, all items must appear in kwargs passed \
  135. to an event
  136. ''' # pylint: disable=invalid-name
  137. for ev, ev_args, ev_kwargs in emitter.fired_events:
  138. if ev != event:
  139. continue
  140. if args is not None and any(i not in ev_args for i in args):
  141. continue
  142. if kwargs is not None and any(i not in ev_kwargs for i in kwargs):
  143. continue
  144. self.fail('event {!r} did fire on {!r}'.format(event, emitter))
  145. return
  146. def assertXMLIsValid(self, xml, file=None, schema=None):
  147. '''Check whether given XML fulfills Relax NG schema.
  148. Schema can be given in a couple of ways:
  149. - As separate file. This is most common, and also the only way to
  150. handle file inclusion. Call with file name as second argument.
  151. - As string containing actual schema. Put that string in *schema*
  152. keyword argument.
  153. :param lxml.etree._Element xml: XML element instance to check
  154. :param str file: filename of Relax NG schema
  155. :param str schema: optional explicit schema string
  156. ''' # pylint: disable=invalid-name,redefined-builtin
  157. if schema is not None and file is None:
  158. relaxng = schema
  159. if isinstance(relaxng, str):
  160. relaxng = lxml.etree.XML(relaxng)
  161. # pylint: disable=protected-access
  162. if isinstance(relaxng, lxml.etree._Element):
  163. relaxng = lxml.etree.RelaxNG(relaxng)
  164. elif file is not None and schema is None:
  165. if not os.path.isabs(file):
  166. basedirs = ['/usr/share/doc/qubes/relaxng']
  167. if in_git:
  168. basedirs.insert(0, os.path.join(in_git, 'relaxng'))
  169. for basedir in basedirs:
  170. abspath = os.path.join(basedir, file)
  171. if os.path.exists(abspath):
  172. file = abspath
  173. break
  174. relaxng = lxml.etree.RelaxNG(file=file)
  175. else:
  176. raise TypeError("There should be excactly one of 'file' and "
  177. "'schema' arguments specified.")
  178. # We have to be extra careful here in case someone messed up with
  179. # self.failureException. It should by default be AssertionError, just
  180. # what is spewed by RelaxNG(), but who knows what might happen.
  181. try:
  182. relaxng.assert_(xml)
  183. except self.failureException:
  184. raise
  185. except AssertionError as e:
  186. self.fail(str(e))