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

tests: various fixes to integration tests

Browse Source 
Mostly add missing self.loop.run_until_complete().
This commit is contained in:
Marek Marczykowski-Górecki 2017-09-05 01:31:40 +02:00
parent d676576600
commit 687ca4092c
No known key found for this signature in database
GPG Key ID: 063938BA42CFA724
5 changed files with 59 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
qubes/tests/integ/backup.py
View File

@ -103,7 +103,8 @@ class BackupTestsMixin(object):
testnet = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=vmname, template=template, provides_network=True,
label='red')
testnet.create_on_disk(pool=pool)
self.loop.run_until_complete(
testnet.create_on_disk(pool=pool))
testnet.features['service.ntpd'] = True
vms.append(testnet)
self.fill_image(testnet.storage.export('private'), 20*1024*1024)
@ -114,7 +115,8 @@ class BackupTestsMixin(object):
name=vmname, template=template, label='red')
testvm1.uses_default_netvm = False
testvm1.netvm = testnet
testvm1.create_on_disk(pool=pool)
self.loop.run_until_complete(
testvm1.create_on_disk(pool=pool))
vms.append(testvm1)
self.fill_image(testvm1.storage.export('private'), 100 * 1024 * 1024)
@ -124,7 +126,8 @@ class BackupTestsMixin(object):
name=vmname,
virt_mode='hvm',
label='red')
testvm2.create_on_disk(pool=pool)
self.loop.run_until_complete(
testvm2.create_on_disk(pool=pool))
self.fill_image(testvm2.storage.export('root'), 1024 * 1024 * 1024, \
True)
vms.append(testvm2)
@ -133,7 +136,8 @@ class BackupTestsMixin(object):
self.log.debug("Creating %s" % vmname)
testvm3 = self.app.add_new_vm(qubes.vm.templatevm.TemplateVM,
name=vmname, label='red')
testvm3.create_on_disk(pool=pool)
self.loop.run_until_complete(
testvm3.create_on_disk(pool=pool))
self.fill_image(testvm3.storage.export('root'), 100 * 1024 * 1024, True)
vms.append(testvm3)
@ -141,7 +145,8 @@ class BackupTestsMixin(object):
self.log.debug("Creating %s" % vmname)
testvm4 = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=vmname, template=testvm3, label='red')
testvm4.create_on_disk(pool=pool)
self.loop.run_until_complete(
testvm4.create_on_disk(pool=pool))
vms.append(testvm4)
self.app.save()
@ -164,7 +169,7 @@ class BackupTestsMixin(object):
backup.target_dir = target
try:
backup.backup_do()
self.loop.run_until_complete(backup.backup_do())
except qubes.exc.QubesException as e:
if not expect_failure:
self.fail("QubesException during backup_do: %s" % str(e))
@ -174,6 +179,8 @@ class BackupTestsMixin(object):
def restore_backup(self, source=None, appvm=None, options=None,
expect_errors=None, manipulate_restore_info=None,
passphrase='qubes'):
self.skipTest('Test not converted to Qubes 4.0')
if source is None:
backupfile = os.path.join(self.backupdir,
sorted(os.listdir(self.backupdir))[-1])
@ -489,11 +496,11 @@ class TC_10_BackupVMMixin(BackupTestsMixin):
name=self.make_vm_name('backupvm'),
template=self.template
)
self.backupvm.create_on_disk()
self.loop.run_until_complete(self.backupvm.create_on_disk())
def test_100_send_to_vm_file_with_spaces(self):
vms = self.create_backup_vms()
self.backupvm.start()
self.loop.run_until_complete(self.backupvm.start())
self.loop.run_until_complete(self.backupvm.run_for_stdio(
"mkdir '/var/tmp/backup directory'"))
self.make_backup(vms, target_vm=self.backupvm,
@ -508,7 +515,7 @@ class TC_10_BackupVMMixin(BackupTestsMixin):
def test_110_send_to_vm_command(self):
vms = self.create_backup_vms()
self.backupvm.start()
self.loop.run_until_complete(self.backupvm.start())
self.make_backup(vms, target_vm=self.backupvm,
compressed=True, encrypted=True,
target='dd of=/var/tmp/backup-test')
@ -523,7 +530,7 @@ class TC_10_BackupVMMixin(BackupTestsMixin):
:return:
"""
vms = self.create_backup_vms()
self.backupvm.start()
self.loop.run_until_complete(self.backupvm.start())
self.loop.run_until_complete(self.backupvm.run_for_stdio(
# Debian 7 has too old losetup to handle loop-control device
"mknod /dev/loop0 b 7 0;"

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

@ -406,7 +406,9 @@ class TC_30_Gui_daemon(qubes.tests.SystemTestCase):
self.loop.run_until_complete(asyncio.wait([
testvm1.start(),
testvm2.start()]))
self.loop.run_until_complete(asyncio.wait([
self.wait_for_session(testvm1),
self.wait_for_session(testvm2)]))
window_title = 'user@{}'.format(testvm1.name)
self.loop.run_until_complete(testvm1.run(
'zenity --text-info --editable --title={}'.format(window_title)))

44
qubes/tests/integ/dispvm.py
View File

@ -99,33 +99,37 @@ class TC_20_DispVMMixin(object):
self.init_default_template(self.template)
self.disp_base = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=self.make_vm_name('dvm'),
label='red',
label='red', template_for_dispvms=True,
)
self.disp_base.create_on_disk()
self.loop.run_until_complete(self.disp_base.create_on_disk())
self.app.default_dispvm = self.disp_base
self.app.save()
def test_010_simple_dvm_run(self):
dispvm = qubes.vm.dispvm.DispVM.from_appvm(self.disp_base)
dispvm = self.loop.run_until_complete(
qubes.vm.dispvm.DispVM.from_appvm(self.disp_base))
try:
dispvm.start()
p = dispvm.run_service('qubes.VMShell', passio_popen=True)
self.loop.run_until_complete(dispvm.start())
p = self.loop.run_until_complete(
dispvm.run_service('qubes.VMShell', passio_popen=True))
(stdout, _) = p.communicate(input=b"echo test")
self.assertEqual(stdout, b"test\n")
finally:
dispvm.cleanup()
self.loop.run_until_complete(dispvm.cleanup())
@unittest.skipUnless(spawn.find_executable('xdotool'),
"xdotool not installed")
def test_020_gui_app(self):
dispvm = qubes.vm.dispvm.DispVM.from_appvm(self.disp_base)
dispvm = self.loop.run_until_complete(
qubes.vm.dispvm.DispVM.from_appvm(self.disp_base))
try:
dispvm.start()
p = dispvm.run_service('qubes.VMShell', passio_popen=True)
self.loop.run_until_complete(dispvm.start())
p = self.loop.run_until_complete(
dispvm.run_service('qubes.VMShell', passio_popen=True))
# wait for DispVM startup:
p.stdin.write(b"echo test\n")
p.stdin.flush()
l = p.stdout.readline()
l = self.loop.run_until_complete(p.stdout.readline())
self.assertEqual(l, b"test\n")
self.assertTrue(dispvm.is_running())
@ -143,7 +147,7 @@ class TC_20_DispVMMixin(object):
finally:
p.stdin.close()
finally:
dispvm.cleanup()
self.loop.run_until_complete(dispvm.cleanup())
self.assertNotIn(dispvm.name, self.app.domains,
"DispVM not removed from qubes.xml")
@ -210,14 +214,15 @@ class TC_20_DispVMMixin(object):
name=self.make_vm_name('vm1'),
label='red',
template=self.app.domains[self.template])
testvm1.create_on_disk()
self.loop.run_until_complete(testvm1.create_on_disk())
self.app.save()
testvm1.start()
testvm1.run("echo test1 > /home/user/test.txt", wait=True)
self.loop.run_until_complete(testvm1.start())
self.loop.run_until_complete(
testvm1.run_for_stdio("echo test1 > /home/user/test.txt"))
p = testvm1.run("qvm-open-in-dvm /home/user/test.txt",
passio_popen=True)
p = self.loop.run_until_complete(
testvm1.run("qvm-open-in-dvm /home/user/test.txt"))
wait_count = 0
winid = None
@ -246,10 +251,9 @@ class TC_20_DispVMMixin(object):
time.sleep(0.5)
self._handle_editor(winid)
p.wait()
p = testvm1.run("cat /home/user/test.txt",
passio_popen=True)
(test_txt_content, _) = p.communicate()
self.loop.run_until_complete(p.wait())
(test_txt_content, _) = self.loop.run_until_complete(
testvm1.run_for_stdio("cat /home/user/test.txt"))
# Drop BOM if added by editor
if test_txt_content.startswith(b'\xef\xbb\xbf'):
test_txt_content = test_txt_content[3:]

4
qubes/tests/integ/network.py
View File

@ -64,12 +64,12 @@ class VmNetworkingMixin(object):
self.testnetvm = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=self.make_vm_name('netvm1'),
label='red')
self.testnetvm.create_on_disk()
self.loop.run_until_complete(self.testnetvm.create_on_disk())
self.testnetvm.provides_network = True
self.testvm1 = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=self.make_vm_name('vm2'),
label='red')
self.testvm1.create_on_disk()
self.loop.run_until_complete(self.testvm1.create_on_disk())
self.testvm1.netvm = self.testnetvm
self.app.save()

22
qubes/tests/integ/vm_qrexec_gui.py
View File

@ -594,12 +594,13 @@ class TC_00_AppVMMixin(object):
self.testvm2.start()]))
self.loop.run_until_complete(self.testvm1.run_for_stdio(
'cp /etc/passwd passwd'))
'cp /etc/passwd /tmp/passwd'))
with self.qrexec_policy('qubes.Filecopy', self.testvm1, self.testvm2):
try:
self.loop.run_until_complete(
self.testvm1.run_for_stdio(
'qvm-move-to-vm {} passwd'.format(self.testvm2.name)))
'qvm-move-to-vm {} /tmp/passwd'.format(
self.testvm2.name)))
except subprocess.CalledProcessError as e:
self.fail('qvm-move-to-vm failed: {}'.format(e.stderr))
@ -612,7 +613,7 @@ class TC_00_AppVMMixin(object):
with self.assertRaises(subprocess.CalledProcessError):
self.loop.run_until_complete(self.testvm1.run_for_stdio(
'test -f passwd'))
'test -f /tmp/passwd'))
def test_101_qrexec_filecopy_with_autostart(self):
self.loop.run_until_complete(self.testvm1.start())
@ -693,7 +694,8 @@ class TC_00_AppVMMixin(object):
# Prepare test file
self.loop.run_until_complete(self.testvm1.run_for_stdio(
'yes teststring | dd of=testfile bs=1M count=50 iflag=fullblock'))
'yes teststring | dd of=/tmp/testfile bs=1M count=50 '
'iflag=fullblock'))
# Prepare target directory with limited size
self.loop.run_until_complete(self.testvm2.run_for_stdio(
@ -705,14 +707,15 @@ class TC_00_AppVMMixin(object):
with self.qrexec_policy('qubes.Filecopy', self.testvm1, self.testvm2):
with self.assertRaises(subprocess.CalledProcessError):
self.loop.run_until_complete(self.testvm1.run_for_stdio(
'qvm-move-to-vm {} testfile'.format(self.testvm2.name)))
'qvm-move-to-vm {} /tmp/testfile'.format(
self.testvm2.name)))
# Close GUI error message
self.enter_keys_in_window('Error', ['Return'])
# the file shouldn't be removed in source vm
self.loop.run_until_complete(self.testvm1.run_for_stdio(
'test -f testfile'))
'test -f /tmp/testfile'))
def test_200_timezone(self):
"""Test whether timezone setting is properly propagated to the VM"""
@ -761,7 +764,7 @@ class TC_00_AppVMMixin(object):
self.testvm1.run_for_stdio('date -u +%s'))
self.assertAlmostEquals(int(vm_time), int(start_time), delta=30)
dom0_time, _ = subprocess.check_output(['date', '-u', '+%s'])
dom0_time = subprocess.check_output(['date', '-u', '+%s'])
self.assertAlmostEquals(int(dom0_time), int(start_time), delta=30)
except:
@ -887,10 +890,11 @@ int main(int argc, char **argv) {
"grep ^MemFree: /proc/meminfo|awk '{print $2}'")
memory_pages = int(stdout) // 4 # 4k pages
alloc1 = yield from self.testvm1.run_for_stdio(
alloc1 = yield from self.testvm1.run(
'ulimit -l unlimited; exec /home/user/allocator {}'.format(
memory_pages),
user="root")
user="root",
stdin=subprocess.PIPE, stdout=subprocess.PIPE)
# wait for memory being allocated; can't use just .read(), because EOF
# passing is unreliable while the process is still running