__init__.py 6.9 KB

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