__init__.py 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518
  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
  7. # Marek Marczykowski-Górecki <marmarek@invisiblethingslab.com>
  8. # Copyright (C) 2015 Wojtek Porczyk <woju@invisiblethingslab.com>
  9. #
  10. # This program is free software; you can redistribute it and/or modify
  11. # it under the terms of the GNU General Public License as published by
  12. # the Free Software Foundation; either version 2 of the License, or
  13. # (at your option) any later version.
  14. #
  15. # This program is distributed in the hope that it will be useful,
  16. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  17. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  18. # GNU General Public License for more details.
  19. #
  20. # You should have received a copy of the GNU General Public License along
  21. # with this program; if not, write to the Free Software Foundation, Inc.,
  22. # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  23. #
  24. import multiprocessing
  25. import logging
  26. import os
  27. import shutil
  28. import subprocess
  29. import unittest
  30. import lxml.etree
  31. import sys
  32. import qubes.backup
  33. import qubes.qubes
  34. VMPREFIX = 'test-'
  35. #: :py:obj:`True` if running in dom0, :py:obj:`False` otherwise
  36. in_dom0 = False
  37. #: :py:obj:`False` if outside of git repo,
  38. #: path to root of the directory otherwise
  39. in_git = False
  40. try:
  41. import libvirt
  42. libvirt.openReadOnly(qubes.qubes.defaults['libvirt_uri']).close()
  43. in_dom0 = True
  44. except libvirt.libvirtError:
  45. pass
  46. try:
  47. in_git = subprocess.check_output(
  48. ['git', 'rev-parse', '--show-toplevel']).strip()
  49. except subprocess.CalledProcessError:
  50. # git returned nonzero, we are outside git repo
  51. pass
  52. except OSError:
  53. # command not found; let's assume we're outside
  54. pass
  55. def skipUnlessDom0(test_item):
  56. '''Decorator that skips test outside dom0.
  57. Some tests (especially integration tests) have to be run in more or less
  58. working dom0. This is checked by connecting to libvirt.
  59. ''' # pylint: disable=invalid-name
  60. return unittest.skipUnless(in_dom0, 'outside dom0')(test_item)
  61. def skipUnlessGit(test_item):
  62. '''Decorator that skips test outside git repo.
  63. There are very few tests that an be run only in git. One example is
  64. correctness of example code that won't get included in RPM.
  65. ''' # pylint: disable=invalid-name
  66. return unittest.skipUnless(in_git, 'outside git tree')(test_item)
  67. class _AssertNotRaisesContext(object):
  68. """A context manager used to implement TestCase.assertNotRaises methods.
  69. Stolen from unittest and hacked. Regexp support stripped.
  70. """
  71. def __init__(self, expected, test_case, expected_regexp=None):
  72. self.expected = expected
  73. self.failureException = test_case.failureException
  74. def __enter__(self):
  75. return self
  76. def __exit__(self, exc_type, exc_value, tb):
  77. if exc_type is None:
  78. return True
  79. try:
  80. exc_name = self.expected.__name__
  81. except AttributeError:
  82. exc_name = str(self.expected)
  83. if issubclass(exc_type, self.expected):
  84. raise self.failureException(
  85. "{0} raised".format(exc_name))
  86. else:
  87. # pass through
  88. return False
  89. self.exception = exc_value # store for later retrieval
  90. class BeforeCleanExit(BaseException):
  91. pass
  92. class QubesTestCase(unittest.TestCase):
  93. '''Base class for Qubes unit tests.
  94. '''
  95. def __init__(self, *args, **kwargs):
  96. super(QubesTestCase, self).__init__(*args, **kwargs)
  97. self.longMessage = True
  98. self.log = logging.getLogger('{}.{}.{}'.format(
  99. self.__class__.__module__,
  100. self.__class__.__name__,
  101. self._testMethodName))
  102. def __str__(self):
  103. return '{}/{}/{}'.format(
  104. '.'.join(self.__class__.__module__.split('.')[2:]),
  105. self.__class__.__name__,
  106. self._testMethodName)
  107. def tearDown(self):
  108. super(QubesTestCase, self).tearDown()
  109. result = self._resultForDoCleanups
  110. l = result.failures \
  111. + result.errors \
  112. + [(tc, None) for tc in result.unexpectedSuccesses]
  113. if getattr(result, 'do_not_clean', False) \
  114. and filter((lambda (tc, exc): tc is self), l):
  115. raise BeforeCleanExit()
  116. def assertNotRaises(self, excClass, callableObj=None, *args, **kwargs):
  117. """Fail if an exception of class excClass is raised
  118. by callableObj when invoked with arguments args and keyword
  119. arguments kwargs. If a different type of exception is
  120. raised, it will not be caught, and the test case will be
  121. deemed to have suffered an error, exactly as for an
  122. unexpected exception.
  123. If called with callableObj omitted or None, will return a
  124. context object used like this::
  125. with self.assertRaises(SomeException):
  126. do_something()
  127. The context manager keeps a reference to the exception as
  128. the 'exception' attribute. This allows you to inspect the
  129. exception after the assertion::
  130. with self.assertRaises(SomeException) as cm:
  131. do_something()
  132. the_exception = cm.exception
  133. self.assertEqual(the_exception.error_code, 3)
  134. """
  135. context = _AssertNotRaisesContext(excClass, self)
  136. if callableObj is None:
  137. return context
  138. with context:
  139. callableObj(*args, **kwargs)
  140. def assertXMLEqual(self, xml1, xml2):
  141. '''Check for equality of two XML objects.
  142. :param xml1: first element
  143. :param xml2: second element
  144. :type xml1: :py:class:`lxml.etree._Element`
  145. :type xml2: :py:class:`lxml.etree._Element`
  146. ''' # pylint: disable=invalid-name
  147. self.assertEqual(xml1.tag, xml2.tag)
  148. self.assertEqual(xml1.text, xml2.text)
  149. self.assertItemsEqual(xml1.keys(), xml2.keys())
  150. for key in xml1.keys():
  151. self.assertEqual(xml1.get(key), xml2.get(key))
  152. class SystemTestsMixin(object):
  153. def setUp(self):
  154. '''Set up the test.
  155. .. warning::
  156. This method instantiates QubesVmCollection acquires write lock for
  157. it. You can use is as :py:attr:`qc`. You can (and probably
  158. should) release the lock at the end of setUp in subclass
  159. '''
  160. super(SystemTestsMixin, self).setUp()
  161. self.qc = qubes.qubes.QubesVmCollection()
  162. self.qc.lock_db_for_writing()
  163. self.qc.load()
  164. self.conn = libvirt.open(qubes.qubes.defaults['libvirt_uri'])
  165. self.remove_test_vms()
  166. def tearDown(self):
  167. super(SystemTestsMixin, self).tearDown()
  168. try: self.qc.lock_db_for_writing()
  169. except qubes.qubes.QubesException: pass
  170. self.qc.load()
  171. self.remove_test_vms()
  172. self.qc.save()
  173. self.qc.unlock_db()
  174. del self.qc
  175. self.conn.close()
  176. def make_vm_name(self, name):
  177. return VMPREFIX + name
  178. def save_and_reload_db(self):
  179. self.qc.save()
  180. self.qc.unlock_db()
  181. self.qc.lock_db_for_writing()
  182. self.qc.load()
  183. def _remove_vm_qubes(self, vm):
  184. vmname = vm.name
  185. try:
  186. # XXX .is_running() may throw libvirtError if undefined
  187. if vm.is_running():
  188. vm.force_shutdown()
  189. except: pass
  190. try: vm.remove_from_disk()
  191. except: pass
  192. try: vm.libvirt_domain.undefine()
  193. except libvirt.libvirtError: pass
  194. self.qc.pop(vm.qid)
  195. del vm
  196. # Now ensure it really went away. This may not have happened,
  197. # for example if vm.libvirtDomain malfunctioned.
  198. try:
  199. dom = self.conn.lookupByName(vmname)
  200. except: pass
  201. else:
  202. self._remove_vm_libvirt(dom)
  203. self._remove_vm_disk(vmname)
  204. def _remove_vm_libvirt(self, dom):
  205. try:
  206. dom.destroy()
  207. except libvirt.libvirtError: # not running
  208. pass
  209. dom.undefine()
  210. def _remove_vm_disk(self, vmname):
  211. for dirspec in (
  212. 'qubes_appvms_dir',
  213. 'qubes_servicevms_dir',
  214. 'qubes_templates_dir'):
  215. dirpath = os.path.join(qubes.qubes.system_path['qubes_base_dir'],
  216. qubes.qubes.system_path[dirspec], vmname)
  217. if os.path.exists(dirpath):
  218. if os.path.isdir(dirpath):
  219. shutil.rmtree(dirpath)
  220. else:
  221. os.unlink(dirpath)
  222. def remove_vms(self, vms):
  223. for vm in vms: self._remove_vm_qubes(vm)
  224. self.save_and_reload_db()
  225. def remove_test_vms(self):
  226. '''Aggresively remove any domain that has name in testing namespace.
  227. .. warning::
  228. The test suite hereby claims any domain whose name starts with
  229. :py:data:`VMPREFIX` as fair game. This is needed to enforce sane
  230. test executing environment. If you have domains named ``test-*``,
  231. don't run the tests.
  232. '''
  233. # first, remove them Qubes-way
  234. something_removed = False
  235. for vm in self.qc.values():
  236. if vm.name.startswith(VMPREFIX):
  237. self._remove_vm_qubes(vm)
  238. something_removed = True
  239. if something_removed:
  240. self.save_and_reload_db()
  241. # now remove what was only in libvirt
  242. for dom in self.conn.listAllDomains():
  243. if dom.name().startswith(VMPREFIX):
  244. self._remove_vm_libvirt(dom)
  245. # finally remove anything that is left on disk
  246. vmnames = set()
  247. for dirspec in (
  248. 'qubes_appvms_dir',
  249. 'qubes_servicevms_dir',
  250. 'qubes_templates_dir'):
  251. dirpath = os.path.join(qubes.qubes.system_path['qubes_base_dir'],
  252. qubes.qubes.system_path[dirspec])
  253. for name in os.listdir(dirpath):
  254. if name.startswith(VMPREFIX):
  255. vmnames.add(name)
  256. for vmname in vmnames:
  257. self._remove_vm_disk(vmname)
  258. class BackupTestsMixin(SystemTestsMixin):
  259. def setUp(self):
  260. super(BackupTestsMixin, self).setUp()
  261. self.error_detected = multiprocessing.Queue()
  262. self.verbose = False
  263. if self.verbose:
  264. print >>sys.stderr, "-> Creating backupvm"
  265. # TODO: allow non-default template
  266. self.backupvm = self.qc.add_new_vm("QubesAppVm",
  267. name=self.make_vm_name('backupvm'),
  268. template=self.qc.get_default_template())
  269. self.backupvm.create_on_disk(verbose=self.verbose)
  270. self.backupdir = os.path.join(os.environ["HOME"], "test-backup")
  271. if os.path.exists(self.backupdir):
  272. shutil.rmtree(self.backupdir)
  273. os.mkdir(self.backupdir)
  274. def tearDown(self):
  275. super(BackupTestsMixin, self).tearDown()
  276. shutil.rmtree(self.backupdir)
  277. def print_progress(self, progress):
  278. if self.verbose:
  279. print >> sys.stderr, "\r-> Backing up files: {0}%...".format(progress)
  280. def error_callback(self, message):
  281. self.error_detected.put(message)
  282. if self.verbose:
  283. print >> sys.stderr, "ERROR: {0}".format(message)
  284. def print_callback(self, msg):
  285. if self.verbose:
  286. print msg
  287. def fill_image(self, path, size=None, sparse=False):
  288. block_size = 4096
  289. if self.verbose:
  290. print >>sys.stderr, "-> Filling %s" % path
  291. f = open(path, 'w+')
  292. if size is None:
  293. f.seek(0, 2)
  294. size = f.tell()
  295. f.seek(0)
  296. for block_num in xrange(size/block_size):
  297. f.write('a' * block_size)
  298. if sparse:
  299. f.seek(block_size, 1)
  300. f.close()
  301. # NOTE: this was create_basic_vms
  302. def create_backup_vms(self):
  303. template=self.qc.get_default_template()
  304. vms = []
  305. vmname = self.make_vm_name('test1')
  306. if self.verbose:
  307. print >>sys.stderr, "-> Creating %s" % vmname
  308. testvm1 = self.qc.add_new_vm('QubesAppVm',
  309. name=vmname, template=template)
  310. testvm1.create_on_disk(verbose=self.verbose)
  311. vms.append(testvm1)
  312. self.fill_image(testvm1.private_img, 100*1024*1024)
  313. vmname = self.make_vm_name('testhvm1')
  314. if self.verbose:
  315. print >>sys.stderr, "-> Creating %s" % vmname
  316. testvm2 = self.qc.add_new_vm('QubesHVm', name=vmname)
  317. testvm2.create_on_disk(verbose=self.verbose)
  318. self.fill_image(testvm2.root_img, 1024*1024*1024, True)
  319. vms.append(testvm2)
  320. self.qc.save()
  321. return vms
  322. def make_backup(self, vms, prepare_kwargs=dict(), do_kwargs=dict(),
  323. target=None):
  324. # XXX: bakup_prepare and backup_do don't support host_collection
  325. self.qc.unlock_db()
  326. if target is None:
  327. target = self.backupdir
  328. try:
  329. files_to_backup = \
  330. qubes.backup.backup_prepare(vms,
  331. print_callback=self.print_callback,
  332. **prepare_kwargs)
  333. except qubes.qubes.QubesException as e:
  334. self.fail("QubesException during backup_prepare: %s" % str(e))
  335. try:
  336. qubes.backup.backup_do(target, files_to_backup, "qubes",
  337. progress_callback=self.print_progress,
  338. **do_kwargs)
  339. except qubes.qubes.QubesException as e:
  340. self.fail("QubesException during backup_do: %s" % str(e))
  341. self.qc.lock_db_for_writing()
  342. self.qc.load()
  343. def restore_backup(self, source=None, appvm=None, options=None):
  344. if source is None:
  345. backupfile = os.path.join(self.backupdir,
  346. sorted(os.listdir(self.backupdir))[-1])
  347. else:
  348. backupfile = source
  349. with self.assertNotRaises(qubes.qubes.QubesException):
  350. backup_info = qubes.backup.backup_restore_prepare(
  351. backupfile, "qubes",
  352. host_collection=self.qc,
  353. print_callback=self.print_callback,
  354. appvm=appvm,
  355. options=options or {})
  356. if self.verbose:
  357. qubes.backup.backup_restore_print_summary(backup_info)
  358. with self.assertNotRaises(qubes.qubes.QubesException):
  359. qubes.backup.backup_restore_do(
  360. backup_info,
  361. host_collection=self.qc,
  362. print_callback=self.print_callback if self.verbose else None,
  363. error_callback=self.error_callback)
  364. # maybe someone forgot to call .save()
  365. self.qc.load()
  366. errors = []
  367. while not self.error_detected.empty():
  368. errors.append(self.error_detected.get())
  369. self.assertTrue(len(errors) == 0,
  370. "Error(s) detected during backup_restore_do: %s" %
  371. '\n'.join(errors))
  372. if not appvm and not os.path.isdir(backupfile):
  373. os.unlink(backupfile)
  374. def create_sparse(self, path, size):
  375. f = open(path, "w")
  376. f.truncate(size)
  377. f.close()
  378. def load_tests(loader, tests, pattern):
  379. # discard any tests from this module, because it hosts base classes
  380. tests = unittest.TestSuite()
  381. for modname in (
  382. 'qubes.tests.basic',
  383. 'qubes.tests.dom0_update',
  384. 'qubes.tests.network',
  385. 'qubes.tests.vm_qrexec_gui',
  386. 'qubes.tests.backup',
  387. 'qubes.tests.backupcompatibility',
  388. 'qubes.tests.regressions',
  389. ):
  390. tests.addTests(loader.loadTestsFromName(modname))
  391. return tests
  392. # vim: ts=4 sw=4 et