__init__.py 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517
  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.log = logging.getLogger('{}.{}.{}'.format(
  98. self.__class__.__module__,
  99. self.__class__.__name__,
  100. self._testMethodName))
  101. def __str__(self):
  102. return '{}/{}/{}'.format(
  103. '.'.join(self.__class__.__module__.split('.')[2:]),
  104. self.__class__.__name__,
  105. self._testMethodName)
  106. def tearDown(self):
  107. super(QubesTestCase, self).tearDown()
  108. result = self._resultForDoCleanups
  109. l = result.failures \
  110. + result.errors \
  111. + [(tc, None) for tc in result.unexpectedSuccesses]
  112. if getattr(result, 'do_not_clean', False) \
  113. and filter((lambda (tc, exc): tc is self), l):
  114. raise BeforeCleanExit()
  115. def assertNotRaises(self, excClass, callableObj=None, *args, **kwargs):
  116. """Fail if an exception of class excClass is raised
  117. by callableObj when invoked with arguments args and keyword
  118. arguments kwargs. If a different type of exception is
  119. raised, it will not be caught, and the test case will be
  120. deemed to have suffered an error, exactly as for an
  121. unexpected exception.
  122. If called with callableObj omitted or None, will return a
  123. context object used like this::
  124. with self.assertRaises(SomeException):
  125. do_something()
  126. The context manager keeps a reference to the exception as
  127. the 'exception' attribute. This allows you to inspect the
  128. exception after the assertion::
  129. with self.assertRaises(SomeException) as cm:
  130. do_something()
  131. the_exception = cm.exception
  132. self.assertEqual(the_exception.error_code, 3)
  133. """
  134. context = _AssertNotRaisesContext(excClass, self)
  135. if callableObj is None:
  136. return context
  137. with context:
  138. callableObj(*args, **kwargs)
  139. def assertXMLEqual(self, xml1, xml2):
  140. '''Check for equality of two XML objects.
  141. :param xml1: first element
  142. :param xml2: second element
  143. :type xml1: :py:class:`lxml.etree._Element`
  144. :type xml2: :py:class:`lxml.etree._Element`
  145. ''' # pylint: disable=invalid-name
  146. self.assertEqual(xml1.tag, xml2.tag)
  147. self.assertEqual(xml1.text, xml2.text)
  148. self.assertItemsEqual(xml1.keys(), xml2.keys())
  149. for key in xml1.keys():
  150. self.assertEqual(xml1.get(key), xml2.get(key))
  151. class SystemTestsMixin(object):
  152. def setUp(self):
  153. '''Set up the test.
  154. .. warning::
  155. This method instantiates QubesVmCollection acquires write lock for
  156. it. You can use is as :py:attr:`qc`. You can (and probably
  157. should) release the lock at the end of setUp in subclass
  158. '''
  159. super(SystemTestsMixin, self).setUp()
  160. self.qc = qubes.qubes.QubesVmCollection()
  161. self.qc.lock_db_for_writing()
  162. self.qc.load()
  163. self.conn = libvirt.open(qubes.qubes.defaults['libvirt_uri'])
  164. self.remove_test_vms()
  165. def tearDown(self):
  166. super(SystemTestsMixin, self).tearDown()
  167. try: self.qc.lock_db_for_writing()
  168. except qubes.qubes.QubesException: pass
  169. self.qc.load()
  170. self.remove_test_vms()
  171. self.qc.save()
  172. self.qc.unlock_db()
  173. del self.qc
  174. self.conn.close()
  175. def make_vm_name(self, name):
  176. return VMPREFIX + name
  177. def save_and_reload_db(self):
  178. self.qc.save()
  179. self.qc.unlock_db()
  180. self.qc.lock_db_for_writing()
  181. self.qc.load()
  182. def _remove_vm_qubes(self, vm):
  183. vmname = vm.name
  184. try:
  185. # XXX .is_running() may throw libvirtError if undefined
  186. if vm.is_running():
  187. vm.force_shutdown()
  188. except: pass
  189. try: vm.remove_from_disk()
  190. except: pass
  191. try: vm.libvirt_domain.undefine()
  192. except libvirt.libvirtError: pass
  193. self.qc.pop(vm.qid)
  194. del vm
  195. # Now ensure it really went away. This may not have happened,
  196. # for example if vm.libvirtDomain malfunctioned.
  197. try:
  198. dom = self.conn.lookupByName(vmname)
  199. except: pass
  200. else:
  201. self._remove_vm_libvirt(dom)
  202. self._remove_vm_disk(vmname)
  203. def _remove_vm_libvirt(self, dom):
  204. try:
  205. dom.destroy()
  206. except libvirt.libvirtError: # not running
  207. pass
  208. dom.undefine()
  209. def _remove_vm_disk(self, vmname):
  210. for dirspec in (
  211. 'qubes_appvms_dir',
  212. 'qubes_servicevms_dir',
  213. 'qubes_templates_dir'):
  214. dirpath = os.path.join(qubes.qubes.system_path['qubes_base_dir'],
  215. qubes.qubes.system_path[dirspec], vmname)
  216. if os.path.exists(dirpath):
  217. if os.path.isdir(dirpath):
  218. shutil.rmtree(dirpath)
  219. else:
  220. os.unlink(dirpath)
  221. def remove_vms(self, vms):
  222. for vm in vms: self._remove_vm_qubes(vm)
  223. self.save_and_reload_db()
  224. def remove_test_vms(self):
  225. '''Aggresively remove any domain that has name in testing namespace.
  226. .. warning::
  227. The test suite hereby claims any domain whose name starts with
  228. :py:data:`VMPREFIX` as fair game. This is needed to enforce sane
  229. test executing environment. If you have domains named ``test-*``,
  230. don't run the tests.
  231. '''
  232. # first, remove them Qubes-way
  233. something_removed = False
  234. for vm in self.qc.values():
  235. if vm.name.startswith(VMPREFIX):
  236. self._remove_vm_qubes(vm)
  237. something_removed = True
  238. if something_removed:
  239. self.save_and_reload_db()
  240. # now remove what was only in libvirt
  241. for dom in self.conn.listAllDomains():
  242. if dom.name().startswith(VMPREFIX):
  243. self._remove_vm_libvirt(dom)
  244. # finally remove anything that is left on disk
  245. vmnames = set()
  246. for dirspec in (
  247. 'qubes_appvms_dir',
  248. 'qubes_servicevms_dir',
  249. 'qubes_templates_dir'):
  250. dirpath = os.path.join(qubes.qubes.system_path['qubes_base_dir'],
  251. qubes.qubes.system_path[dirspec])
  252. for name in os.listdir(dirpath):
  253. if name.startswith(VMPREFIX):
  254. vmnames.add(name)
  255. for vmname in vmnames:
  256. self._remove_vm_disk(vmname)
  257. class BackupTestsMixin(SystemTestsMixin):
  258. def setUp(self):
  259. super(BackupTestsMixin, self).setUp()
  260. self.error_detected = multiprocessing.Queue()
  261. self.verbose = False
  262. if self.verbose:
  263. print >>sys.stderr, "-> Creating backupvm"
  264. # TODO: allow non-default template
  265. self.backupvm = self.qc.add_new_vm("QubesAppVm",
  266. name=self.make_vm_name('backupvm'),
  267. template=self.qc.get_default_template())
  268. self.backupvm.create_on_disk(verbose=self.verbose)
  269. self.backupdir = os.path.join(os.environ["HOME"], "test-backup")
  270. if os.path.exists(self.backupdir):
  271. shutil.rmtree(self.backupdir)
  272. os.mkdir(self.backupdir)
  273. def tearDown(self):
  274. super(BackupTestsMixin, self).tearDown()
  275. shutil.rmtree(self.backupdir)
  276. def print_progress(self, progress):
  277. if self.verbose:
  278. print >> sys.stderr, "\r-> Backing up files: {0}%...".format(progress)
  279. def error_callback(self, message):
  280. self.error_detected.put(message)
  281. if self.verbose:
  282. print >> sys.stderr, "ERROR: {0}".format(message)
  283. def print_callback(self, msg):
  284. if self.verbose:
  285. print msg
  286. def fill_image(self, path, size=None, sparse=False):
  287. block_size = 4096
  288. if self.verbose:
  289. print >>sys.stderr, "-> Filling %s" % path
  290. f = open(path, 'w+')
  291. if size is None:
  292. f.seek(0, 2)
  293. size = f.tell()
  294. f.seek(0)
  295. for block_num in xrange(size/block_size):
  296. f.write('a' * block_size)
  297. if sparse:
  298. f.seek(block_size, 1)
  299. f.close()
  300. # NOTE: this was create_basic_vms
  301. def create_backup_vms(self):
  302. template=self.qc.get_default_template()
  303. vms = []
  304. vmname = self.make_vm_name('test1')
  305. if self.verbose:
  306. print >>sys.stderr, "-> Creating %s" % vmname
  307. testvm1 = self.qc.add_new_vm('QubesAppVm',
  308. name=vmname, template=template)
  309. testvm1.create_on_disk(verbose=self.verbose)
  310. vms.append(testvm1)
  311. self.fill_image(testvm1.private_img, 100*1024*1024)
  312. vmname = self.make_vm_name('testhvm1')
  313. if self.verbose:
  314. print >>sys.stderr, "-> Creating %s" % vmname
  315. testvm2 = self.qc.add_new_vm('QubesHVm', name=vmname)
  316. testvm2.create_on_disk(verbose=self.verbose)
  317. self.fill_image(testvm2.root_img, 1024*1024*1024, True)
  318. vms.append(testvm2)
  319. self.qc.save()
  320. return vms
  321. def make_backup(self, vms, prepare_kwargs=dict(), do_kwargs=dict(),
  322. target=None):
  323. # XXX: bakup_prepare and backup_do don't support host_collection
  324. self.qc.unlock_db()
  325. if target is None:
  326. target = self.backupdir
  327. try:
  328. files_to_backup = \
  329. qubes.backup.backup_prepare(vms,
  330. print_callback=self.print_callback,
  331. **prepare_kwargs)
  332. except qubes.qubes.QubesException as e:
  333. self.fail("QubesException during backup_prepare: %s" % str(e))
  334. try:
  335. qubes.backup.backup_do(target, files_to_backup, "qubes",
  336. progress_callback=self.print_progress,
  337. **do_kwargs)
  338. except qubes.qubes.QubesException as e:
  339. self.fail("QubesException during backup_do: %s" % str(e))
  340. self.qc.lock_db_for_writing()
  341. self.qc.load()
  342. def restore_backup(self, source=None, appvm=None, options=None):
  343. if source is None:
  344. backupfile = os.path.join(self.backupdir,
  345. sorted(os.listdir(self.backupdir))[-1])
  346. else:
  347. backupfile = source
  348. with self.assertNotRaises(qubes.qubes.QubesException):
  349. backup_info = qubes.backup.backup_restore_prepare(
  350. backupfile, "qubes",
  351. host_collection=self.qc,
  352. print_callback=self.print_callback,
  353. appvm=appvm,
  354. options=options or {})
  355. if self.verbose:
  356. qubes.backup.backup_restore_print_summary(backup_info)
  357. with self.assertNotRaises(qubes.qubes.QubesException):
  358. qubes.backup.backup_restore_do(
  359. backup_info,
  360. host_collection=self.qc,
  361. print_callback=self.print_callback if self.verbose else None,
  362. error_callback=self.error_callback)
  363. # maybe someone forgot to call .save()
  364. self.qc.load()
  365. errors = []
  366. while not self.error_detected.empty():
  367. errors.append(self.error_detected.get())
  368. self.assertTrue(len(errors) == 0,
  369. "Error(s) detected during backup_restore_do: %s" %
  370. '\n'.join(errors))
  371. if not appvm and not os.path.isdir(backupfile):
  372. os.unlink(backupfile)
  373. def create_sparse(self, path, size):
  374. f = open(path, "w")
  375. f.truncate(size)
  376. f.close()
  377. def load_tests(loader, tests, pattern):
  378. # discard any tests from this module, because it hosts base classes
  379. tests = unittest.TestSuite()
  380. for modname in (
  381. 'qubes.tests.basic',
  382. 'qubes.tests.dom0_update',
  383. 'qubes.tests.network',
  384. 'qubes.tests.vm_qrexec_gui',
  385. 'qubes.tests.backup',
  386. 'qubes.tests.backupcompatibility',
  387. 'qubes.tests.regressions',
  388. ):
  389. tests.addTests(loader.loadTestsFromName(modname))
  390. return tests
  391. # vim: ts=4 sw=4 et