summaryrefslogtreecommitdiffstats
path: root/lib/oeqa/selftest/cases/updater.py
diff options
context:
space:
mode:
authorLaurent Bonnans <laurent.bonnans@here.com>2019-03-19 16:45:34 +0100
committerLaurent Bonnans <laurent.bonnans@here.com>2019-03-20 14:13:41 +0100
commit8b98e1e0f908ef30f5a4459f2bd62442d2b6649b (patch)
tree59d17466158dbba27c0371294819fa52e94861e1 /lib/oeqa/selftest/cases/updater.py
parentececedcbd58a7cd04eea0a7faf7b04939536a555 (diff)
downloadmeta-updater-8b98e1e0f908ef30f5a4459f2bd62442d2b6649b.tar.gz
Split oe-selftests by target machines
To allow for more targeted testing Signed-off-by: Laurent Bonnans <laurent.bonnans@here.com>
Diffstat (limited to 'lib/oeqa/selftest/cases/updater.py')
-rw-r--r--lib/oeqa/selftest/cases/updater.py704
1 files changed, 0 insertions, 704 deletions
diff --git a/lib/oeqa/selftest/cases/updater.py b/lib/oeqa/selftest/cases/updater.py
deleted file mode 100644
index 473b2a8..0000000
--- a/lib/oeqa/selftest/cases/updater.py
+++ /dev/null
@@ -1,704 +0,0 @@
1# pylint: disable=C0111,C0325
2import os
3import logging
4import re
5import subprocess
6import unittest
7from time import sleep
8
9from oeqa.selftest.case import OESelftestTestCase
10from oeqa.utils.commands import runCmd, bitbake, get_bb_var, get_bb_vars
11from qemucommand import QemuCommand
12
13
14class SotaToolsTests(OESelftestTestCase):
15
16 @classmethod
17 def setUpClass(cls):
18 super(SotaToolsTests, cls).setUpClass()
19 logger = logging.getLogger("selftest")
20 logger.info('Running bitbake to build aktualizr-native tools')
21 bitbake('aktualizr-native')
22
23 def test_push_help(self):
24 akt_native_run(self, 'garage-push --help')
25
26 def test_deploy_help(self):
27 akt_native_run(self, 'garage-deploy --help')
28
29 def test_garagesign_help(self):
30 akt_native_run(self, 'garage-sign --help')
31
32
33class GeneralTests(OESelftestTestCase):
34
35 def test_feature_sota(self):
36 result = get_bb_var('DISTRO_FEATURES').find('sota')
37 self.assertNotEqual(result, -1, 'Feature "sota" not set at DISTRO_FEATURES')
38
39 def test_feature_usrmerge(self):
40 result = get_bb_var('DISTRO_FEATURES').find('usrmerge')
41 self.assertNotEqual(result, -1, 'Feature "sota" not set at DISTRO_FEATURES')
42
43 def test_feature_systemd(self):
44 result = get_bb_var('DISTRO_FEATURES').find('systemd')
45 self.assertNotEqual(result, -1, 'Feature "systemd" not set at DISTRO_FEATURES')
46
47 def test_credentials(self):
48 logger = logging.getLogger("selftest")
49 logger.info('Running bitbake to build core-image-minimal')
50 self.append_config('SOTA_CLIENT_PROV = "aktualizr-auto-prov"')
51 bitbake('core-image-minimal')
52 credentials = get_bb_var('SOTA_PACKED_CREDENTIALS')
53 # skip the test if the variable SOTA_PACKED_CREDENTIALS is not set
54 if credentials is None:
55 raise unittest.SkipTest("Variable 'SOTA_PACKED_CREDENTIALS' not set.")
56 # Check if the file exists
57 self.assertTrue(os.path.isfile(credentials), "File %s does not exist" % credentials)
58 deploydir = get_bb_var('DEPLOY_DIR_IMAGE')
59 imagename = get_bb_var('IMAGE_LINK_NAME', 'core-image-minimal')
60 # Check if the credentials are included in the output image
61 result = runCmd('tar -jtvf %s/%s.tar.bz2 | grep sota_provisioning_credentials.zip' %
62 (deploydir, imagename), ignore_status=True)
63 self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output)
64
65 def test_java(self):
66 result = runCmd('which java', ignore_status=True)
67 self.assertEqual(result.status, 0,
68 "Java not found. Do you have a JDK installed on your host machine?")
69
70
71class AktualizrToolsTests(OESelftestTestCase):
72
73 @classmethod
74 def setUpClass(cls):
75 super(AktualizrToolsTests, cls).setUpClass()
76 logger = logging.getLogger("selftest")
77 logger.info('Running bitbake to build aktualizr-native tools')
78 bitbake('aktualizr-native')
79
80 def test_cert_provider_help(self):
81 akt_native_run(self, 'aktualizr-cert-provider --help')
82
83 def test_cert_provider_local_output(self):
84 logger = logging.getLogger("selftest")
85 logger.info('Running bitbake to build aktualizr-ca-implicit-prov')
86 bitbake('aktualizr-ca-implicit-prov')
87 bb_vars = get_bb_vars(['SOTA_PACKED_CREDENTIALS', 'T'], 'aktualizr-native')
88 creds = bb_vars['SOTA_PACKED_CREDENTIALS']
89 temp_dir = bb_vars['T']
90 bb_vars_prov = get_bb_vars(['STAGING_DIR_HOST', 'libdir'], 'aktualizr-ca-implicit-prov')
91 config = bb_vars_prov['STAGING_DIR_HOST'] + bb_vars_prov['libdir'] + '/sota/sota_implicit_prov_ca.toml'
92
93 akt_native_run(self, 'aktualizr-cert-provider -c {creds} -r -l {temp} -g {config}'
94 .format(creds=creds, temp=temp_dir, config=config))
95
96 # Might be nice if these names weren't hardcoded.
97 cert_path = temp_dir + '/var/sota/import/client.pem'
98 self.assertTrue(os.path.isfile(cert_path), "Client certificate not found at %s." % cert_path)
99 self.assertTrue(os.path.getsize(cert_path) > 0, "Client certificate at %s is empty." % cert_path)
100 pkey_path = temp_dir + '/var/sota/import/pkey.pem'
101 self.assertTrue(os.path.isfile(pkey_path), "Private key not found at %s." % pkey_path)
102 self.assertTrue(os.path.getsize(pkey_path) > 0, "Private key at %s is empty." % pkey_path)
103 ca_path = temp_dir + '/var/sota/import/root.crt'
104 self.assertTrue(os.path.isfile(ca_path), "Client certificate not found at %s." % ca_path)
105 self.assertTrue(os.path.getsize(ca_path) > 0, "Client certificate at %s is empty." % ca_path)
106
107
108class AutoProvTests(OESelftestTestCase):
109
110 def setUpLocal(self):
111 layer = "meta-updater-qemux86-64"
112 result = runCmd('bitbake-layers show-layers')
113 if re.search(layer, result.output) is None:
114 # Assume the directory layout for finding other layers. We could also
115 # make assumptions by using 'show-layers', but either way, if the
116 # layers we need aren't where we expect them, we are out of luck.
117 path = os.path.abspath(os.path.dirname(__file__))
118 metadir = path + "/../../../../../"
119 self.meta_qemu = metadir + layer
120 runCmd('bitbake-layers add-layer "%s"' % self.meta_qemu)
121 else:
122 self.meta_qemu = None
123 self.append_config('MACHINE = "qemux86-64"')
124 self.append_config('SOTA_CLIENT_PROV = " aktualizr-auto-prov "')
125 self.qemu, self.s = qemu_launch(machine='qemux86-64')
126
127 def tearDownLocal(self):
128 qemu_terminate(self.s)
129 if self.meta_qemu:
130 runCmd('bitbake-layers remove-layer "%s"' % self.meta_qemu, ignore_status=True)
131
132 def qemu_command(self, command):
133 return qemu_send_command(self.qemu.ssh_port, command)
134
135 def test_provisioning(self):
136 print('Checking machine name (hostname) of device:')
137 stdout, stderr, retcode = self.qemu_command('hostname')
138 self.assertEqual(retcode, 0, "Unable to check hostname. " +
139 "Is an ssh daemon (such as dropbear or openssh) installed on the device?")
140 machine = get_bb_var('MACHINE', 'core-image-minimal')
141 self.assertEqual(stderr, b'', 'Error: ' + stderr.decode())
142 # Strip off line ending.
143 value = stdout.decode()[:-1]
144 self.assertEqual(value, machine,
145 'MACHINE does not match hostname: ' + machine + ', ' + value)
146 print(value)
147 print('Checking output of aktualizr-info:')
148 ran_ok = False
149 for delay in [1, 2, 5, 10, 15]:
150 stdout, stderr, retcode = self.qemu_command('aktualizr-info')
151 if retcode == 0 and stderr == b'':
152 ran_ok = True
153 break
154 sleep(delay)
155 self.assertTrue(ran_ok, 'aktualizr-info failed: ' + stderr.decode() + stdout.decode())
156
157 verifyProvisioned(self, machine)
158
159
160class ManualControlTests(OESelftestTestCase):
161
162 def setUpLocal(self):
163 layer = "meta-updater-qemux86-64"
164 result = runCmd('bitbake-layers show-layers')
165 if re.search(layer, result.output) is None:
166 # Assume the directory layout for finding other layers. We could also
167 # make assumptions by using 'show-layers', but either way, if the
168 # layers we need aren't where we expect them, we are out of like.
169 path = os.path.abspath(os.path.dirname(__file__))
170 metadir = path + "/../../../../../"
171 self.meta_qemu = metadir + layer
172 runCmd('bitbake-layers add-layer "%s"' % self.meta_qemu)
173 else:
174 self.meta_qemu = None
175 self.append_config('MACHINE = "qemux86-64"')
176 self.append_config('SOTA_CLIENT_PROV = " aktualizr-auto-prov "')
177 self.append_config('SYSTEMD_AUTO_ENABLE_aktualizr = "disable"')
178 self.qemu, self.s = qemu_launch(machine='qemux86-64')
179
180 def tearDownLocal(self):
181 qemu_terminate(self.s)
182 if self.meta_qemu:
183 runCmd('bitbake-layers remove-layer "%s"' % self.meta_qemu, ignore_status=True)
184
185 def qemu_command(self, command):
186 return qemu_send_command(self.qemu.ssh_port, command)
187
188 def test_manual_run_mode_once(self):
189 """
190 Disable the systemd service then run aktualizr manually
191 """
192 sleep(20)
193 stdout, stderr, retcode = self.qemu_command('aktualizr-info')
194 self.assertIn(b'Can\'t open database', stderr,
195 'Aktualizr should not have run yet' + stderr.decode() + stdout.decode())
196
197 stdout, stderr, retcode = self.qemu_command('aktualizr once')
198
199 stdout, stderr, retcode = self.qemu_command('aktualizr-info')
200 self.assertIn(b'Fetched metadata: yes', stdout,
201 'Aktualizr should have run' + stderr.decode() + stdout.decode())
202
203
204class RpiTests(OESelftestTestCase):
205
206 def setUpLocal(self):
207 # Add layers before changing the machine type, otherwise the sanity
208 # checker complains loudly.
209 layer_rpi = "meta-raspberrypi"
210 layer_upd_rpi = "meta-updater-raspberrypi"
211 result = runCmd('bitbake-layers show-layers')
212 # Assume the directory layout for finding other layers. We could also
213 # make assumptions by using 'show-layers', but either way, if the
214 # layers we need aren't where we expect them, we are out of luck.
215 path = os.path.abspath(os.path.dirname(__file__))
216 metadir = path + "/../../../../../"
217 if re.search(layer_rpi, result.output) is None:
218 self.meta_rpi = metadir + layer_rpi
219 runCmd('bitbake-layers add-layer "%s"' % self.meta_rpi)
220 else:
221 self.meta_rpi = None
222 if re.search(layer_upd_rpi, result.output) is None:
223 self.meta_upd_rpi = metadir + layer_upd_rpi
224 runCmd('bitbake-layers add-layer "%s"' % self.meta_upd_rpi)
225 else:
226 self.meta_upd_rpi = None
227
228 # This is trickier that I would've thought. The fundamental problem is
229 # that the qemu layer changes the u-boot file extension to .rom, but
230 # raspberrypi still expects .bin. To prevent this, the qemu layer must
231 # be temporarily removed if it is present. It has to be removed by name
232 # without the complete path, but to add it back when we are done, we
233 # need the full path.
234 p = re.compile(r'meta-updater-qemux86-64\s*(\S*meta-updater-qemux86-64)\s')
235 m = p.search(result.output)
236 if m and m.lastindex > 0:
237 self.meta_qemu = m.group(1)
238 runCmd('bitbake-layers remove-layer meta-updater-qemux86-64')
239 else:
240 self.meta_qemu = None
241
242 self.append_config('MACHINE = "raspberrypi3"')
243 self.append_config('SOTA_CLIENT_PROV = " aktualizr-auto-prov "')
244
245 def tearDownLocal(self):
246 if self.meta_qemu:
247 runCmd('bitbake-layers add-layer "%s"' % self.meta_qemu, ignore_status=True)
248 if self.meta_upd_rpi:
249 runCmd('bitbake-layers remove-layer "%s"' % self.meta_upd_rpi, ignore_status=True)
250 if self.meta_rpi:
251 runCmd('bitbake-layers remove-layer "%s"' % self.meta_rpi, ignore_status=True)
252
253 def test_rpi(self):
254 logger = logging.getLogger("selftest")
255 logger.info('Running bitbake to build core-image-minimal')
256 self.append_config('SOTA_CLIENT_PROV = "aktualizr-auto-prov"')
257 bitbake('core-image-minimal')
258 credentials = get_bb_var('SOTA_PACKED_CREDENTIALS')
259 # Skip the test if the variable SOTA_PACKED_CREDENTIALS is not set.
260 if credentials is None:
261 raise unittest.SkipTest("Variable 'SOTA_PACKED_CREDENTIALS' not set.")
262 # Check if the file exists.
263 self.assertTrue(os.path.isfile(credentials), "File %s does not exist" % credentials)
264 deploydir = get_bb_var('DEPLOY_DIR_IMAGE')
265 imagename = get_bb_var('IMAGE_LINK_NAME', 'core-image-minimal')
266 # Check if the credentials are included in the output image.
267 result = runCmd('tar -jtvf %s/%s.tar.bz2 | grep sota_provisioning_credentials.zip' %
268 (deploydir, imagename), ignore_status=True)
269 self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output)
270
271
272class GrubTests(OESelftestTestCase):
273
274 def setUpLocal(self):
275 layer_intel = "meta-intel"
276 layer_minnow = "meta-updater-minnowboard"
277 result = runCmd('bitbake-layers show-layers')
278 # Assume the directory layout for finding other layers. We could also
279 # make assumptions by using 'show-layers', but either way, if the
280 # layers we need aren't where we expect them, we are out of luck.
281 path = os.path.abspath(os.path.dirname(__file__))
282 metadir = path + "/../../../../../"
283 if re.search(layer_intel, result.output) is None:
284 self.meta_intel = metadir + layer_intel
285 runCmd('bitbake-layers add-layer "%s"' % self.meta_intel)
286 else:
287 self.meta_intel = None
288 if re.search(layer_minnow, result.output) is None:
289 self.meta_minnow = metadir + layer_minnow
290 runCmd('bitbake-layers add-layer "%s"' % self.meta_minnow)
291 else:
292 self.meta_minnow = None
293 self.append_config('MACHINE = "intel-corei7-64"')
294 self.append_config('OSTREE_BOOTLOADER = "grub"')
295 self.append_config('SOTA_CLIENT_PROV = " aktualizr-auto-prov "')
296 self.qemu, self.s = qemu_launch(efi=True, machine='intel-corei7-64')
297
298 def tearDownLocal(self):
299 qemu_terminate(self.s)
300 if self.meta_intel:
301 runCmd('bitbake-layers remove-layer "%s"' % self.meta_intel, ignore_status=True)
302 if self.meta_minnow:
303 runCmd('bitbake-layers remove-layer "%s"' % self.meta_minnow, ignore_status=True)
304
305 def qemu_command(self, command):
306 return qemu_send_command(self.qemu.ssh_port, command)
307
308 def test_grub(self):
309 print('Checking machine name (hostname) of device:')
310 stdout, stderr, retcode = self.qemu_command('hostname')
311 self.assertEqual(retcode, 0, "Unable to check hostname. " +
312 "Is an ssh daemon (such as dropbear or openssh) installed on the device?")
313 machine = get_bb_var('MACHINE', 'core-image-minimal')
314 self.assertEqual(stderr, b'', 'Error: ' + stderr.decode())
315 # Strip off line ending.
316 value = stdout.decode()[:-1]
317 self.assertEqual(value, machine,
318 'MACHINE does not match hostname: ' + machine + ', ' + value +
319 '\nIs TianoCore ovmf installed on your host machine?')
320 print(value)
321 print('Checking output of aktualizr-info:')
322 ran_ok = False
323 for delay in [1, 2, 5, 10, 15]:
324 stdout, stderr, retcode = self.qemu_command('aktualizr-info')
325 if retcode == 0 and stderr == b'':
326 ran_ok = True
327 break
328 sleep(delay)
329 self.assertTrue(ran_ok, 'aktualizr-info failed: ' + stderr.decode() + stdout.decode())
330
331 verifyProvisioned(self, machine)
332
333
334class ImplProvTests(OESelftestTestCase):
335
336 def setUpLocal(self):
337 layer = "meta-updater-qemux86-64"
338 result = runCmd('bitbake-layers show-layers')
339 if re.search(layer, result.output) is None:
340 # Assume the directory layout for finding other layers. We could also
341 # make assumptions by using 'show-layers', but either way, if the
342 # layers we need aren't where we expect them, we are out of luck.
343 path = os.path.abspath(os.path.dirname(__file__))
344 metadir = path + "/../../../../../"
345 self.meta_qemu = metadir + layer
346 runCmd('bitbake-layers add-layer "%s"' % self.meta_qemu)
347 else:
348 self.meta_qemu = None
349 self.append_config('MACHINE = "qemux86-64"')
350 self.append_config('SOTA_CLIENT_PROV = " aktualizr-ca-implicit-prov "')
351 self.append_config('SOTA_DEPLOY_CREDENTIALS = "0"')
352 runCmd('bitbake -c cleanall aktualizr aktualizr-ca-implicit-prov')
353 self.qemu, self.s = qemu_launch(machine='qemux86-64')
354
355 def tearDownLocal(self):
356 qemu_terminate(self.s)
357 if self.meta_qemu:
358 runCmd('bitbake-layers remove-layer "%s"' % self.meta_qemu, ignore_status=True)
359
360 def qemu_command(self, command):
361 return qemu_send_command(self.qemu.ssh_port, command)
362
363 def test_provisioning(self):
364 print('Checking machine name (hostname) of device:')
365 stdout, stderr, retcode = self.qemu_command('hostname')
366 self.assertEqual(retcode, 0, "Unable to check hostname. " +
367 "Is an ssh daemon (such as dropbear or openssh) installed on the device?")
368 machine = get_bb_var('MACHINE', 'core-image-minimal')
369 self.assertEqual(stderr, b'', 'Error: ' + stderr.decode())
370 # Strip off line ending.
371 value = stdout.decode()[:-1]
372 self.assertEqual(value, machine,
373 'MACHINE does not match hostname: ' + machine + ', ' + value)
374 print(value)
375 print('Checking output of aktualizr-info:')
376 ran_ok = False
377 for delay in [1, 2, 5, 10, 15]:
378 stdout, stderr, retcode = self.qemu_command('aktualizr-info')
379 if retcode == 0 and stderr == b'':
380 ran_ok = True
381 break
382 sleep(delay)
383 self.assertTrue(ran_ok, 'aktualizr-info failed: ' + stderr.decode() + stdout.decode())
384 # Verify that device has NOT yet provisioned.
385 self.assertIn(b'Couldn\'t load device ID', stdout,
386 'Device already provisioned!? ' + stderr.decode() + stdout.decode())
387 self.assertIn(b'Couldn\'t load ECU serials', stdout,
388 'Device already provisioned!? ' + stderr.decode() + stdout.decode())
389 self.assertIn(b'Provisioned on server: no', stdout,
390 'Device already provisioned!? ' + stderr.decode() + stdout.decode())
391 self.assertIn(b'Fetched metadata: no', stdout,
392 'Device already provisioned!? ' + stderr.decode() + stdout.decode())
393
394 # Run aktualizr-cert-provider.
395 bb_vars = get_bb_vars(['SOTA_PACKED_CREDENTIALS'], 'aktualizr-native')
396 creds = bb_vars['SOTA_PACKED_CREDENTIALS']
397 bb_vars_prov = get_bb_vars(['STAGING_DIR_HOST', 'libdir'], 'aktualizr-ca-implicit-prov')
398 config = bb_vars_prov['STAGING_DIR_HOST'] + bb_vars_prov['libdir'] + '/sota/sota_implicit_prov_ca.toml'
399
400 print('Provisining at root@localhost:%d' % self.qemu.ssh_port)
401 akt_native_run(self, 'aktualizr-cert-provider -c {creds} -t root@localhost -p {port} -s -u -r -g {config}'
402 .format(creds=creds, port=self.qemu.ssh_port, config=config))
403
404 verifyProvisioned(self, machine)
405
406
407class HsmTests(OESelftestTestCase):
408
409 def setUpLocal(self):
410 layer = "meta-updater-qemux86-64"
411 result = runCmd('bitbake-layers show-layers')
412 if re.search(layer, result.output) is None:
413 # Assume the directory layout for finding other layers. We could also
414 # make assumptions by using 'show-layers', but either way, if the
415 # layers we need aren't where we expect them, we are out of luck.
416 path = os.path.abspath(os.path.dirname(__file__))
417 metadir = path + "/../../../../../"
418 self.meta_qemu = metadir + layer
419 runCmd('bitbake-layers add-layer "%s"' % self.meta_qemu)
420 else:
421 self.meta_qemu = None
422 self.append_config('MACHINE = "qemux86-64"')
423 self.append_config('SOTA_CLIENT_PROV = "aktualizr-hsm-prov"')
424 self.append_config('SOTA_DEPLOY_CREDENTIALS = "0"')
425 self.append_config('SOTA_CLIENT_FEATURES = "hsm"')
426 self.append_config('IMAGE_INSTALL_append = " softhsm-testtoken"')
427 runCmd('bitbake -c cleanall aktualizr aktualizr-hsm-prov')
428 self.qemu, self.s = qemu_launch(machine='qemux86-64')
429
430 def tearDownLocal(self):
431 qemu_terminate(self.s)
432 if self.meta_qemu:
433 runCmd('bitbake-layers remove-layer "%s"' % self.meta_qemu, ignore_status=True)
434
435 def qemu_command(self, command):
436 return qemu_send_command(self.qemu.ssh_port, command)
437
438 def test_provisioning(self):
439 print('Checking machine name (hostname) of device:')
440 stdout, stderr, retcode = self.qemu_command('hostname')
441 self.assertEqual(retcode, 0, "Unable to check hostname. " +
442 "Is an ssh daemon (such as dropbear or openssh) installed on the device?")
443 machine = get_bb_var('MACHINE', 'core-image-minimal')
444 self.assertEqual(stderr, b'', 'Error: ' + stderr.decode())
445 # Strip off line ending.
446 value = stdout.decode()[:-1]
447 self.assertEqual(value, machine,
448 'MACHINE does not match hostname: ' + machine + ', ' + value)
449 print(value)
450 print('Checking output of aktualizr-info:')
451 ran_ok = False
452 for delay in [1, 2, 5, 10, 15]:
453 stdout, stderr, retcode = self.qemu_command('aktualizr-info')
454 if retcode == 0 and stderr == b'':
455 ran_ok = True
456 break
457 sleep(delay)
458 self.assertTrue(ran_ok, 'aktualizr-info failed: ' + stderr.decode() + stdout.decode())
459 # Verify that device has NOT yet provisioned.
460 self.assertIn(b'Couldn\'t load device ID', stdout,
461 'Device already provisioned!? ' + stderr.decode() + stdout.decode())
462 self.assertIn(b'Couldn\'t load ECU serials', stdout,
463 'Device already provisioned!? ' + stderr.decode() + stdout.decode())
464 self.assertIn(b'Provisioned on server: no', stdout,
465 'Device already provisioned!? ' + stderr.decode() + stdout.decode())
466 self.assertIn(b'Fetched metadata: no', stdout,
467 'Device already provisioned!? ' + stderr.decode() + stdout.decode())
468
469 # Verify that HSM is not yet initialized.
470 pkcs11_command = 'pkcs11-tool --module=/usr/lib/softhsm/libsofthsm2.so -O'
471 stdout, stderr, retcode = self.qemu_command(pkcs11_command)
472 self.assertNotEqual(retcode, 0, 'pkcs11-tool succeeded before initialization: ' +
473 stdout.decode() + stderr.decode())
474 softhsm2_command = 'softhsm2-util --show-slots'
475 stdout, stderr, retcode = self.qemu_command(softhsm2_command)
476 self.assertNotEqual(retcode, 0, 'softhsm2-tool succeeded before initialization: ' +
477 stdout.decode() + stderr.decode())
478
479 # Run aktualizr-cert-provider.
480 bb_vars = get_bb_vars(['SOTA_PACKED_CREDENTIALS'], 'aktualizr-native')
481 creds = bb_vars['SOTA_PACKED_CREDENTIALS']
482 bb_vars_prov = get_bb_vars(['STAGING_DIR_HOST', 'libdir'], 'aktualizr-hsm-prov')
483 config = bb_vars_prov['STAGING_DIR_HOST'] + bb_vars_prov['libdir'] + '/sota/sota_hsm_prov.toml'
484
485 akt_native_run(self, 'aktualizr-cert-provider -c {creds} -t root@localhost -p {port} -r -s -u -g {config}'
486 .format(creds=creds, port=self.qemu.ssh_port, config=config))
487
488 # Verify that HSM is able to initialize.
489 ran_ok = False
490 for delay in [5, 5, 5, 5, 10]:
491 sleep(delay)
492 p11_out, p11_err, p11_ret = self.qemu_command(pkcs11_command)
493 hsm_out, hsm_err, hsm_ret = self.qemu_command(softhsm2_command)
494 if p11_ret == 0 and hsm_ret == 0 and hsm_err == b'':
495 ran_ok = True
496 break
497 self.assertTrue(ran_ok, 'pkcs11-tool or softhsm2-tool failed: ' + p11_err.decode() +
498 p11_out.decode() + hsm_err.decode() + hsm_out.decode())
499 self.assertIn(b'present token', p11_err, 'pkcs11-tool failed: ' + p11_err.decode() + p11_out.decode())
500 self.assertIn(b'X.509 cert', p11_out, 'pkcs11-tool failed: ' + p11_err.decode() + p11_out.decode())
501 self.assertIn(b'Initialized: yes', hsm_out, 'softhsm2-tool failed: ' +
502 hsm_err.decode() + hsm_out.decode())
503 self.assertIn(b'User PIN init.: yes', hsm_out, 'softhsm2-tool failed: ' +
504 hsm_err.decode() + hsm_out.decode())
505
506 # Check that pkcs11 output matches sofhsm output.
507 p11_p = re.compile(r'Using slot [0-9] with a present token \((0x[0-9a-f]*)\)\s')
508 p11_m = p11_p.search(p11_err.decode())
509 self.assertTrue(p11_m, 'Slot number not found with pkcs11-tool: ' + p11_err.decode() + p11_out.decode())
510 self.assertGreater(p11_m.lastindex, 0, 'Slot number not found with pkcs11-tool: ' +
511 p11_err.decode() + p11_out.decode())
512 hsm_p = re.compile(r'Description:\s*SoftHSM slot ID (0x[0-9a-f]*)\s')
513 hsm_m = hsm_p.search(hsm_out.decode())
514 self.assertTrue(hsm_m, 'Slot number not found with softhsm2-tool: ' + hsm_err.decode() + hsm_out.decode())
515 self.assertGreater(hsm_m.lastindex, 0, 'Slot number not found with softhsm2-tool: ' +
516 hsm_err.decode() + hsm_out.decode())
517 self.assertEqual(p11_m.group(1), hsm_m.group(1), 'Slot number does not match: ' +
518 p11_err.decode() + p11_out.decode() + hsm_err.decode() + hsm_out.decode())
519
520 verifyProvisioned(self, machine)
521
522
523class SecondaryTests(OESelftestTestCase):
524 @classmethod
525 def setUpClass(cls):
526 super(SecondaryTests, cls).setUpClass()
527 logger = logging.getLogger("selftest")
528 logger.info('Running bitbake to build secondary-image')
529 bitbake('secondary-image')
530
531 def setUpLocal(self):
532 layer = "meta-updater-qemux86-64"
533 result = runCmd('bitbake-layers show-layers')
534 if re.search(layer, result.output) is None:
535 # Assume the directory layout for finding other layers. We could also
536 # make assumptions by using 'show-layers', but either way, if the
537 # layers we need aren't where we expect them, we are out of luck.
538 path = os.path.abspath(os.path.dirname(__file__))
539 metadir = path + "/../../../../../"
540 self.meta_qemu = metadir + layer
541 runCmd('bitbake-layers add-layer "%s"' % self.meta_qemu)
542 else:
543 self.meta_qemu = None
544 self.append_config('MACHINE = "qemux86-64"')
545 self.append_config('SOTA_CLIENT_PROV = " aktualizr-auto-prov "')
546 self.qemu, self.s = qemu_launch(machine='qemux86-64', imagename='secondary-image')
547
548 def tearDownLocal(self):
549 qemu_terminate(self.s)
550 if self.meta_qemu:
551 runCmd('bitbake-layers remove-layer "%s"' % self.meta_qemu, ignore_status=True)
552
553 def qemu_command(self, command):
554 return qemu_send_command(self.qemu.ssh_port, command)
555
556 def test_secondary_present(self):
557 print('Checking aktualizr-secondary is present')
558 stdout, stderr, retcode = self.qemu_command('aktualizr-secondary --help')
559 self.assertEqual(retcode, 0, "Unable to run aktualizr-secondary --help")
560 self.assertEqual(stderr, b'', 'Error: ' + stderr.decode())
561
562 def test_secondary_listening(self):
563 print('Checking aktualizr-secondary service is listening')
564 stdout, stderr, retcode = self.qemu_command('aktualizr-check-discovery')
565 self.assertEqual(retcode, 0, "Unable to connect to secondary")
566
567
568class PrimaryTests(OESelftestTestCase):
569 @classmethod
570 def setUpClass(cls):
571 super(PrimaryTests, cls).setUpClass()
572 logger = logging.getLogger("selftest")
573 logger.info('Running bitbake to build primary-image')
574 bitbake('primary-image')
575
576 def setUpLocal(self):
577 layer = "meta-updater-qemux86-64"
578 result = runCmd('bitbake-layers show-layers')
579 if re.search(layer, result.output) is None:
580 # Assume the directory layout for finding other layers. We could also
581 # make assumptions by using 'show-layers', but either way, if the
582 # layers we need aren't where we expect them, we are out of luck.
583 path = os.path.abspath(os.path.dirname(__file__))
584 metadir = path + "/../../../../../"
585 self.meta_qemu = metadir + layer
586 runCmd('bitbake-layers add-layer "%s"' % self.meta_qemu)
587 else:
588 self.meta_qemu = None
589 self.append_config('MACHINE = "qemux86-64"')
590 self.append_config('SOTA_CLIENT_PROV = " aktualizr-auto-prov "')
591 self.append_config('SOTA_CLIENT_FEATURES = "secondary-network"')
592 self.qemu, self.s = qemu_launch(machine='qemux86-64', imagename='primary-image')
593
594 def tearDownLocal(self):
595 qemu_terminate(self.s)
596 if self.meta_qemu:
597 runCmd('bitbake-layers remove-layer "%s"' % self.meta_qemu, ignore_status=True)
598
599 def qemu_command(self, command):
600 return qemu_send_command(self.qemu.ssh_port, command)
601
602 def test_aktualizr_present(self):
603 print('Checking aktualizr is present')
604 stdout, stderr, retcode = self.qemu_command('aktualizr --help')
605 self.assertEqual(retcode, 0, "Unable to run aktualizr --help")
606 self.assertEqual(stderr, b'', 'Error: ' + stderr.decode())
607
608
609def qemu_launch(efi=False, machine=None, imagename=None):
610 logger = logging.getLogger("selftest")
611 logger.info('Running bitbake to build core-image-minimal')
612 bitbake('core-image-minimal')
613 # Create empty object.
614 args = type('', (), {})()
615 if imagename:
616 args.imagename = imagename
617 else:
618 args.imagename = 'core-image-minimal'
619 args.mac = None
620 # Could use DEPLOY_DIR_IMAGE here but it's already in the machine
621 # subdirectory.
622 args.dir = 'tmp/deploy/images'
623 args.efi = efi
624 args.machine = machine
625 qemu_use_kvm = get_bb_var("QEMU_USE_KVM")
626 if qemu_use_kvm and \
627 (qemu_use_kvm == 'True' and 'x86' in machine or \
628 get_bb_var('MACHINE') in qemu_use_kvm.split()):
629 args.kvm = True
630 else:
631 args.kvm = None # Autodetect
632 args.no_gui = True
633 args.gdb = False
634 args.pcap = None
635 args.overlay = None
636 args.dry_run = False
637 args.secondary_network = False
638
639 qemu = QemuCommand(args)
640 cmdline = qemu.command_line()
641 print('Booting image with run-qemu-ota...')
642 s = subprocess.Popen(cmdline)
643 sleep(10)
644 return qemu, s
645
646
647def qemu_terminate(s):
648 try:
649 s.terminate()
650 except KeyboardInterrupt:
651 pass
652
653
654def qemu_send_command(port, command):
655 command = ['ssh -q -o UserKnownHostsFile=/dev/null -o StrictHostKeyChecking=no root@localhost -p ' +
656 str(port) + ' "' + command + '"']
657 s2 = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
658 stdout, stderr = s2.communicate(timeout=60)
659 return stdout, stderr, s2.returncode
660
661
662def akt_native_run(testInst, cmd, **kwargs):
663 # run a command supplied by aktualizr-native and checks that:
664 # - the executable exists
665 # - the command runs without error
666 # NOTE: the base test class must have built aktualizr-native (in
667 # setUpClass, for example)
668 bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'base_prefix', 'libdir', 'bindir'],
669 'aktualizr-native')
670 sysroot = bb_vars['SYSROOT_DESTDIR'] + bb_vars['base_prefix']
671 sysrootbin = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir']
672 libdir = bb_vars['libdir']
673
674 program, *_ = cmd.split(' ')
675 p = '{}/{}'.format(sysrootbin, program)
676 testInst.assertTrue(os.path.isfile(p), msg="No {} found ({})".format(program, p))
677 env = dict(os.environ)
678 env['LD_LIBRARY_PATH'] = libdir
679 result = runCmd(cmd, env=env, native_sysroot=sysroot, ignore_status=True, **kwargs)
680 testInst.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output)
681
682
683def verifyProvisioned(testInst, machine):
684 # Verify that device HAS provisioned.
685 ran_ok = False
686 for delay in [5, 5, 5, 5, 10, 10, 10, 10]:
687 stdout, stderr, retcode = testInst.qemu_command('aktualizr-info')
688 if retcode == 0 and stderr == b'' and stdout.decode().find('Fetched metadata: yes') >= 0:
689 ran_ok = True
690 break
691 sleep(delay)
692 testInst.assertIn(b'Device ID: ', stdout, 'Provisioning failed: ' + stderr.decode() + stdout.decode())
693 testInst.assertIn(b'Primary ecu hardware ID: ' + machine.encode(), stdout,
694 'Provisioning failed: ' + stderr.decode() + stdout.decode())
695 testInst.assertIn(b'Fetched metadata: yes', stdout, 'Provisioning failed: ' + stderr.decode() + stdout.decode())
696 p = re.compile(r'Device ID: ([a-z0-9-]*)\n')
697 m = p.search(stdout.decode())
698 testInst.assertTrue(m, 'Device ID could not be read: ' + stderr.decode() + stdout.decode())
699 testInst.assertGreater(m.lastindex, 0, 'Device ID could not be read: ' + stderr.decode() + stdout.decode())
700 logger = logging.getLogger("selftest")
701 logger.info('Device successfully provisioned with ID: ' + m.group(1))
702
703
704# vim:set ts=4 sw=4 sts=4 expandtab: