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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
import os
import oe.path
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
logger = logging.getLogger("selftest")
def qemu_launch(efi=False, machine=None, imagename='core-image-minimal', **kwargs):
qemu_bake_image(imagename)
return qemu_boot_image(efi=efi, machine=machine, imagename=imagename, **kwargs)
def qemu_terminate(s):
try:
s.terminate()
s.wait(timeout=10)
except KeyboardInterrupt:
pass
def qemu_boot_image(imagename, **kwargs):
# Create empty object.
args = type('', (), {})()
args.imagename = imagename
args.mac = kwargs.get('mac', None)
# Could use DEPLOY_DIR_IMAGE here but it's already in the machine
# subdirectory.
args.dir = 'tmp/deploy/images'
args.efi = kwargs.get('efi', False)
args.machine = kwargs.get('machine', None)
args.mem = kwargs.get('mem', '128M')
qemu_use_kvm = get_bb_var("QEMU_USE_KVM")
if qemu_use_kvm and \
(qemu_use_kvm == 'True' and 'x86' in args.machine or
get_bb_var('MACHINE') in qemu_use_kvm.split()):
args.kvm = True
else:
args.kvm = None # Autodetect
args.no_gui = kwargs.get('no_gui', True)
args.gdb = kwargs.get('gdb', False)
args.pcap = kwargs.get('pcap', None)
args.overlay = kwargs.get('overlay', None)
args.dry_run = kwargs.get('dry_run', False)
args.secondary_network = kwargs.get('secondary_network', False)
qemu = QemuCommand(args)
cmdline = qemu.command_line()
print('Booting image with run-qemu-ota...')
s = subprocess.Popen(cmdline)
sleep(kwargs.get('wait_for_boot_time', 10))
return qemu, s
def qemu_bake_image(imagename):
logger.info('Running bitbake to build {}'.format(imagename))
bitbake(imagename)
def qemu_send_command(port, command, timeout=120):
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=timeout)
return stdout, stderr, s2.returncode
def metadir():
# Assume the directory layout for finding other layers. We could also
# make assumptions by using 'show-layers', but either way, if the
# layers we need aren't where we expect them, we are out of luck.
path = os.path.abspath(os.path.dirname(__file__))
metadir = path + "/../../../../../"
return metadir
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
#
# Requirements in base test class (setUpClass for example):
# bitbake aktualizr-native
# bitbake build-sysroots -c build_native_sysroot
#
# (technique found in poky/meta/lib/oeqa/selftest/cases/package.py)
bb_vars = get_bb_vars(['STAGING_DIR', 'BUILD_ARCH'])
sysroot = oe.path.join(bb_vars['STAGING_DIR'], bb_vars['BUILD_ARCH'])
result = runCmd(cmd, native_sysroot=sysroot, ignore_status=True, **kwargs)
testInst.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output)
def verifyNotProvisioned(testInst, machine):
print('Checking output of aktualizr-info:')
ran_ok = False
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'':
ran_ok = True
break
sleep(delay)
testInst.assertTrue(ran_ok, 'aktualizr-info failed: ' + stderr.decode() + stdout.decode())
# Verify that device has NOT yet provisioned.
testInst.assertIn(b'Couldn\'t load device ID', stdout,
'Device already provisioned!? ' + stderr.decode() + stdout.decode())
testInst.assertIn(b'Couldn\'t load ECU serials', stdout,
'Device already provisioned!? ' + stderr.decode() + stdout.decode())
testInst.assertIn(b'Provisioned on server: no', stdout,
'Device already provisioned!? ' + stderr.decode() + stdout.decode())
testInst.assertIn(b'Fetched metadata: no', stdout,
'Device already provisioned!? ' + stderr.decode() + stdout.decode())
def verifyProvisioned(testInst, machine):
# Verify that device HAS provisioned.
ran_ok = False
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:
ran_ok = True
break
sleep(delay)
testInst.assertTrue(ran_ok, 'aktualizr-info failed: ' + stderr.decode() + stdout.decode())
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.info('Device successfully provisioned with ID: ' + m.group(1))
# vim:set ts=4 sw=4 sts=4 expandtab:
|
