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