Qubes/core-admin
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

tests: improve cleanup after LVM tests

Browse Source 
Remove test volumes - this way if a test fails, subsequent tests have a
chance to succeed.
This commit is contained in:
Marek Marczykowski-Górecki 2019-07-28 21:41:04 +02:00
parent d94e13f753
commit 8414d0153f
No known key found for this signature in database
GPG Key ID: 063938BA42CFA724
1 changed files with 32 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
qubes/tests/storage_lvm.py
View File

@ -79,8 +79,24 @@ class ThinPoolBase(qubes.tests.QubesTestCase):
self.app.add_pool(**POOL_CONF)) self.app.add_pool(**POOL_CONF))
self.created_pool = True self.created_pool = True
def cleanup_test_volumes(self):
p = self.loop.run_until_complete(asyncio.create_subprocess_exec(
'sudo', 'lvs', '--noheadings', '-o', 'lv_name', self.pool.volume_group,
stdout=subprocess.PIPE
))
volumes, _ = self.loop.run_until_complete(p.communicate())
for volume in volumes.decode().splitlines():
volume = volume.strip()
if not volume.startswith('vm-' + qubes.tests.VMPREFIX):
continue
p = self.loop.run_until_complete(asyncio.create_subprocess_exec(
'sudo', 'lvremove', '-f', '/'.join([self.pool.volume_group, volume])
))
self.loop.run_until_complete(p.wait())
def tearDown(self): def tearDown(self):
''' Remove the default lvm pool if it was created only for this test ''' ''' Remove the default lvm pool if it was created only for this test '''
self.cleanup_test_volumes()
if self.created_pool: if self.created_pool:
self.loop.run_until_complete(self.app.remove_pool(self.pool.name)) self.loop.run_until_complete(self.app.remove_pool(self.pool.name))
super(ThinPoolBase, self).tearDown() super(ThinPoolBase, self).tearDown()
@ -597,10 +613,16 @@ class TC_00_ThinPool(ThinPoolBase):
} }
vm = qubes.tests.storage.TestVM(self) vm = qubes.tests.storage.TestVM(self)
volume = self.app.get_pool(self.pool.name).init_volume(vm, config) volume = self.app.get_pool(self.pool.name).init_volume(vm, config)
# mock logging, to not interfere with time.time() mock
volume.log = unittest.mock.Mock()
self.loop.run_until_complete(volume.create()) self.loop.run_until_complete(volume.create())
self.loop.run_until_complete(volume.start()) self.loop.run_until_complete(volume.start())
with unittest.mock.patch('time.time') as mock_time:
mock_time.side_effect = [521065906]
self.loop.run_until_complete(volume.stop()) self.loop.run_until_complete(volume.stop())
self.loop.run_until_complete(volume.start()) self.loop.run_until_complete(volume.start())
with unittest.mock.patch('time.time') as mock_time:
mock_time.side_effect = [521065907]
self.loop.run_until_complete(volume.stop()) self.loop.run_until_complete(volume.stop())
self.assertEqual(len(volume.revisions), 2) self.assertEqual(len(volume.revisions), 2)
revisions = volume.revisions revisions = volume.revisions
@ -632,10 +654,16 @@ class TC_00_ThinPool(ThinPoolBase):
} }
vm = qubes.tests.storage.TestVM(self) vm = qubes.tests.storage.TestVM(self)
volume = self.app.get_pool(self.pool.name).init_volume(vm, config) volume = self.app.get_pool(self.pool.name).init_volume(vm, config)
# mock logging, to not interfere with time.time() mock
volume.log = unittest.mock.Mock()
self.loop.run_until_complete(volume.create()) self.loop.run_until_complete(volume.create())
self.loop.run_until_complete(volume.start()) self.loop.run_until_complete(volume.start())
with unittest.mock.patch('time.time') as mock_time:
mock_time.side_effect = [521065906]
self.loop.run_until_complete(volume.stop()) self.loop.run_until_complete(volume.stop())
self.loop.run_until_complete(volume.start()) self.loop.run_until_complete(volume.start())
with unittest.mock.patch('time.time') as mock_time:
mock_time.side_effect = [521065907]
self.loop.run_until_complete(volume.stop()) self.loop.run_until_complete(volume.stop())
self.assertEqual(len(volume.revisions), 2) self.assertEqual(len(volume.revisions), 2)
revisions = volume.revisions revisions = volume.revisions