qvm_backup_restore.py 9.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233
  1. # -*- encoding: utf8 -*-
  2. #
  3. # The Qubes OS Project, http://www.qubes-os.org
  4. #
  5. # Copyright (C) 2017 Marek Marczykowski-Górecki
  6. # <marmarek@invisiblethingslab.com>
  7. #
  8. # This program is free software; you can redistribute it and/or modify
  9. # it under the terms of the GNU Lesser General Public License as published by
  10. # the Free Software Foundation; either version 2.1 of the License, or
  11. # (at your option) any later version.
  12. #
  13. # This program is distributed in the hope that it will be useful,
  14. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  15. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  16. # GNU Lesser General Public License for more details.
  17. #
  18. # You should have received a copy of the GNU Lesser General Public License along
  19. # with this program; if not, see <http://www.gnu.org/licenses/>.
  20. import qubesadmin.tests
  21. import qubesadmin.tests.tools
  22. import qubesadmin.tools.qvm_backup_restore
  23. from unittest import mock
  24. from qubesadmin.backup import BackupVM
  25. from qubesadmin.backup.restore import BackupRestore
  26. class TC_00_qvm_backup_restore(qubesadmin.tests.QubesTestCase):
  27. def setUp(self):
  28. super(TC_00_qvm_backup_restore, self).setUp()
  29. def tearDown(self):
  30. super(TC_00_qvm_backup_restore, self).tearDown()
  31. @mock.patch('qubesadmin.tools.qvm_backup_restore.input', create=True)
  32. @mock.patch('getpass.getpass')
  33. @mock.patch('qubesadmin.tools.qvm_backup_restore.BackupRestore')
  34. def test_000_simple(self, mock_backup, mock_getpass, mock_input):
  35. mock_getpass.return_value = 'testpass'
  36. mock_input.return_value = 'Y'
  37. vm1 = BackupVM()
  38. vm1.name = 'test-vm'
  39. vm1.backup_path = 'path/in/backup'
  40. vm1.template = None
  41. vm1.klass = 'StandaloneVM'
  42. vm1.label = 'red'
  43. mock_restore_info = {
  44. 1: BackupRestore.VMToRestore(vm1),
  45. }
  46. mock_backup.configure_mock(**{
  47. 'return_value.get_restore_summary.return_value': '',
  48. 'return_value.get_restore_info.return_value': mock_restore_info,
  49. })
  50. with mock.patch('qubesadmin.tools.qvm_backup_restore.handle_broken') \
  51. as mock_handle_broken:
  52. qubesadmin.tools.qvm_backup_restore.main(['/some/path'],
  53. app=self.app)
  54. mock_handle_broken.assert_called_once_with(
  55. self.app, mock.ANY, mock_restore_info)
  56. mock_backup.assert_called_once_with(
  57. self.app, '/some/path', None, 'testpass',
  58. force_compression_filter=None)
  59. self.assertAllCalled()
  60. @mock.patch('qubesadmin.tools.qvm_backup_restore.input', create=True)
  61. @mock.patch('getpass.getpass')
  62. @mock.patch('qubesadmin.tools.qvm_backup_restore.BackupRestore')
  63. def test_001_selected_vms(self, mock_backup, mock_getpass, mock_input):
  64. mock_getpass.return_value = 'testpass'
  65. mock_input.return_value = 'Y'
  66. vm1 = BackupVM()
  67. vm1.name = 'test-vm'
  68. vm1.backup_path = 'path/in/backup'
  69. vm1.template = None
  70. vm1.klass = 'StandaloneVM'
  71. vm1.label = 'red'
  72. vm2 = BackupVM()
  73. vm2.name = 'test-vm2'
  74. vm2.backup_path = 'path/in/backup2'
  75. vm2.template = None
  76. vm2.klass = 'StandaloneVM'
  77. vm2.label = 'red'
  78. mock_restore_info = {
  79. 1: BackupRestore.VMToRestore(vm1),
  80. 2: BackupRestore.VMToRestore(vm2),
  81. }
  82. exclude_list = []
  83. mock_backup.configure_mock(**{
  84. 'return_value.get_restore_summary.return_value': '',
  85. 'return_value.options.exclude': exclude_list,
  86. 'return_value.get_restore_info.return_value': mock_restore_info,
  87. })
  88. qubesadmin.tools.qvm_backup_restore.main(['/some/path', 'test-vm'],
  89. app=self.app)
  90. mock_backup.assert_called_once_with(
  91. self.app, '/some/path', None, 'testpass',
  92. force_compression_filter=None)
  93. self.assertEqual(mock_backup.return_value.options.exclude, ['test-vm2'])
  94. self.assertAllCalled()
  95. def test_010_handle_broken_no_problems(self):
  96. vm1 = BackupVM()
  97. vm1.name = 'test-vm'
  98. vm1.backup_path = 'path/in/backup'
  99. vm1.template = None
  100. vm1.klass = 'StandaloneVM'
  101. vm1.label = 'red'
  102. mock_restore_info = {
  103. 1: BackupRestore.VMToRestore(vm1),
  104. }
  105. mock_args = mock.Mock()
  106. mock_args.verify_only = False
  107. self.app.log = mock.Mock()
  108. qubesadmin.tools.qvm_backup_restore.handle_broken(
  109. self.app, mock_args, mock_restore_info)
  110. self.assertEqual(self.app.log.mock_calls, [
  111. mock.call.info(
  112. 'The above VMs will be copied and added to your system.'),
  113. mock.call.info(
  114. 'Exisiting VMs will NOT be removed.'),
  115. ])
  116. def assertAppropriateLogging(self, missing_name, action):
  117. '''
  118. :param missing_name: NetVM or TemplateVM
  119. :param action: 'skip_broken', 'ignore_missing'
  120. :return:
  121. '''
  122. expected_calls = [
  123. mock.call.info(
  124. 'The above VMs will be copied and added to your system.'),
  125. mock.call.info(
  126. 'Exisiting VMs will NOT be removed.'),
  127. mock.call.warning(
  128. '*** One or more {}s are missing on the host! ***'.format(
  129. missing_name)),
  130. ]
  131. if action == 'skip_broken':
  132. expected_calls.append(
  133. mock.call.warning(
  134. 'Skipping broken entries: VMs that depend on missing {}s '
  135. 'will NOT be restored.'.format(missing_name))
  136. )
  137. elif action == 'ignore_missing':
  138. expected_calls.append(
  139. mock.call.warning(
  140. 'Ignoring missing entries: VMs that depend on missing '
  141. '{}s will have default value assigned.'.format(
  142. missing_name))
  143. )
  144. self.assertEqual(self.app.log.mock_calls, expected_calls)
  145. def test_011_handle_broken_missing_template(self):
  146. vm1 = BackupVM()
  147. vm1.name = 'test-vm'
  148. vm1.backup_path = 'path/in/backup'
  149. vm1.template = 'not-existing-template'
  150. vm1.klass = 'AppVM'
  151. vm1.label = 'red'
  152. mock_restore_info = {
  153. 1: BackupRestore.VMToRestore(vm1),
  154. }
  155. mock_restore_info[1].problems.add(
  156. BackupRestore.VMToRestore.MISSING_TEMPLATE)
  157. with self.subTest('skip_broken'):
  158. mock_args = mock.Mock()
  159. mock_args.skip_broken = True
  160. mock_args.verify_only = False
  161. self.app.log = mock.Mock()
  162. qubesadmin.tools.qvm_backup_restore.handle_broken(
  163. self.app, mock_args, mock_restore_info)
  164. self.assertAppropriateLogging('TemplateVM', 'skip_broken')
  165. with self.subTest('ignore_missing'):
  166. mock_args = mock.Mock()
  167. mock_args.skip_broken = False
  168. mock_args.ignore_missing = True
  169. mock_args.verify_only = False
  170. self.app.log = mock.Mock()
  171. qubesadmin.tools.qvm_backup_restore.handle_broken(
  172. self.app, mock_args, mock_restore_info)
  173. self.assertAppropriateLogging('TemplateVM', 'ignore_missing')
  174. with self.subTest('error'):
  175. mock_args = mock.Mock()
  176. mock_args.skip_broken = False
  177. mock_args.ignore_missing = False
  178. mock_args.verify_only = False
  179. self.app.log = mock.Mock()
  180. with self.assertRaises(qubesadmin.exc.QubesException):
  181. qubesadmin.tools.qvm_backup_restore.handle_broken(
  182. self.app, mock_args, mock_restore_info)
  183. self.assertAppropriateLogging('TemplateVM', 'error')
  184. def test_012_handle_broken_missing_netvm(self):
  185. vm1 = BackupVM()
  186. vm1.name = 'test-vm'
  187. vm1.backup_path = 'path/in/backup'
  188. vm1.netvm = 'not-existing-netvm'
  189. vm1.klass = 'StandaloneVM'
  190. vm1.label = 'red'
  191. mock_restore_info = {
  192. 1: BackupRestore.VMToRestore(vm1),
  193. }
  194. mock_restore_info[1].problems.add(
  195. BackupRestore.VMToRestore.MISSING_NETVM)
  196. with self.subTest('skip_broken'):
  197. mock_args = mock.Mock()
  198. mock_args.skip_broken = True
  199. mock_args.verify_only = False
  200. self.app.log = mock.Mock()
  201. qubesadmin.tools.qvm_backup_restore.handle_broken(
  202. self.app, mock_args, mock_restore_info)
  203. self.assertAppropriateLogging('NetVM', 'skip_broken')
  204. with self.subTest('ignore_missing'):
  205. mock_args = mock.Mock()
  206. mock_args.skip_broken = False
  207. mock_args.ignore_missing = True
  208. mock_args.verify_only = False
  209. self.app.log = mock.Mock()
  210. qubesadmin.tools.qvm_backup_restore.handle_broken(
  211. self.app, mock_args, mock_restore_info)
  212. self.assertAppropriateLogging('NetVM', 'ignore_missing')
  213. with self.subTest('error'):
  214. mock_args = mock.Mock()
  215. mock_args.skip_broken = False
  216. mock_args.ignore_missing = False
  217. mock_args.verify_only = False
  218. self.app.log = mock.Mock()
  219. with self.assertRaises(qubesadmin.exc.QubesException):
  220. qubesadmin.tools.qvm_backup_restore.handle_broken(
  221. self.app, mock_args, mock_restore_info)
  222. self.assertAppropriateLogging('NetVM', 'error')