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

tests: fix various object and FD leaks

Browse Source 
- Prefer instance attributes over local variables - the former ones do
not leak into traceback object and are cleaned up by tests framework.
- Use 'with' syntax for handling files.
- Use subprocess.DEVNULL instead of open('/dev/null') where applicable
- Delete local variables when not needed anymore.
This commit is contained in:
Marek Marczykowski-Górecki 2017-09-28 03:03:30 +02:00
parent d96d4a94e1
commit 3c54244fdc
No known key found for this signature in database
GPG Key ID: 063938BA42CFA724
3 changed files with 52 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
qubes/tests/integ/basic.py
View File

@ -183,6 +183,10 @@ class TC_01_Properties(qubes.tests.SystemTestCase):
self.assertEqual(testvm1.firewall.rules,
testvm2.firewall.rules)
finally:
try:
del firewall
except NameError:
pass
try:
del testvm1
except NameError:
@ -483,29 +487,29 @@ class TC_05_StandaloneVM(qubes.tests.SystemTestCase):
self.init_default_template()
def test_000_create_start(self):
testvm1 = self.app.add_new_vm(qubes.vm.standalonevm.StandaloneVM,
self.testvm1 = self.app.add_new_vm(qubes.vm.standalonevm.StandaloneVM,
name=self.make_vm_name('vm1'), label='red')
testvm1.features['qrexec'] = True
self.testvm1.features['qrexec'] = True
self.loop.run_until_complete(
testvm1.clone_disk_files(self.app.default_template))
self.testvm1.clone_disk_files(self.app.default_template))
self.app.save()
self.loop.run_until_complete(testvm1.start())
self.assertEqual(testvm1.get_power_state(), "Running")
self.loop.run_until_complete(self.testvm1.start())
self.assertEqual(self.testvm1.get_power_state(), "Running")
def test_100_resize_root_img(self):
testvm1 = self.app.add_new_vm(qubes.vm.standalonevm.StandaloneVM,
self.testvm1 = self.app.add_new_vm(qubes.vm.standalonevm.StandaloneVM,
name=self.make_vm_name('vm1'), label='red')
testvm1.features['qrexec'] = True
self.testvm1.features['qrexec'] = True
self.loop.run_until_complete(
testvm1.clone_disk_files(self.app.default_template))
self.testvm1.clone_disk_files(self.app.default_template))
self.app.save()
self.loop.run_until_complete(
testvm1.storage.resize(testvm1.volumes['root'], 20 * 1024 ** 3))
self.assertEqual(testvm1.volumes['root'].size, 20 * 1024 ** 3)
self.loop.run_until_complete(testvm1.start())
self.testvm1.storage.resize(self.testvm1.volumes['root'], 20 * 1024 ** 3))
self.assertEqual(self.testvm1.volumes['root'].size, 20 * 1024 ** 3)
self.loop.run_until_complete(self.testvm1.start())
# new_size in 1k-blocks
(new_size, _) = self.loop.run_until_complete(
testvm1.run_for_stdio('df --output=size /|tail -n 1'))
self.testvm1.run_for_stdio('df --output=size /|tail -n 1'))
# some safety margin for FS metadata
self.assertGreater(int(new_size.strip()), 19 * 1024 ** 2)

31
qubes/tests/integ/dom0_update.py
View File

@ -115,7 +115,7 @@ enabled = 1
self.app.updatevm = self.updatevm
self.app.save()
subprocess.call(['sudo', 'rpm', '-e', self.pkg_name],
stderr=open(os.devnull, 'w'))
stderr=subprocess.DEVNULL)
subprocess.check_call(['sudo', 'rpm', '--import',
os.path.join(self.tmpdir, 'pubkey.asc')])
self.loop.run_until_complete(self.updatevm.start())
@ -129,10 +129,10 @@ enabled = 1
del self.repo_proc
super(TC_00_Dom0UpgradeMixin, self).tearDown()
subprocess.call(['sudo', 'rpm', '-e', self.pkg_name], stderr=open(
os.devnull, 'w'))
subprocess.call(['sudo', 'rpm', '-e', self.pkg_name],
stderr=subprocess.DEVNULL)
subprocess.call(['sudo', 'rpm', '-e', 'gpg-pubkey-{}'.format(
self.keyid)], stderr=open(os.devnull, 'w'))
self.keyid)], stderr=subprocess.DEVNULL)
for pkg in os.listdir(self.tmpdir):
if pkg.endswith('.rpm'):
@ -171,17 +171,18 @@ Test package
['rpm', '--quiet', '--define=_gpg_path {}'.format(dir),
'--define=_gpg_name {}'.format("Qubes test"),
'--addsign', pkg_path],
stdin=open(os.devnull),
stdout=open(os.devnull, 'w'),
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.STDOUT)
subprocess.check_call(['sudo', 'chmod', 'go+rw', '/dev/tty'])
return pkg_path
def send_pkg(self, filename):
self.loop.run_until_complete(self.updatevm.run_for_stdio(
'mkdir -p /tmp/repo; cat > /tmp/repo/{}'.format(
os.path.basename(filename)),
input=open(filename, 'rb').read()))
with open(filename, 'rb') as f_pkg:
self.loop.run_until_complete(self.updatevm.run_for_stdio(
'mkdir -p /tmp/repo; cat > /tmp/repo/{}'.format(
os.path.basename(filename)),
input=f_pkg.read()))
try:
self.loop.run_until_complete(
self.updatevm.run_for_stdio('cd /tmp/repo; createrepo .'))
@ -229,11 +230,11 @@ Test package
del proc
retcode = subprocess.call(['rpm', '-q', '{}-1.0'.format(
self.pkg_name)], stdout=open(os.devnull, 'w'))
self.pkg_name)], stdout=subprocess.DEVNULL)
self.assertEqual(retcode, 1, 'Package {}-1.0 still installed after '
'update'.format(self.pkg_name))
retcode = subprocess.call(['rpm', '-q', '{}-2.0'.format(
self.pkg_name)], stdout=open(os.devnull, 'w'))
self.pkg_name)], stdout=subprocess.DEVNULL)
self.assertEqual(retcode, 0, 'Package {}-2.0 not installed after '
'update'.format(self.pkg_name))
self.assertFalse(os.path.exists(self.update_flag_path),
@ -350,7 +351,7 @@ Test package
del proc
retcode = subprocess.call(['rpm', '-q', '{}-1.0'.format(
self.pkg_name)], stdout=open('/dev/null', 'w'))
self.pkg_name)], stdout=subprocess.DEVNULL)
self.assertEqual(retcode, 1,
'Package {}-1.0 installed although '
'signature is invalid'.format(self.pkg_name))
@ -358,7 +359,7 @@ Test package
def test_030_install_unsigned(self):
filename = self.create_pkg(self.tmpdir, self.pkg_name, '1.0')
subprocess.check_call(['rpm', '--delsign', filename],
stdout=open(os.devnull, 'w'),
stdout=subprocess.DEVNULL,
stderr=subprocess.STDOUT)
self.send_pkg(filename)
@ -378,7 +379,7 @@ Test package
del proc
retcode = subprocess.call(['rpm', '-q', '{}-1.0'.format(
self.pkg_name)], stdout=open('/dev/null', 'w'))
self.pkg_name)], stdout=subprocess.DEVNULL)
self.assertEqual(retcode, 1,
'UNSIGNED package {}-1.0 installed'.format(self.pkg_name))

20
qubes/tests/integ/storage.py
View File

@ -46,6 +46,12 @@ class StorageTestMixin(object):
self.init_pool()
self.app.save()
def tearDown(self):
del self.vm1
del self.vm2
del self.pool
super(StorageTestMixin, self).tearDown()
def init_pool(self):
''' Initialize storage pool to be tested, store it in self.pool'''
raise NotImplementedError
@ -65,8 +71,10 @@ class StorageTestMixin(object):
}
testvol = self.vm1.storage.init_volume('testvol', volume_config)
coro_maybe = testvol.create()
del testvol
if asyncio.iscoroutine(coro_maybe):
yield from coro_maybe
del coro_maybe
self.app.save()
yield from (self.vm1.start())
@ -98,8 +106,10 @@ class StorageTestMixin(object):
}
testvol = self.vm1.storage.init_volume('testvol', volume_config)
coro_maybe = testvol.create()
del testvol
if asyncio.iscoroutine(coro_maybe):
yield from coro_maybe
del coro_maybe
self.app.save()
yield from self.vm1.start()
# non-volatile image not clean
@ -133,8 +143,10 @@ class StorageTestMixin(object):
}
testvol = self.vm1.storage.init_volume('testvol', volume_config)
coro_maybe = testvol.create()
del testvol
if asyncio.iscoroutine(coro_maybe):
yield from coro_maybe
del coro_maybe
self.app.save()
yield from self.vm1.start()
# non-volatile image not clean
@ -167,6 +179,7 @@ class StorageTestMixin(object):
coro_maybe = testvol.create()
if asyncio.iscoroutine(coro_maybe):
yield from coro_maybe
del coro_maybe
volume_config = {
'pool': self.pool.name,
'size': size,
@ -174,10 +187,13 @@ class StorageTestMixin(object):
'source': testvol.vid,
'rw': True,
}
del testvol
testvol_snap = self.vm2.storage.init_volume('testvol', volume_config)
coro_maybe = testvol_snap.create()
del testvol_snap
if asyncio.iscoroutine(coro_maybe):
yield from coro_maybe
del coro_maybe
self.app.save()
yield from self.vm1.start()
yield from self.vm2.start()
@ -252,6 +268,7 @@ class StorageTestMixin(object):
coro_maybe = testvol.create()
if asyncio.iscoroutine(coro_maybe):
yield from coro_maybe
del coro_maybe
volume_config = {
'pool': self.pool.name,
'size': size,
@ -259,10 +276,13 @@ class StorageTestMixin(object):
'source': testvol.vid,
'rw': True,
}
del testvol
testvol_snap = self.vm2.storage.init_volume('testvol', volume_config)
coro_maybe = testvol_snap.create()
del testvol_snap
if asyncio.iscoroutine(coro_maybe):
yield from coro_maybe
del coro_maybe
self.app.save()
yield from self.vm2.start()