12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592 |
- # -*- encoding: utf8 -*-
- #
- # The Qubes OS Project, http://www.qubes-os.org
- #
- # Copyright (C) 2017 Marek Marczykowski-Górecki
- # <marmarek@invisiblethingslab.com>
- #
- # This program is free software; you can redistribute it and/or modify
- # it under the terms of the GNU General Public License as published by
- # the Free Software Foundation; either version 2 of the License, or
- # (at your option) any later version.
- #
- # This program is distributed in the hope that it will be useful,
- # but WITHOUT ANY WARRANTY; without even the implied warranty of
- # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- # GNU General Public License for more details.
- #
- # You should have received a copy of the GNU General Public License along
- # with this program; if not, see <http://www.gnu.org/licenses/>.
- ''' Tests for management calls endpoints '''
- import asyncio
- import os
- import shutil
- import unittest.mock
- import libvirt
- import qubes
- import qubes.api.admin
- import qubes.tests
- # properties defined in API
- volume_properties = [
- 'pool', 'vid', 'size', 'usage', 'rw', 'internal', 'source',
- 'save_on_stop', 'snap_on_start']
- class AdminAPITestCase(qubes.tests.QubesTestCase):
- def setUp(self):
- super().setUp()
- app = qubes.Qubes('/tmp/qubes-test.xml', load=False)
- app.vmm = unittest.mock.Mock(spec=qubes.app.VMMConnection)
- app.load_initial_values()
- app.default_kernel = '1.0'
- app.default_netvm = None
- self.template = app.add_new_vm('TemplateVM', label='black',
- name='test-template')
- app.default_template = 'test-template'
- with qubes.tests.substitute_entry_points('qubes.storage',
- 'qubes.tests.storage'):
- app.add_pool('test', driver='test')
- app.save = unittest.mock.Mock()
- self.vm = app.add_new_vm('AppVM', label='red', name='test-vm1',
- template='test-template')
- self.app = app
- libvirt_attrs = {
- 'libvirt_conn.lookupByUUID.return_value.isActive.return_value':
- False,
- 'libvirt_conn.lookupByUUID.return_value.state.return_value':
- [libvirt.VIR_DOMAIN_SHUTOFF],
- }
- app.vmm.configure_mock(**libvirt_attrs)
- self.emitter = qubes.tests.TestEmitter()
- self.app.domains[0].fire_event = self.emitter.fire_event
- self.app.domains[0].fire_event_pre = self.emitter.fire_event_pre
- self.test_base_dir = '/tmp/qubes-test-dir'
- self.base_dir_patch = unittest.mock.patch.dict(qubes.config.system_path,
- {'qubes_base_dir': self.test_base_dir})
- self.base_dir_patch.start()
- def tearDown(self):
- self.base_dir_patch.stop()
- if os.path.exists(self.test_base_dir):
- shutil.rmtree(self.test_base_dir)
- super(AdminAPITestCase, self).tearDown()
- def call_mgmt_func(self, method, dest, arg=b'', payload=b''):
- mgmt_obj = qubes.api.admin.QubesAdminAPI(self.app, b'dom0', method, dest, arg)
- loop = asyncio.get_event_loop()
- response = loop.run_until_complete(
- mgmt_obj.execute(untrusted_payload=payload))
- self.assertEventFired(self.emitter,
- 'mgmt-permission:' + method.decode('ascii'))
- return response
- class TC_00_VMs(AdminAPITestCase):
- def test_000_vm_list(self):
- value = self.call_mgmt_func(b'admin.vm.List', b'dom0')
- self.assertEqual(value,
- 'dom0 class=AdminVM state=Running\n'
- 'test-template class=TemplateVM state=Halted\n'
- 'test-vm1 class=AppVM state=Halted\n')
- def test_001_vm_list_single(self):
- value = self.call_mgmt_func(b'admin.vm.List', b'test-vm1')
- self.assertEqual(value,
- 'test-vm1 class=AppVM state=Halted\n')
- def test_010_vm_property_list(self):
- # this test is kind of stupid, but at least check if appropriate
- # mgmt-permission event is fired
- value = self.call_mgmt_func(b'admin.vm.property.List', b'test-vm1')
- properties = self.app.domains['test-vm1'].property_list()
- self.assertEqual(value,
- ''.join('{}\n'.format(prop.__name__) for prop in properties))
- def test_020_vm_property_get_str(self):
- value = self.call_mgmt_func(b'admin.vm.property.Get', b'test-vm1',
- b'name')
- self.assertEqual(value, 'default=False type=str test-vm1')
- def test_021_vm_property_get_int(self):
- value = self.call_mgmt_func(b'admin.vm.property.Get', b'test-vm1',
- b'vcpus')
- self.assertEqual(value, 'default=True type=int 42')
- def test_022_vm_property_get_bool(self):
- value = self.call_mgmt_func(b'admin.vm.property.Get', b'test-vm1',
- b'provides_network')
- self.assertEqual(value, 'default=True type=bool False')
- def test_023_vm_property_get_label(self):
- value = self.call_mgmt_func(b'admin.vm.property.Get', b'test-vm1',
- b'label')
- self.assertEqual(value, 'default=False type=label red')
- def test_024_vm_property_get_vm(self):
- value = self.call_mgmt_func(b'admin.vm.property.Get', b'test-vm1',
- b'template')
- self.assertEqual(value, 'default=False type=vm test-template')
- def test_025_vm_property_get_vm_none(self):
- value = self.call_mgmt_func(b'admin.vm.property.Get', b'test-vm1',
- b'netvm')
- self.assertEqual(value, 'default=True type=vm ')
- def test_030_vm_property_set_vm(self):
- netvm = self.app.add_new_vm('AppVM', label='red', name='test-net',
- template='test-template', provides_network=True)
- with unittest.mock.patch('qubes.vm.VMProperty.__set__') as mock:
- value = self.call_mgmt_func(b'admin.vm.property.Set', b'test-vm1',
- b'netvm', b'test-net')
- self.assertIsNone(value)
- mock.assert_called_once_with(self.vm, 'test-net')
- self.app.save.assert_called_once_with()
- def test_032_vm_property_set_vm_invalid1(self):
- with unittest.mock.patch('qubes.vm.VMProperty.__set__') as mock:
- with self.assertRaises(qubes.exc.QubesValueError):
- self.call_mgmt_func(b'admin.vm.property.Set', b'test-vm1',
- b'netvm', b'forbidden-chars/../!')
- self.assertFalse(mock.called)
- self.assertFalse(self.app.save.called)
- def test_033_vm_property_set_vm_invalid2(self):
- with unittest.mock.patch('qubes.vm.VMProperty.__set__') as mock:
- with self.assertRaises(qubes.exc.QubesValueError):
- self.call_mgmt_func(b'admin.vm.property.Set', b'test-vm1',
- b'netvm', b'\x80\x90\xa0')
- self.assertFalse(mock.called)
- self.assertFalse(self.app.save.called)
- def test_034_vm_propert_set_bool_true(self):
- with unittest.mock.patch('qubes.property.__set__') as mock:
- value = self.call_mgmt_func(b'admin.vm.property.Set', b'test-vm1',
- b'autostart', b'True')
- self.assertIsNone(value)
- mock.assert_called_once_with(self.vm, True)
- self.app.save.assert_called_once_with()
- def test_035_vm_propert_set_bool_false(self):
- with unittest.mock.patch('qubes.property.__set__') as mock:
- value = self.call_mgmt_func(b'admin.vm.property.Set', b'test-vm1',
- b'autostart', b'False')
- self.assertIsNone(value)
- mock.assert_called_once_with(self.vm, False)
- self.app.save.assert_called_once_with()
- def test_036_vm_propert_set_bool_invalid1(self):
- with unittest.mock.patch('qubes.property.__set__') as mock:
- with self.assertRaises(qubes.exc.QubesValueError):
- self.call_mgmt_func(b'admin.vm.property.Set', b'test-vm1',
- b'autostart', b'some string')
- self.assertFalse(mock.called)
- self.assertFalse(self.app.save.called)
- def test_037_vm_propert_set_bool_invalid2(self):
- with unittest.mock.patch('qubes.property.__set__') as mock:
- with self.assertRaises(qubes.exc.QubesValueError):
- self.call_mgmt_func(b'admin.vm.property.Set', b'test-vm1',
- b'autostart', b'\x80\x90@#$%^&*(')
- self.assertFalse(mock.called)
- self.assertFalse(self.app.save.called)
- def test_038_vm_propert_set_str(self):
- with unittest.mock.patch('qubes.property.__set__') as mock:
- value = self.call_mgmt_func(b'admin.vm.property.Set', b'test-vm1',
- b'kernel', b'1.0')
- self.assertIsNone(value)
- mock.assert_called_once_with(self.vm, '1.0')
- self.app.save.assert_called_once_with()
- def test_039_vm_propert_set_str_invalid1(self):
- with unittest.mock.patch('qubes.property.__set__') as mock:
- with self.assertRaises(qubes.exc.QubesValueError):
- self.call_mgmt_func(b'admin.vm.property.Set', b'test-vm1',
- b'kernel', b'some, non-ASCII: \x80\xd2')
- self.assertFalse(mock.called)
- self.assertFalse(self.app.save.called)
- def test_040_vm_propert_set_int(self):
- with unittest.mock.patch('qubes.property.__set__') as mock:
- value = self.call_mgmt_func(b'admin.vm.property.Set', b'test-vm1',
- b'maxmem', b'1024000')
- self.assertIsNone(value)
- mock.assert_called_once_with(self.vm, 1024000)
- self.app.save.assert_called_once_with()
- def test_041_vm_propert_set_int_invalid1(self):
- with unittest.mock.patch('qubes.property.__set__') as mock:
- with self.assertRaises(qubes.exc.QubesValueError):
- self.call_mgmt_func(b'admin.vm.property.Set', b'test-vm1',
- b'maxmem', b'fourty two')
- self.assertFalse(mock.called)
- self.assertFalse(self.app.save.called)
- def test_042_vm_propert_set_label(self):
- with unittest.mock.patch('qubes.property.__set__') as mock:
- value = self.call_mgmt_func(b'admin.vm.property.Set', b'test-vm1',
- b'label', b'green')
- self.assertIsNone(value)
- mock.assert_called_once_with(self.vm, 'green')
- self.app.save.assert_called_once_with()
- def test_043_vm_propert_set_label_invalid1(self):
- with unittest.mock.patch('qubes.property.__set__') as mock:
- with self.assertRaises(qubes.exc.QubesValueError):
- self.call_mgmt_func(b'admin.vm.property.Set', b'test-vm1',
- b'maxmem', b'some, non-ASCII: \x80\xd2')
- self.assertFalse(mock.called)
- self.assertFalse(self.app.save.called)
- @unittest.skip('label existence not checked before actual setter yet')
- def test_044_vm_propert_set_label_invalid2(self):
- with unittest.mock.patch('qubes.property.__set__') as mock:
- with self.assertRaises(qubes.exc.QubesValueError):
- self.call_mgmt_func(b'admin.vm.property.Set', b'test-vm1',
- b'maxmem', b'non-existing-color')
- self.assertFalse(mock.called)
- self.assertFalse(self.app.save.called)
- def test_050_vm_property_help(self):
- value = self.call_mgmt_func(b'admin.vm.property.Help', b'test-vm1',
- b'label')
- self.assertEqual(value,
- 'Colourful label assigned to VM. This is where the colour of the '
- 'padlock is set.')
- self.assertFalse(self.app.save.called)
- def test_052_vm_property_help_invalid_property(self):
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.vm.property.Help', b'test-vm1',
- b'no-such-property')
- self.assertFalse(self.app.save.called)
- def test_060_vm_property_reset(self):
- with unittest.mock.patch('qubes.property.__delete__') as mock:
- value = self.call_mgmt_func(b'admin.vm.property.Reset', b'test-vm1',
- b'default_user')
- mock.assert_called_with(self.vm)
- self.assertIsNone(value)
- self.app.save.assert_called_once_with()
- def test_062_vm_property_reset_invalid_property(self):
- with unittest.mock.patch('qubes.property.__delete__') as mock:
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.vm.property.Help', b'test-vm1',
- b'no-such-property')
- self.assertFalse(mock.called)
- self.assertFalse(self.app.save.called)
- def test_070_vm_volume_list(self):
- self.vm.volumes = unittest.mock.Mock()
- volumes_conf = {
- 'keys.return_value': ['root', 'private', 'volatile', 'kernel']
- }
- self.vm.volumes.configure_mock(**volumes_conf)
- value = self.call_mgmt_func(b'admin.vm.volume.List', b'test-vm1')
- self.assertEqual(value, 'root\nprivate\nvolatile\nkernel\n')
- # check if _only_ keys were accessed
- self.assertEqual(self.vm.volumes.mock_calls,
- [unittest.mock.call.keys()])
- def test_080_vm_volume_info(self):
- self.vm.volumes = unittest.mock.MagicMock()
- volumes_conf = {
- 'keys.return_value': ['root', 'private', 'volatile', 'kernel']
- }
- for prop in volume_properties:
- volumes_conf[
- '__getitem__.return_value.{}'.format(prop)] = prop +'-value'
- self.vm.volumes.configure_mock(**volumes_conf)
- value = self.call_mgmt_func(b'admin.vm.volume.Info', b'test-vm1',
- b'private')
- self.assertEqual(value,
- ''.join('{p}={p}-value\n'.format(p=p) for p in volume_properties))
- self.assertEqual(self.vm.volumes.mock_calls,
- [unittest.mock.call.keys(),
- unittest.mock.call.__getattr__('__getitem__')('private')])
- def test_080_vm_volume_info_invalid_volume(self):
- self.vm.volumes = unittest.mock.MagicMock()
- volumes_conf = {
- 'keys.return_value': ['root', 'private', 'volatile', 'kernel']
- }
- self.vm.volumes.configure_mock(**volumes_conf)
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.vm.volume.Info', b'test-vm1',
- b'no-such-volume')
- self.assertEqual(self.vm.volumes.mock_calls,
- [unittest.mock.call.keys()])
- def test_090_vm_volume_listsnapshots(self):
- self.vm.volumes = unittest.mock.MagicMock()
- volumes_conf = {
- 'keys.return_value': ['root', 'private', 'volatile', 'kernel'],
- '__getitem__.return_value.revisions': ['rev1', 'rev2'],
- }
- self.vm.volumes.configure_mock(**volumes_conf)
- value = self.call_mgmt_func(b'admin.vm.volume.ListSnapshots',
- b'test-vm1', b'private')
- self.assertEqual(value,
- 'rev1\nrev2\n')
- self.assertEqual(self.vm.volumes.mock_calls,
- [unittest.mock.call.keys(),
- unittest.mock.call.__getattr__('__getitem__')('private')])
- def test_090_vm_volume_listsnapshots_invalid_volume(self):
- self.vm.volumes = unittest.mock.MagicMock()
- volumes_conf = {
- 'keys.return_value': ['root', 'private', 'volatile', 'kernel']
- }
- self.vm.volumes.configure_mock(**volumes_conf)
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.vm.volume.ListSnapshots', b'test-vm1',
- b'no-such-volume')
- self.assertEqual(self.vm.volumes.mock_calls,
- [unittest.mock.call.keys()])
- @unittest.skip('method not implemented yet')
- def test_100_vm_volume_snapshot(self):
- pass
- @unittest.skip('method not implemented yet')
- def test_100_vm_volume_snapshot_invlid_volume(self):
- self.vm.volumes = unittest.mock.MagicMock()
- volumes_conf = {
- 'keys.return_value': ['root', 'private', 'volatile', 'kernel'],
- '__getitem__.return_value.revisions': ['rev1', 'rev2'],
- }
- self.vm.volumes.configure_mock(**volumes_conf)
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.vm.volume.Snapshots',
- b'test-vm1', b'no-such-volume')
- self.assertEqual(self.vm.volumes.mock_calls,
- [unittest.mock.call.keys()])
- @unittest.skip('method not implemented yet')
- def test_100_vm_volume_snapshot_invalid_revision(self):
- self.vm.volumes = unittest.mock.MagicMock()
- volumes_conf = {
- 'keys.return_value': ['root', 'private', 'volatile', 'kernel']
- }
- self.vm.volumes.configure_mock(**volumes_conf)
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.vm.volume.Snapshots',
- b'test-vm1', b'private', b'no-such-rev')
- self.assertEqual(self.vm.volumes.mock_calls,
- [unittest.mock.call.keys(),
- unittest.mock.call.__getattr__('__getitem__')('private')])
- def test_110_vm_volume_revert(self):
- self.vm.volumes = unittest.mock.MagicMock()
- volumes_conf = {
- 'keys.return_value': ['root', 'private', 'volatile', 'kernel'],
- '__getitem__.return_value.revisions': ['rev1', 'rev2'],
- }
- self.vm.volumes.configure_mock(**volumes_conf)
- self.vm.storage = unittest.mock.Mock()
- value = self.call_mgmt_func(b'admin.vm.volume.Revert',
- b'test-vm1', b'private', b'rev1')
- self.assertIsNone(value)
- self.assertEqual(self.vm.volumes.mock_calls,
- [unittest.mock.call.keys(),
- unittest.mock.call.__getattr__('__getitem__')('private')])
- self.assertEqual(self.vm.storage.mock_calls,
- [unittest.mock.call.get_pool(self.vm.volumes['private']),
- unittest.mock.call.get_pool().revert('rev1')])
- def test_110_vm_volume_revert_invalid_rev(self):
- self.vm.volumes = unittest.mock.MagicMock()
- volumes_conf = {
- 'keys.return_value': ['root', 'private', 'volatile', 'kernel'],
- '__getitem__.return_value.revisions': ['rev1', 'rev2'],
- }
- self.vm.volumes.configure_mock(**volumes_conf)
- self.vm.storage = unittest.mock.Mock()
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.vm.volume.Revert',
- b'test-vm1', b'private', b'no-such-rev')
- self.assertEqual(self.vm.volumes.mock_calls,
- [unittest.mock.call.keys(),
- unittest.mock.call.__getattr__('__getitem__')('private')])
- self.assertFalse(self.vm.storage.called)
- def test_120_vm_volume_resize(self):
- self.vm.volumes = unittest.mock.MagicMock()
- volumes_conf = {
- 'keys.return_value': ['root', 'private', 'volatile', 'kernel'],
- }
- self.vm.volumes.configure_mock(**volumes_conf)
- self.vm.storage = unittest.mock.Mock()
- value = self.call_mgmt_func(b'admin.vm.volume.Resize',
- b'test-vm1', b'private', b'1024000000')
- self.assertIsNone(value)
- self.assertEqual(self.vm.volumes.mock_calls,
- [unittest.mock.call.keys()])
- self.assertEqual(self.vm.storage.mock_calls,
- [unittest.mock.call.resize('private', 1024000000)])
- def test_120_vm_volume_resize_invalid_size1(self):
- self.vm.volumes = unittest.mock.MagicMock()
- volumes_conf = {
- 'keys.return_value': ['root', 'private', 'volatile', 'kernel'],
- }
- self.vm.volumes.configure_mock(**volumes_conf)
- self.vm.storage = unittest.mock.Mock()
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.vm.volume.Resize',
- b'test-vm1', b'private', b'no-int-size')
- self.assertEqual(self.vm.volumes.mock_calls,
- [unittest.mock.call.keys()])
- self.assertFalse(self.vm.storage.called)
- def test_120_vm_volume_resize_invalid_size2(self):
- self.vm.volumes = unittest.mock.MagicMock()
- volumes_conf = {
- 'keys.return_value': ['root', 'private', 'volatile', 'kernel'],
- }
- self.vm.volumes.configure_mock(**volumes_conf)
- self.vm.storage = unittest.mock.Mock()
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.vm.volume.Resize',
- b'test-vm1', b'private', b'-1')
- self.assertEqual(self.vm.volumes.mock_calls,
- [unittest.mock.call.keys()])
- self.assertFalse(self.vm.storage.called)
- def test_130_pool_list(self):
- self.app.pools = ['file', 'lvm']
- value = self.call_mgmt_func(b'admin.pool.List', b'dom0')
- self.assertEqual(value, 'file\nlvm\n')
- self.assertFalse(self.app.save.called)
- @unittest.mock.patch('qubes.storage.pool_drivers')
- @unittest.mock.patch('qubes.storage.driver_parameters')
- def test_140_pool_listdrivers(self, mock_parameters, mock_drivers):
- self.app.pools = ['file', 'lvm']
- mock_drivers.return_value = ['driver1', 'driver2']
- mock_parameters.side_effect = \
- lambda driver: {
- 'driver1': ['param1', 'param2'],
- 'driver2': ['param3', 'param4']
- }[driver]
- value = self.call_mgmt_func(b'admin.pool.ListDrivers', b'dom0')
- self.assertEqual(value,
- 'driver1 param1 param2\ndriver2 param3 param4\n')
- self.assertEqual(mock_drivers.mock_calls, [unittest.mock.call()])
- self.assertEqual(mock_parameters.mock_calls,
- [unittest.mock.call('driver1'), unittest.mock.call('driver2')])
- self.assertFalse(self.app.save.called)
- def test_150_pool_info(self):
- self.app.pools = {
- 'pool1': unittest.mock.Mock(config={
- 'param1': 'value1', 'param2': 'value2'})
- }
- value = self.call_mgmt_func(b'admin.pool.Info', b'dom0', b'pool1')
- self.assertEqual(value, 'param1=value1\nparam2=value2\n')
- self.assertFalse(self.app.save.called)
- @unittest.mock.patch('qubes.storage.pool_drivers')
- @unittest.mock.patch('qubes.storage.driver_parameters')
- def test_160_pool_add(self, mock_parameters, mock_drivers):
- self.app.pools = {
- 'file': unittest.mock.Mock(),
- 'lvm': unittest.mock.Mock()
- }
- mock_drivers.return_value = ['driver1', 'driver2']
- mock_parameters.side_effect = \
- lambda driver: {
- 'driver1': ['param1', 'param2'],
- 'driver2': ['param3', 'param4']
- }[driver]
- self.app.add_pool = unittest.mock.Mock()
- value = self.call_mgmt_func(b'admin.pool.Add', b'dom0', b'driver1',
- b'name=test-pool\nparam1=some-value\n')
- self.assertIsNone(value)
- self.assertEqual(mock_drivers.mock_calls, [unittest.mock.call()])
- self.assertEqual(mock_parameters.mock_calls,
- [unittest.mock.call('driver1')])
- self.assertEqual(self.app.add_pool.mock_calls,
- [unittest.mock.call(name='test-pool', driver='driver1',
- param1='some-value')])
- self.assertTrue(self.app.save.called)
- @unittest.mock.patch('qubes.storage.pool_drivers')
- @unittest.mock.patch('qubes.storage.driver_parameters')
- def test_160_pool_add_invalid_driver(self, mock_parameters, mock_drivers):
- self.app.pools = {
- 'file': unittest.mock.Mock(),
- 'lvm': unittest.mock.Mock()
- }
- mock_drivers.return_value = ['driver1', 'driver2']
- mock_parameters.side_effect = \
- lambda driver: {
- 'driver1': ['param1', 'param2'],
- 'driver2': ['param3', 'param4']
- }[driver]
- self.app.add_pool = unittest.mock.Mock()
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.pool.Add', b'dom0',
- b'no-such-driver', b'name=test-pool\nparam1=some-value\n')
- self.assertEqual(mock_drivers.mock_calls, [unittest.mock.call()])
- self.assertEqual(mock_parameters.mock_calls, [])
- self.assertEqual(self.app.add_pool.mock_calls, [])
- self.assertFalse(self.app.save.called)
- @unittest.mock.patch('qubes.storage.pool_drivers')
- @unittest.mock.patch('qubes.storage.driver_parameters')
- def test_160_pool_add_invalid_param(self, mock_parameters, mock_drivers):
- self.app.pools = {
- 'file': unittest.mock.Mock(),
- 'lvm': unittest.mock.Mock()
- }
- mock_drivers.return_value = ['driver1', 'driver2']
- mock_parameters.side_effect = \
- lambda driver: {
- 'driver1': ['param1', 'param2'],
- 'driver2': ['param3', 'param4']
- }[driver]
- self.app.add_pool = unittest.mock.Mock()
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.pool.Add', b'dom0',
- b'driver1', b'name=test-pool\nparam3=some-value\n')
- self.assertEqual(mock_drivers.mock_calls, [unittest.mock.call()])
- self.assertEqual(mock_parameters.mock_calls,
- [unittest.mock.call('driver1')])
- self.assertEqual(self.app.add_pool.mock_calls, [])
- self.assertFalse(self.app.save.called)
- @unittest.mock.patch('qubes.storage.pool_drivers')
- @unittest.mock.patch('qubes.storage.driver_parameters')
- def test_160_pool_add_missing_name(self, mock_parameters, mock_drivers):
- self.app.pools = {
- 'file': unittest.mock.Mock(),
- 'lvm': unittest.mock.Mock()
- }
- mock_drivers.return_value = ['driver1', 'driver2']
- mock_parameters.side_effect = \
- lambda driver: {
- 'driver1': ['param1', 'param2'],
- 'driver2': ['param3', 'param4']
- }[driver]
- self.app.add_pool = unittest.mock.Mock()
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.pool.Add', b'dom0',
- b'driver1', b'param1=value\nparam2=some-value\n')
- self.assertEqual(mock_drivers.mock_calls, [unittest.mock.call()])
- self.assertEqual(mock_parameters.mock_calls, [])
- self.assertEqual(self.app.add_pool.mock_calls, [])
- self.assertFalse(self.app.save.called)
- @unittest.mock.patch('qubes.storage.pool_drivers')
- @unittest.mock.patch('qubes.storage.driver_parameters')
- def test_160_pool_add_existing_pool(self, mock_parameters, mock_drivers):
- self.app.pools = {
- 'file': unittest.mock.Mock(),
- 'lvm': unittest.mock.Mock()
- }
- mock_drivers.return_value = ['driver1', 'driver2']
- mock_parameters.side_effect = \
- lambda driver: {
- 'driver1': ['param1', 'param2'],
- 'driver2': ['param3', 'param4']
- }[driver]
- self.app.add_pool = unittest.mock.Mock()
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.pool.Add', b'dom0',
- b'driver1', b'name=file\nparam1=value\nparam2=some-value\n')
- self.assertEqual(mock_drivers.mock_calls, [unittest.mock.call()])
- self.assertEqual(mock_parameters.mock_calls, [])
- self.assertEqual(self.app.add_pool.mock_calls, [])
- self.assertFalse(self.app.save.called)
- @unittest.mock.patch('qubes.storage.pool_drivers')
- @unittest.mock.patch('qubes.storage.driver_parameters')
- def test_160_pool_add_invalid_config_format(self, mock_parameters,
- mock_drivers):
- self.app.pools = {
- 'file': unittest.mock.Mock(),
- 'lvm': unittest.mock.Mock()
- }
- mock_drivers.return_value = ['driver1', 'driver2']
- mock_parameters.side_effect = \
- lambda driver: {
- 'driver1': ['param1', 'param2'],
- 'driver2': ['param3', 'param4']
- }[driver]
- self.app.add_pool = unittest.mock.Mock()
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.pool.Add', b'dom0',
- b'driver1', b'name=test-pool\nparam 1=value\n_param2\n')
- self.assertEqual(mock_drivers.mock_calls, [unittest.mock.call()])
- self.assertEqual(mock_parameters.mock_calls, [])
- self.assertEqual(self.app.add_pool.mock_calls, [])
- self.assertFalse(self.app.save.called)
- def test_170_pool_remove(self):
- self.app.pools = {
- 'file': unittest.mock.Mock(),
- 'lvm': unittest.mock.Mock(),
- 'test-pool': unittest.mock.Mock(),
- }
- self.app.remove_pool = unittest.mock.Mock()
- value = self.call_mgmt_func(b'admin.pool.Remove', b'dom0', b'test-pool')
- self.assertIsNone(value)
- self.assertEqual(self.app.remove_pool.mock_calls,
- [unittest.mock.call('test-pool')])
- self.assertTrue(self.app.save.called)
- def test_170_pool_remove_invalid_pool(self):
- self.app.pools = {
- 'file': unittest.mock.Mock(),
- 'lvm': unittest.mock.Mock(),
- 'test-pool': unittest.mock.Mock(),
- }
- self.app.remove_pool = unittest.mock.Mock()
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.pool.Remove', b'dom0',
- b'no-such-pool')
- self.assertEqual(self.app.remove_pool.mock_calls, [])
- self.assertFalse(self.app.save.called)
- def test_180_label_list(self):
- value = self.call_mgmt_func(b'admin.label.List', b'dom0')
- self.assertEqual(value,
- ''.join('{}\n'.format(l.name) for l in self.app.labels.values()))
- self.assertFalse(self.app.save.called)
- def test_190_label_get(self):
- self.app.get_label = unittest.mock.Mock()
- self.app.get_label.configure_mock(**{'return_value.color': '0xff0000'})
- value = self.call_mgmt_func(b'admin.label.Get', b'dom0', b'red')
- self.assertEqual(value, '0xff0000')
- self.assertEqual(self.app.get_label.mock_calls,
- [unittest.mock.call('red')])
- self.assertFalse(self.app.save.called)
- def test_195_label_index(self):
- self.app.get_label = unittest.mock.Mock()
- self.app.get_label.configure_mock(**{'return_value.index': 1})
- value = self.call_mgmt_func(b'admin.label.Index', b'dom0', b'red')
- self.assertEqual(value, '1')
- self.assertEqual(self.app.get_label.mock_calls,
- [unittest.mock.call('red')])
- self.assertFalse(self.app.save.called)
- def test_200_label_create(self):
- self.app.get_label = unittest.mock.Mock()
- self.app.get_label.side_effect=KeyError
- self.app.labels = unittest.mock.MagicMock()
- labels_config = {
- 'keys.return_value': range(1, 9),
- }
- self.app.labels.configure_mock(**labels_config)
- value = self.call_mgmt_func(b'admin.label.Create', b'dom0', b'cyan',
- b'0x00ffff')
- self.assertIsNone(value)
- self.assertEqual(self.app.get_label.mock_calls,
- [unittest.mock.call('cyan')])
- self.assertEqual(self.app.labels.mock_calls,
- [unittest.mock.call.keys(),
- unittest.mock.call.__getattr__('__setitem__')(9,
- qubes.Label(9, '0x00ffff', 'cyan'))])
- self.assertTrue(self.app.save.called)
- def test_200_label_create_invalid_color(self):
- self.app.get_label = unittest.mock.Mock()
- self.app.get_label.side_effect=KeyError
- self.app.labels = unittest.mock.MagicMock()
- labels_config = {
- 'keys.return_value': range(1, 9),
- }
- self.app.labels.configure_mock(**labels_config)
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.label.Create', b'dom0', b'cyan',
- b'abcd')
- self.assertEqual(self.app.get_label.mock_calls,
- [unittest.mock.call('cyan')])
- self.assertEqual(self.app.labels.mock_calls, [])
- self.assertFalse(self.app.save.called)
- def test_200_label_create_invalid_name(self):
- self.app.get_label = unittest.mock.Mock()
- self.app.get_label.side_effect=KeyError
- self.app.labels = unittest.mock.MagicMock()
- labels_config = {
- 'keys.return_value': range(1, 9),
- }
- self.app.labels.configure_mock(**labels_config)
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.label.Create', b'dom0', b'01',
- b'0xff0000')
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.label.Create', b'dom0', b'../xxx',
- b'0xff0000')
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.label.Create', b'dom0',
- b'strange-name!@#$',
- b'0xff0000')
- self.assertEqual(self.app.get_label.mock_calls, [])
- self.assertEqual(self.app.labels.mock_calls, [])
- self.assertFalse(self.app.save.called)
- def test_200_label_create_already_exists(self):
- self.app.get_label = unittest.mock.Mock(wraps=self.app.get_label)
- with self.assertRaises(qubes.exc.QubesValueError):
- self.call_mgmt_func(b'admin.label.Create', b'dom0', b'red',
- b'abcd')
- self.assertEqual(self.app.get_label.mock_calls,
- [unittest.mock.call('red')])
- self.assertFalse(self.app.save.called)
- def test_210_label_remove(self):
- label = qubes.Label(9, '0x00ffff', 'cyan')
- self.app.labels[9] = label
- self.app.get_label = unittest.mock.Mock(wraps=self.app.get_label,
- **{'return_value.index': 9})
- self.app.labels = unittest.mock.MagicMock(wraps=self.app.labels)
- value = self.call_mgmt_func(b'admin.label.Remove', b'dom0', b'cyan')
- self.assertIsNone(value)
- self.assertEqual(self.app.get_label.mock_calls,
- [unittest.mock.call('cyan')])
- self.assertEqual(self.app.labels.mock_calls,
- [unittest.mock.call.__delitem__(9)])
- self.assertTrue(self.app.save.called)
- def test_210_label_remove_invalid_label(self):
- with self.assertRaises(qubes.exc.QubesValueError):
- self.call_mgmt_func(b'admin.label.Remove', b'dom0',
- b'no-such-label')
- self.assertFalse(self.app.save.called)
- def test_210_label_remove_default_label(self):
- self.app.labels = unittest.mock.MagicMock(wraps=self.app.labels)
- self.app.get_label = unittest.mock.Mock(wraps=self.app.get_label,
- **{'return_value.index': 6})
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.label.Remove', b'dom0',
- b'blue')
- self.assertEqual(self.app.labels.mock_calls, [])
- self.assertFalse(self.app.save.called)
- def test_210_label_remove_in_use(self):
- self.app.labels = unittest.mock.MagicMock(wraps=self.app.labels)
- self.app.get_label = unittest.mock.Mock(wraps=self.app.get_label,
- **{'return_value.index': 1})
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.label.Remove', b'dom0',
- b'red')
- self.assertEqual(self.app.labels.mock_calls, [])
- self.assertFalse(self.app.save.called)
- def test_220_start(self):
- func_mock = unittest.mock.Mock()
- @asyncio.coroutine
- def coroutine_mock(*args, **kwargs):
- return func_mock(*args, **kwargs)
- self.vm.start = coroutine_mock
- value = self.call_mgmt_func(b'admin.vm.Start', b'test-vm1')
- self.assertIsNone(value)
- func_mock.assert_called_once_with()
- def test_230_shutdown(self):
- func_mock = unittest.mock.Mock()
- @asyncio.coroutine
- def coroutine_mock(*args, **kwargs):
- return func_mock(*args, **kwargs)
- self.vm.shutdown = coroutine_mock
- value = self.call_mgmt_func(b'admin.vm.Shutdown', b'test-vm1')
- self.assertIsNone(value)
- func_mock.assert_called_once_with()
- def test_240_pause(self):
- func_mock = unittest.mock.Mock()
- @asyncio.coroutine
- def coroutine_mock(*args, **kwargs):
- return func_mock(*args, **kwargs)
- self.vm.pause = coroutine_mock
- value = self.call_mgmt_func(b'admin.vm.Pause', b'test-vm1')
- self.assertIsNone(value)
- func_mock.assert_called_once_with()
- def test_250_unpause(self):
- func_mock = unittest.mock.Mock()
- @asyncio.coroutine
- def coroutine_mock(*args, **kwargs):
- return func_mock(*args, **kwargs)
- self.vm.unpause = coroutine_mock
- value = self.call_mgmt_func(b'admin.vm.Unpause', b'test-vm1')
- self.assertIsNone(value)
- func_mock.assert_called_once_with()
- def test_260_kill(self):
- func_mock = unittest.mock.Mock()
- @asyncio.coroutine
- def coroutine_mock(*args, **kwargs):
- return func_mock(*args, **kwargs)
- self.vm.kill = coroutine_mock
- value = self.call_mgmt_func(b'admin.vm.Kill', b'test-vm1')
- self.assertIsNone(value)
- func_mock.assert_called_once_with()
- def test_270_events(self):
- send_event = unittest.mock.Mock(spec=[])
- mgmt_obj = qubes.api.admin.QubesAdminAPI(self.app, b'dom0', b'admin.Events',
- b'dom0', b'', send_event=send_event)
- @asyncio.coroutine
- def fire_event():
- self.vm.fire_event('test-event', arg1='abc')
- mgmt_obj.cancel()
- loop = asyncio.get_event_loop()
- execute_task = asyncio.ensure_future(
- mgmt_obj.execute(untrusted_payload=b''))
- asyncio.ensure_future(fire_event())
- loop.run_until_complete(execute_task)
- self.assertIsNone(execute_task.result())
- self.assertEventFired(self.emitter,
- 'mgmt-permission:' + 'admin.Events')
- self.assertEqual(send_event.mock_calls,
- [
- unittest.mock.call(self.app, 'connection-established'),
- unittest.mock.call(self.vm, 'test-event', arg1='abc')
- ])
- def test_280_feature_list(self):
- self.vm.features['test-feature'] = 'some-value'
- value = self.call_mgmt_func(b'admin.vm.feature.List', b'test-vm1')
- self.assertEqual(value, 'test-feature\n')
- self.assertFalse(self.app.save.called)
- def test_290_feature_get(self):
- self.vm.features['test-feature'] = 'some-value'
- value = self.call_mgmt_func(b'admin.vm.feature.Get', b'test-vm1',
- b'test-feature')
- self.assertEqual(value, 'some-value')
- self.assertFalse(self.app.save.called)
- def test_291_feature_get_none(self):
- with self.assertRaises(qubes.exc.QubesFeatureNotFoundError):
- self.call_mgmt_func(b'admin.vm.feature.Get',
- b'test-vm1', b'test-feature')
- self.assertFalse(self.app.save.called)
- def test_300_feature_remove(self):
- self.vm.features['test-feature'] = 'some-value'
- value = self.call_mgmt_func(b'admin.vm.feature.Remove', b'test-vm1',
- b'test-feature')
- self.assertIsNone(value, None)
- self.assertNotIn('test-feature', self.vm.features)
- self.assertTrue(self.app.save.called)
- def test_301_feature_remove_none(self):
- with self.assertRaises(qubes.exc.QubesFeatureNotFoundError):
- self.call_mgmt_func(b'admin.vm.feature.Remove',
- b'test-vm1', b'test-feature')
- self.assertFalse(self.app.save.called)
- def test_310_feature_checkwithtemplate(self):
- self.vm.features['test-feature'] = 'some-value'
- value = self.call_mgmt_func(b'admin.vm.feature.CheckWithTemplate',
- b'test-vm1', b'test-feature')
- self.assertEqual(value, 'some-value')
- self.assertFalse(self.app.save.called)
- def test_311_feature_checkwithtemplate_tpl(self):
- self.template.features['test-feature'] = 'some-value'
- value = self.call_mgmt_func(b'admin.vm.feature.CheckWithTemplate',
- b'test-vm1', b'test-feature')
- self.assertEqual(value, 'some-value')
- self.assertFalse(self.app.save.called)
- def test_312_feature_checkwithtemplate_none(self):
- with self.assertRaises(qubes.exc.QubesFeatureNotFoundError):
- self.call_mgmt_func(b'admin.vm.feature.CheckWithTemplate',
- b'test-vm1', b'test-feature')
- self.assertFalse(self.app.save.called)
- def test_320_feature_set(self):
- value = self.call_mgmt_func(b'admin.vm.feature.Set',
- b'test-vm1', b'test-feature', b'some-value')
- self.assertIsNone(value)
- self.assertEqual(self.vm.features['test-feature'], 'some-value')
- self.assertTrue(self.app.save.called)
- def test_321_feature_set_empty(self):
- value = self.call_mgmt_func(b'admin.vm.feature.Set',
- b'test-vm1', b'test-feature', b'')
- self.assertIsNone(value)
- self.assertEqual(self.vm.features['test-feature'], '')
- self.assertTrue(self.app.save.called)
- def test_320_feature_set_invalid(self):
- with self.assertRaises(UnicodeDecodeError):
- self.call_mgmt_func(b'admin.vm.feature.Set',
- b'test-vm1', b'test-feature', b'\x02\x03\xffsome-value')
- self.assertNotIn('test-feature', self.vm.features)
- self.assertFalse(self.app.save.called)
- @asyncio.coroutine
- def dummy_coro(self, *args, **kwargs):
- pass
- @unittest.mock.patch('qubes.storage.Storage.create')
- def test_330_vm_create_standalone(self, storage_mock):
- storage_mock.side_effect = self.dummy_coro
- self.call_mgmt_func(b'admin.vm.Create.StandaloneVM',
- b'dom0', b'', b'name=test-vm2 label=red')
- self.assertIn('test-vm2', self.app.domains)
- vm = self.app.domains['test-vm2']
- self.assertIsInstance(vm, qubes.vm.standalonevm.StandaloneVM)
- self.assertEqual(vm.label, self.app.get_label('red'))
- self.assertEqual(storage_mock.mock_calls,
- [unittest.mock.call(self.app.domains['test-vm2']).create()])
- self.assertTrue(os.path.exists(os.path.join(
- self.test_base_dir, 'appvms', 'test-vm2')))
- self.assertTrue(self.app.save.called)
- @unittest.mock.patch('qubes.storage.Storage.create')
- def test_331_vm_create_standalone_spurious_template(self, storage_mock):
- storage_mock.side_effect = self.dummy_coro
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.vm.Create.StandaloneVM',
- b'dom0', b'test-template', b'name=test-vm2 label=red')
- self.assertNotIn('test-vm2', self.app.domains)
- self.assertEqual(storage_mock.mock_calls, [])
- self.assertFalse(os.path.exists(os.path.join(
- self.test_base_dir, 'appvms', 'test-vm2')))
- self.assertNotIn('test-vm2', self.app.domains)
- self.assertFalse(self.app.save.called)
- @unittest.mock.patch('qubes.storage.Storage.create')
- def test_332_vm_create_app(self, storage_mock):
- storage_mock.side_effect = self.dummy_coro
- self.call_mgmt_func(b'admin.vm.Create.AppVM',
- b'dom0', b'test-template', b'name=test-vm2 label=red')
- self.assertIn('test-vm2', self.app.domains)
- vm = self.app.domains['test-vm2']
- self.assertEqual(vm.label, self.app.get_label('red'))
- self.assertEqual(vm.template, self.app.domains['test-template'])
- self.assertEqual(storage_mock.mock_calls,
- [unittest.mock.call(self.app.domains['test-vm2']).create()])
- self.assertTrue(os.path.exists(os.path.join(
- self.test_base_dir, 'appvms', 'test-vm2')))
- self.assertTrue(self.app.save.called)
- @unittest.mock.patch('qubes.storage.Storage.create')
- def test_333_vm_create_app_missing_template(self, storage_mock):
- storage_mock.side_effect = self.dummy_coro
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.vm.Create.AppVM',
- b'dom0', b'', b'name=test-vm2 label=red')
- self.assertNotIn('test-vm2', self.app.domains)
- self.assertEqual(storage_mock.mock_calls, [])
- self.assertFalse(os.path.exists(os.path.join(
- self.test_base_dir, 'appvms', 'test-vm2')))
- self.assertNotIn('test-vm2', self.app.domains)
- self.assertFalse(self.app.save.called)
- @unittest.mock.patch('qubes.storage.Storage.create')
- def test_334_vm_create_invalid_name(self, storage_mock):
- storage_mock.side_effect = self.dummy_coro
- with self.assertRaises(qubes.exc.QubesValueError):
- self.call_mgmt_func(b'admin.vm.Create.AppVM',
- b'dom0', b'test-template', b'name=test-###')
- self.assertNotIn('test-###', self.app.domains)
- self.assertFalse(self.app.save.called)
- @unittest.mock.patch('qubes.storage.Storage.create')
- def test_335_vm_create_missing_name(self, storage_mock):
- storage_mock.side_effect = self.dummy_coro
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.vm.Create.AppVM',
- b'dom0', b'test-template', b'label=red')
- self.assertFalse(self.app.save.called)
- @unittest.mock.patch('qubes.storage.Storage.create')
- def test_336_vm_create_spurious_pool(self, storage_mock):
- storage_mock.side_effect = self.dummy_coro
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.vm.Create.AppVM',
- b'dom0', b'test-template',
- b'name=test-vm2 label=red pool=default')
- self.assertNotIn('test-vm2', self.app.domains)
- self.assertFalse(self.app.save.called)
- @unittest.mock.patch('qubes.storage.Storage.create')
- def test_337_vm_create_duplicate_name(self, storage_mock):
- storage_mock.side_effect = self.dummy_coro
- with self.assertRaises(qubes.exc.QubesException):
- self.call_mgmt_func(b'admin.vm.Create.AppVM',
- b'dom0', b'test-template',
- b'name=test-vm1 label=red')
- self.assertFalse(self.app.save.called)
- @unittest.mock.patch('qubes.storage.Storage.create')
- def test_338_vm_create_name_twice(self, storage_mock):
- storage_mock.side_effect = self.dummy_coro
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.vm.Create.AppVM',
- b'dom0', b'test-template',
- b'name=test-vm2 name=test-vm3 label=red')
- self.assertNotIn('test-vm2', self.app.domains)
- self.assertNotIn('test-vm3', self.app.domains)
- self.assertFalse(self.app.save.called)
- @unittest.mock.patch('qubes.storage.Storage.create')
- def test_340_vm_create_in_pool_app(self, storage_mock):
- storage_mock.side_effect = self.dummy_coro
- self.call_mgmt_func(b'admin.vm.CreateInPool.AppVM',
- b'dom0', b'test-template', b'name=test-vm2 label=red '
- b'pool=test')
- self.assertIn('test-vm2', self.app.domains)
- vm = self.app.domains['test-vm2']
- self.assertEqual(vm.label, self.app.get_label('red'))
- self.assertEqual(vm.template, self.app.domains['test-template'])
- # setting pool= affect only volumes actually created for this VM,
- # not used from a template or so
- self.assertEqual(vm.volume_config['root']['pool'], 'default')
- self.assertEqual(vm.volume_config['private']['pool'], 'test')
- self.assertEqual(vm.volume_config['volatile']['pool'], 'test')
- self.assertEqual(vm.volume_config['kernel']['pool'], 'linux-kernel')
- self.assertEqual(storage_mock.mock_calls,
- [unittest.mock.call(self.app.domains['test-vm2']).create()])
- self.assertTrue(os.path.exists(os.path.join(
- self.test_base_dir, 'appvms', 'test-vm2')))
- self.assertTrue(self.app.save.called)
- @unittest.mock.patch('qubes.storage.Storage.create')
- def test_341_vm_create_in_pool_private(self, storage_mock):
- storage_mock.side_effect = self.dummy_coro
- self.call_mgmt_func(b'admin.vm.CreateInPool.AppVM',
- b'dom0', b'test-template', b'name=test-vm2 label=red '
- b'pool:private=test')
- self.assertIn('test-vm2', self.app.domains)
- vm = self.app.domains['test-vm2']
- self.assertEqual(vm.label, self.app.get_label('red'))
- self.assertEqual(vm.template, self.app.domains['test-template'])
- self.assertEqual(vm.volume_config['root']['pool'], 'default')
- self.assertEqual(vm.volume_config['private']['pool'], 'test')
- self.assertEqual(vm.volume_config['volatile']['pool'], 'default')
- self.assertEqual(vm.volume_config['kernel']['pool'], 'linux-kernel')
- self.assertEqual(storage_mock.mock_calls,
- [unittest.mock.call(self.app.domains['test-vm2']).create()])
- self.assertTrue(os.path.exists(os.path.join(
- self.test_base_dir, 'appvms', 'test-vm2')))
- self.assertTrue(self.app.save.called)
- @unittest.mock.patch('qubes.storage.Storage.create')
- def test_342_vm_create_in_pool_invalid_pool(self, storage_mock):
- storage_mock.side_effect = self.dummy_coro
- with self.assertRaises(qubes.exc.QubesException):
- self.call_mgmt_func(b'admin.vm.CreateInPool.AppVM',
- b'dom0', b'test-template', b'name=test-vm2 label=red '
- b'pool=no-such-pool')
- self.assertFalse(self.app.save.called)
- @unittest.mock.patch('qubes.storage.Storage.create')
- def test_343_vm_create_in_pool_invalid_pool2(self, storage_mock):
- storage_mock.side_effect = self.dummy_coro
- with self.assertRaises(qubes.exc.QubesException):
- self.call_mgmt_func(b'admin.vm.CreateInPool.AppVM',
- b'dom0', b'test-template', b'name=test-vm2 label=red '
- b'pool:private=no-such-pool')
- self.assertNotIn('test-vm2', self.app.domains)
- self.assertFalse(self.app.save.called)
- @unittest.mock.patch('qubes.storage.Storage.create')
- def test_344_vm_create_in_pool_invalid_volume(self, storage_mock):
- storage_mock.side_effect = self.dummy_coro
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.vm.CreateInPool.AppVM',
- b'dom0', b'test-template', b'name=test-vm2 label=red '
- b'pool:invalid=test')
- self.assertNotIn('test-vm2', self.app.domains)
- self.assertFalse(self.app.save.called)
- @unittest.mock.patch('qubes.storage.Storage.create')
- def test_345_vm_create_in_pool_app_root(self, storage_mock):
- # setting custom pool for 'root' volume of AppVM should not be
- # allowed - this volume belongs to the template
- storage_mock.side_effect = self.dummy_coro
- with self.assertRaises(qubes.exc.QubesException):
- self.call_mgmt_func(b'admin.vm.CreateInPool.AppVM',
- b'dom0', b'test-template', b'name=test-vm2 label=red '
- b'pool:root=test')
- self.assertNotIn('test-vm2', self.app.domains)
- self.assertFalse(self.app.save.called)
- @unittest.mock.patch('qubes.storage.Storage.create')
- def test_346_vm_create_in_pool_duplicate_pool(self, storage_mock):
- # setting custom pool for 'root' volume of AppVM should not be
- # allowed - this volume belongs to the template
- storage_mock.side_effect = self.dummy_coro
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(b'admin.vm.CreateInPool.AppVM',
- b'dom0', b'test-template', b'name=test-vm2 label=red '
- b'pool=test pool:root=test')
- self.assertNotIn('test-vm2', self.app.domains)
- self.assertFalse(self.app.save.called)
- @unittest.mock.patch('qubes.storage.Storage.clone')
- @unittest.mock.patch('qubes.storage.Storage.verify')
- def test_350_vm_clone(self, mock_verify, mock_clone):
- mock_clone.side_effect = self.dummy_coro
- mock_verify.side_effect = self.dummy_coro
- self.call_mgmt_func(b'admin.vm.Clone',
- b'test-vm1', b'', b'name=test-vm2')
- self.assertIn('test-vm2', self.app.domains)
- vm = self.app.domains['test-vm2']
- self.assertEqual(vm.label, self.app.get_label('red'))
- self.assertEqual(vm.template, self.app.domains['test-template'])
- self.assertEqual(mock_clone.mock_calls,
- [unittest.mock.call(self.app.domains['test-vm2']).clone(
- self.app.domains['test-vm1'])])
- self.assertTrue(os.path.exists(os.path.join(
- self.test_base_dir, 'appvms', 'test-vm2')))
- self.assertTrue(self.app.save.called)
- @unittest.mock.patch('qubes.storage.Storage.clone')
- @unittest.mock.patch('qubes.storage.Storage.verify')
- def test_351_vm_clone_extra_params(self, mock_verify, mock_clone):
- mock_clone.side_effect = self.dummy_coro
- mock_verify.side_effect = self.dummy_coro
- with self.assertRaises(qubes.exc.QubesException):
- self.call_mgmt_func(b'admin.vm.Clone',
- b'test-vm1', b'', b'name=test-vm2 label=red')
- self.assertNotIn('test-vm2', self.app.domains)
- self.assertEqual(mock_clone.mock_calls, [])
- self.assertFalse(os.path.exists(os.path.join(
- self.test_base_dir, 'appvms', 'test-vm2')))
- self.assertFalse(self.app.save.called)
- @unittest.mock.patch('qubes.storage.Storage.clone')
- @unittest.mock.patch('qubes.storage.Storage.verify')
- def test_352_vm_clone_duplicate_name(self, mock_verify, mock_clone):
- mock_clone.side_effect = self.dummy_coro
- mock_verify.side_effect = self.dummy_coro
- with self.assertRaises(qubes.exc.QubesException):
- self.call_mgmt_func(b'admin.vm.Clone',
- b'test-vm1', b'', b'name=test-vm1')
- self.assertFalse(self.app.save.called)
- @unittest.mock.patch('qubes.storage.Storage.clone')
- @unittest.mock.patch('qubes.storage.Storage.verify')
- def test_353_vm_clone_invalid_name(self, mock_verify, mock_clone):
- mock_clone.side_effect = self.dummy_coro
- mock_verify.side_effect = self.dummy_coro
- with self.assertRaises(qubes.exc.QubesException):
- self.call_mgmt_func(b'admin.vm.Clone',
- b'test-vm1', b'', b'name=test-vm2/..')
- self.assertNotIn('test-vm2/..', self.app.domains)
- self.assertEqual(mock_clone.mock_calls, [])
- self.assertFalse(os.path.exists(os.path.join(
- self.test_base_dir, 'appvms', 'test-vm2/..')))
- self.assertFalse(self.app.save.called)
- def test_990_vm_unexpected_payload(self):
- methods_with_no_payload = [
- b'admin.vm.List',
- b'admin.vm.Remove',
- b'admin.vm.property.List',
- b'admin.vm.property.Get',
- b'admin.vm.property.Help',
- b'admin.vm.property.HelpRst',
- b'admin.vm.property.Reset',
- b'admin.vm.feature.List',
- b'admin.vm.feature.Get',
- b'admin.vm.feature.CheckWithTemplate',
- b'admin.vm.feature.Remove',
- b'admin.vm.tag.List',
- b'admin.vm.tag.Get',
- b'admin.vm.tag.Remove',
- b'admin.vm.tag.Set',
- b'admin.vm.firewall.Get',
- b'admin.vm.firewall.RemoveRule',
- b'admin.vm.firewall.Flush',
- b'admin.vm.device.pci.Attach',
- b'admin.vm.device.pci.Detach',
- b'admin.vm.device.pci.List',
- b'admin.vm.device.pci.Available',
- b'admin.vm.microphone.Attach',
- b'admin.vm.microphone.Detach',
- b'admin.vm.microphone.Status',
- b'admin.vm.volume.ListSnapshots',
- b'admin.vm.volume.List',
- b'admin.vm.volume.Info',
- b'admin.vm.Start',
- b'admin.vm.Shutdown',
- b'admin.vm.Pause',
- b'admin.vm.Unpause',
- b'admin.vm.Kill',
- b'admin.Events',
- b'admin.vm.feature.List',
- b'admin.vm.feature.Get',
- b'admin.vm.feature.Remove',
- b'admin.vm.feature.CheckWithTemplate',
- ]
- # make sure also no methods on actual VM gets called
- vm_mock = unittest.mock.MagicMock()
- vm_mock.name = self.vm.name
- vm_mock.qid = self.vm.qid
- vm_mock.__lt__ = (lambda x, y: x.qid < y.qid)
- self.app.domains._dict[self.vm.qid] = vm_mock
- for method in methods_with_no_payload:
- # should reject payload regardless of having argument or not
- with self.subTest(method.decode('ascii')):
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(method, b'test-vm1', b'',
- b'unexpected-payload')
- self.assertFalse(vm_mock.called)
- self.assertFalse(self.app.save.called)
- with self.subTest(method.decode('ascii') + '+arg'):
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(method, b'test-vm1', b'some-arg',
- b'unexpected-payload')
- self.assertFalse(vm_mock.called)
- self.assertFalse(self.app.save.called)
- def test_991_vm_unexpected_argument(self):
- methods_with_no_argument = [
- b'admin.vm.List',
- b'admin.vm.Clone',
- b'admin.vm.Remove',
- b'admin.vm.property.List',
- b'admin.vm.feature.List',
- b'admin.vm.tag.List',
- b'admin.vm.firewall.List',
- b'admin.vm.firewall.Flush',
- b'admin.vm.device.pci.List',
- b'admin.vm.device.pci.Available',
- b'admin.vm.microphone.Attach',
- b'admin.vm.microphone.Detach',
- b'admin.vm.microphone.Status',
- b'admin.vm.volume.List',
- b'admin.vm.Start',
- b'admin.vm.Shutdown',
- b'admin.vm.Pause',
- b'admin.vm.Unpause',
- b'admin.vm.Kill',
- b'admin.Events',
- b'admin.vm.feature.List',
- ]
- # make sure also no methods on actual VM gets called
- vm_mock = unittest.mock.MagicMock()
- vm_mock.name = self.vm.name
- vm_mock.qid = self.vm.qid
- vm_mock.__lt__ = (lambda x, y: x.qid < y.qid)
- self.app.domains._dict[self.vm.qid] = vm_mock
- for method in methods_with_no_argument:
- # should reject argument regardless of having payload or not
- with self.subTest(method.decode('ascii')):
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(method, b'test-vm1', b'some-arg',
- b'')
- self.assertFalse(vm_mock.called)
- self.assertFalse(self.app.save.called)
- with self.subTest(method.decode('ascii') + '+payload'):
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(method, b'test-vm1', b'unexpected-arg',
- b'some-payload')
- self.assertFalse(vm_mock.called)
- self.assertFalse(self.app.save.called)
- def test_992_dom0_unexpected_payload(self):
- methods_with_no_payload = [
- b'admin.vmclass.List',
- b'admin.vm.List',
- b'admin.label.List',
- b'admin.label.Get',
- b'admin.label.Remove',
- b'admin.property.List',
- b'admin.property.Get',
- b'admin.property.Help',
- b'admin.property.HelpRst',
- b'admin.property.Reset',
- b'admin.pool.List',
- b'admin.pool.ListDrivers',
- b'admin.pool.Info',
- b'admin.pool.Remove',
- b'admin.backup.Execute',
- b'admin.Events',
- ]
- # make sure also no methods on actual VM gets called
- vm_mock = unittest.mock.MagicMock()
- vm_mock.name = self.vm.name
- vm_mock.qid = self.vm.qid
- vm_mock.__lt__ = (lambda x, y: x.qid < y.qid)
- self.app.domains._dict[self.vm.qid] = vm_mock
- for method in methods_with_no_payload:
- # should reject payload regardless of having argument or not
- with self.subTest(method.decode('ascii')):
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(method, b'dom0', b'',
- b'unexpected-payload')
- self.assertFalse(vm_mock.called)
- self.assertFalse(self.app.save.called)
- with self.subTest(method.decode('ascii') + '+arg'):
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(method, b'dom0', b'some-arg',
- b'unexpected-payload')
- self.assertFalse(vm_mock.called)
- self.assertFalse(self.app.save.called)
- def test_993_dom0_unexpected_argument(self):
- methods_with_no_argument = [
- b'admin.vmclass.List',
- b'admin.vm.List',
- b'admin.label.List',
- b'admin.property.List',
- b'admin.pool.List',
- b'admin.pool.ListDrivers',
- b'admin.Events',
- ]
- # make sure also no methods on actual VM gets called
- vm_mock = unittest.mock.MagicMock()
- vm_mock.name = self.vm.name
- vm_mock.qid = self.vm.qid
- vm_mock.__lt__ = (lambda x, y: x.qid < y.qid)
- self.app.domains._dict[self.vm.qid] = vm_mock
- for method in methods_with_no_argument:
- # should reject argument regardless of having payload or not
- with self.subTest(method.decode('ascii')):
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(method, b'dom0', b'some-arg',
- b'')
- self.assertFalse(vm_mock.called)
- self.assertFalse(self.app.save.called)
- with self.subTest(method.decode('ascii') + '+payload'):
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(method, b'dom0', b'unexpected-arg',
- b'some-payload')
- self.assertFalse(vm_mock.called)
- self.assertFalse(self.app.save.called)
- def test_994_dom0_only_calls(self):
- # TODO set some better arguments, to make sure the call was rejected
- # because of invalid destination, not invalid arguments
- methods_for_dom0_only = [
- b'admin.vmclass.List',
- b'admin.vm.Create.AppVM',
- b'admin.vm.CreateInPool.AppVM',
- b'admin.vm.CreateTemplate',
- b'admin.label.List',
- b'admin.label.Create',
- b'admin.label.Get',
- b'admin.label.Remove',
- b'admin.property.List',
- b'admin.property.Get',
- b'admin.property.Set',
- b'admin.property.Help',
- b'admin.property.HelpRst',
- b'admin.property.Reset',
- b'admin.pool.List',
- b'admin.pool.ListDrivers',
- b'admin.pool.Info',
- b'admin.pool.Add',
- b'admin.pool.Remove',
- b'admin.pool.volume.List',
- b'admin.pool.volume.Info',
- b'admin.pool.volume.ListSnapshots',
- b'admin.pool.volume.Snapshot',
- b'admin.pool.volume.Revert',
- b'admin.pool.volume.Resize',
- b'admin.backup.Execute',
- b'admin.backup.Info',
- b'admin.backup.Restore',
- ]
- # make sure also no methods on actual VM gets called
- vm_mock = unittest.mock.MagicMock()
- vm_mock.name = self.vm.name
- vm_mock.qid = self.vm.qid
- vm_mock.__lt__ = (lambda x, y: x.qid < y.qid)
- self.app.domains._dict[self.vm.qid] = vm_mock
- for method in methods_for_dom0_only:
- # should reject call regardless of having payload or not
- with self.subTest(method.decode('ascii')):
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(method, b'test-vm1', b'',
- b'')
- self.assertFalse(vm_mock.called)
- self.assertFalse(self.app.save.called)
- with self.subTest(method.decode('ascii') + '+arg'):
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(method, b'test-vm1', b'some-arg',
- b'')
- self.assertFalse(vm_mock.called)
- self.assertFalse(self.app.save.called)
- with self.subTest(method.decode('ascii') + '+payload'):
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(method, b'test-vm1', b'',
- b'payload')
- self.assertFalse(vm_mock.called)
- self.assertFalse(self.app.save.called)
- with self.subTest(method.decode('ascii') + '+arg+payload'):
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(method, b'test-vm1', b'some-arg',
- b'some-payload')
- self.assertFalse(vm_mock.called)
- self.assertFalse(self.app.save.called)
- @unittest.skip('undecided')
- def test_995_vm_only_calls(self):
- # XXX is it really a good idea to prevent those calls this early?
- # TODO set some better arguments, to make sure the call was rejected
- # because of invalid destination, not invalid arguments
- methods_for_vm_only = [
- b'admin.vm.Clone',
- b'admin.vm.Remove',
- b'admin.vm.property.List',
- b'admin.vm.property.Get',
- b'admin.vm.property.Set',
- b'admin.vm.property.Help',
- b'admin.vm.property.HelpRst',
- b'admin.vm.property.Reset',
- b'admin.vm.feature.List',
- b'admin.vm.feature.Get',
- b'admin.vm.feature.Set',
- b'admin.vm.feature.CheckWithTemplate',
- b'admin.vm.feature.Remove',
- b'admin.vm.tag.List',
- b'admin.vm.tag.Get',
- b'admin.vm.tag.Remove',
- b'admin.vm.tag.Set',
- b'admin.vm.firewall.Get',
- b'admin.vm.firewall.RemoveRule',
- b'admin.vm.firewall.InsertRule',
- b'admin.vm.firewall.Flush',
- b'admin.vm.device.pci.Attach',
- b'admin.vm.device.pci.Detach',
- b'admin.vm.device.pci.List',
- b'admin.vm.device.pci.Available',
- b'admin.vm.microphone.Attach',
- b'admin.vm.microphone.Detach',
- b'admin.vm.microphone.Status',
- b'admin.vm.volume.ListSnapshots',
- b'admin.vm.volume.List',
- b'admin.vm.volume.Info',
- b'admin.vm.volume.Revert',
- b'admin.vm.volume.Resize',
- b'admin.vm.Start',
- b'admin.vm.Shutdown',
- b'admin.vm.Pause',
- b'admin.vm.Unpause',
- b'admin.vm.Kill',
- b'admin.vm.feature.List',
- b'admin.vm.feature.Get',
- b'admin.vm.feature.Set',
- b'admin.vm.feature.Remove',
- b'admin.vm.feature.CheckWithTemplate',
- ]
- # make sure also no methods on actual VM gets called
- vm_mock = unittest.mock.MagicMock()
- vm_mock.name = self.vm.name
- vm_mock.qid = self.vm.qid
- vm_mock.__lt__ = (lambda x, y: x.qid < y.qid)
- self.app.domains._dict[self.vm.qid] = vm_mock
- for method in methods_for_vm_only:
- # should reject payload regardless of having argument or not
- # should reject call regardless of having payload or not
- with self.subTest(method.decode('ascii')):
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(method, b'dom0', b'',
- b'')
- self.assertFalse(vm_mock.called)
- self.assertFalse(self.app.save.called)
- with self.subTest(method.decode('ascii') + '+arg'):
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(method, b'dom0', b'some-arg',
- b'')
- self.assertFalse(vm_mock.called)
- self.assertFalse(self.app.save.called)
- with self.subTest(method.decode('ascii') + '+payload'):
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(method, b'dom0', b'',
- b'payload')
- self.assertFalse(vm_mock.called)
- self.assertFalse(self.app.save.called)
- with self.subTest(method.decode('ascii') + '+arg+payload'):
- with self.assertRaises(AssertionError):
- self.call_mgmt_func(method, b'dom0', b'some-arg',
- b'some-payload')
- self.assertFalse(vm_mock.called)
- self.assertFalse(self.app.save.called)
|