summaryrefslogtreecommitdiffstats
path: root/lib/oeqa/selftest/cases/testutils.py
diff options
context:
space:
mode:
Diffstat (limited to 'lib/oeqa/selftest/cases/testutils.py')
-rw-r--r--lib/oeqa/selftest/cases/testutils.py103
1 files changed, 103 insertions, 0 deletions
diff --git a/lib/oeqa/selftest/cases/testutils.py b/lib/oeqa/selftest/cases/testutils.py
new file mode 100644
index 0000000..77bcad7
--- /dev/null
+++ b/lib/oeqa/selftest/cases/testutils.py
@@ -0,0 +1,103 @@
1import os
2import logging
3import re
4import subprocess
5from time import sleep
6
7from oeqa.utils.commands import runCmd, bitbake, get_bb_var, get_bb_vars
8from qemucommand import QemuCommand
9
10
11def qemu_launch(efi=False, machine=None, imagename=None):
12 logger = logging.getLogger("selftest")
13 logger.info('Running bitbake to build core-image-minimal')
14 bitbake('core-image-minimal')
15 # Create empty object.
16 args = type('', (), {})()
17 if imagename:
18 args.imagename = imagename
19 else:
20 args.imagename = 'core-image-minimal'
21 args.mac = None
22 # Could use DEPLOY_DIR_IMAGE here but it's already in the machine
23 # subdirectory.
24 args.dir = 'tmp/deploy/images'
25 args.efi = efi
26 args.machine = machine
27 qemu_use_kvm = get_bb_var("QEMU_USE_KVM")
28 if qemu_use_kvm and \
29 (qemu_use_kvm == 'True' and 'x86' in machine or
30 get_bb_var('MACHINE') in qemu_use_kvm.split()):
31 args.kvm = True
32 else:
33 args.kvm = None # Autodetect
34 args.no_gui = True
35 args.gdb = False
36 args.pcap = None
37 args.overlay = None
38 args.dry_run = False
39 args.secondary_network = False
40
41 qemu = QemuCommand(args)
42 cmdline = qemu.command_line()
43 print('Booting image with run-qemu-ota...')
44 s = subprocess.Popen(cmdline)
45 sleep(10)
46 return qemu, s
47
48
49def qemu_terminate(s):
50 try:
51 s.terminate()
52 except KeyboardInterrupt:
53 pass
54
55
56def qemu_send_command(port, command):
57 command = ['ssh -q -o UserKnownHostsFile=/dev/null -o StrictHostKeyChecking=no root@localhost -p ' +
58 str(port) + ' "' + command + '"']
59 s2 = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
60 stdout, stderr = s2.communicate(timeout=60)
61 return stdout, stderr, s2.returncode
62
63
64def akt_native_run(testInst, cmd, **kwargs):
65 # run a command supplied by aktualizr-native and checks that:
66 # - the executable exists
67 # - the command runs without error
68 # NOTE: the base test class must have built aktualizr-native (in
69 # setUpClass, for example)
70 bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'base_prefix', 'libdir', 'bindir'],
71 'aktualizr-native')
72 sysroot = bb_vars['SYSROOT_DESTDIR'] + bb_vars['base_prefix']
73 sysrootbin = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir']
74 libdir = bb_vars['libdir']
75
76 program, *_ = cmd.split(' ')
77 p = '{}/{}'.format(sysrootbin, program)
78 testInst.assertTrue(os.path.isfile(p), msg="No {} found ({})".format(program, p))
79 env = dict(os.environ)
80 env['LD_LIBRARY_PATH'] = libdir
81 result = runCmd(cmd, env=env, native_sysroot=sysroot, ignore_status=True, **kwargs)
82 testInst.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output)
83
84
85def verifyProvisioned(testInst, machine):
86 # Verify that device HAS provisioned.
87 for delay in [5, 5, 5, 5, 10, 10, 10, 10]:
88 stdout, stderr, retcode = testInst.qemu_command('aktualizr-info')
89 if retcode == 0 and stderr == b'' and stdout.decode().find('Fetched metadata: yes') >= 0:
90 break
91 sleep(delay)
92 testInst.assertIn(b'Device ID: ', stdout, 'Provisioning failed: ' + stderr.decode() + stdout.decode())
93 testInst.assertIn(b'Primary ecu hardware ID: ' + machine.encode(), stdout,
94 'Provisioning failed: ' + stderr.decode() + stdout.decode())
95 testInst.assertIn(b'Fetched metadata: yes', stdout, 'Provisioning failed: ' + stderr.decode() + stdout.decode())
96 p = re.compile(r'Device ID: ([a-z0-9-]*)\n')
97 m = p.search(stdout.decode())
98 testInst.assertTrue(m, 'Device ID could not be read: ' + stderr.decode() + stdout.decode())
99 testInst.assertGreater(m.lastindex, 0, 'Device ID could not be read: ' + stderr.decode() + stdout.decode())
100 logger = logging.getLogger("selftest")
101 logger.info('Device successfully provisioned with ID: ' + m.group(1))
102
103# vim:set ts=4 sw=4 sts=4 expandtab: