diff options
Diffstat (limited to 'lib/oeqa/selftest/cases')
l--------- | lib/oeqa/selftest/cases/qemucommand.py | 1 | ||||
-rw-r--r-- | lib/oeqa/selftest/cases/updater.py | 475 |
2 files changed, 476 insertions, 0 deletions
diff --git a/lib/oeqa/selftest/cases/qemucommand.py b/lib/oeqa/selftest/cases/qemucommand.py new file mode 120000 index 0000000..075cdb8 --- /dev/null +++ b/lib/oeqa/selftest/cases/qemucommand.py | |||
@@ -0,0 +1 @@ | |||
../../../../scripts/qemucommand.py \ No newline at end of file | |||
diff --git a/lib/oeqa/selftest/cases/updater.py b/lib/oeqa/selftest/cases/updater.py new file mode 100644 index 0000000..b544762 --- /dev/null +++ b/lib/oeqa/selftest/cases/updater.py | |||
@@ -0,0 +1,475 @@ | |||
1 | # pylint: disable=C0111,C0325 | ||
2 | import os | ||
3 | import logging | ||
4 | import re | ||
5 | import subprocess | ||
6 | import unittest | ||
7 | from time import sleep | ||
8 | |||
9 | from oeqa.selftest.case import OESelftestTestCase | ||
10 | from oeqa.utils.commands import runCmd, bitbake, get_bb_var, get_bb_vars | ||
11 | from qemucommand import QemuCommand | ||
12 | |||
13 | |||
14 | class 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 | |||
33 | class 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_systemd(self): | ||
40 | result = get_bb_var('DISTRO_FEATURES').find('systemd') | ||
41 | self.assertNotEqual(result, -1, 'Feature "systemd" not set at DISTRO_FEATURES') | ||
42 | |||
43 | def test_credentials(self): | ||
44 | logger = logging.getLogger("selftest") | ||
45 | logger.info('Running bitbake to build core-image-minimal') | ||
46 | self.append_config('SOTA_CLIENT_PROV = "aktualizr-auto-prov"') | ||
47 | bitbake('core-image-minimal') | ||
48 | credentials = get_bb_var('SOTA_PACKED_CREDENTIALS') | ||
49 | # skip the test if the variable SOTA_PACKED_CREDENTIALS is not set | ||
50 | if credentials is None: | ||
51 | raise unittest.SkipTest("Variable 'SOTA_PACKED_CREDENTIALS' not set.") | ||
52 | # Check if the file exists | ||
53 | self.assertTrue(os.path.isfile(credentials), "File %s does not exist" % credentials) | ||
54 | deploydir = get_bb_var('DEPLOY_DIR_IMAGE') | ||
55 | imagename = get_bb_var('IMAGE_LINK_NAME', 'core-image-minimal') | ||
56 | # Check if the credentials are included in the output image | ||
57 | result = runCmd('tar -jtvf %s/%s.tar.bz2 | grep sota_provisioning_credentials.zip' % | ||
58 | (deploydir, imagename), ignore_status=True) | ||
59 | self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output) | ||
60 | |||
61 | def test_java(self): | ||
62 | result = runCmd('which java', ignore_status=True) | ||
63 | self.assertEqual(result.status, 0, | ||
64 | "Java not found. Do you have a JDK installed on your host machine?") | ||
65 | |||
66 | def test_add_package(self): | ||
67 | print('') | ||
68 | deploydir = get_bb_var('DEPLOY_DIR_IMAGE') | ||
69 | imagename = get_bb_var('IMAGE_LINK_NAME', 'core-image-minimal') | ||
70 | image_path = deploydir + '/' + imagename + '.otaimg' | ||
71 | logger = logging.getLogger("selftest") | ||
72 | |||
73 | logger.info('Running bitbake with man in the image package list') | ||
74 | self.append_config('IMAGE_INSTALL_append = " man "') | ||
75 | bitbake('-c cleanall man') | ||
76 | bitbake('core-image-minimal') | ||
77 | result = runCmd('oe-pkgdata-util find-path /usr/bin/man') | ||
78 | self.assertEqual(result.output, 'man: /usr/bin/man') | ||
79 | path1 = os.path.realpath(image_path) | ||
80 | size1 = os.path.getsize(path1) | ||
81 | logger.info('First image %s has size %i' % (path1, size1)) | ||
82 | |||
83 | logger.info('Running bitbake without man in the image package list') | ||
84 | self.append_config('IMAGE_INSTALL_remove = " man "') | ||
85 | bitbake('-c cleanall man') | ||
86 | bitbake('core-image-minimal') | ||
87 | result = runCmd('oe-pkgdata-util find-path /usr/bin/man', ignore_status=True) | ||
88 | self.assertEqual(result.status, 1, "Status different than 1. output: %s" % result.output) | ||
89 | self.assertEqual(result.output, 'ERROR: Unable to find any package producing path /usr/bin/man') | ||
90 | path2 = os.path.realpath(image_path) | ||
91 | size2 = os.path.getsize(path2) | ||
92 | logger.info('Second image %s has size %i', path2, size2) | ||
93 | self.assertNotEqual(path1, path2, "Image paths are identical; image was not rebuilt.") | ||
94 | self.assertNotEqual(size1, size2, "Image sizes are identical; image was not rebuilt.") | ||
95 | |||
96 | |||
97 | class AktualizrToolsTests(OESelftestTestCase): | ||
98 | |||
99 | @classmethod | ||
100 | def setUpClass(cls): | ||
101 | super(AktualizrToolsTests, cls).setUpClass() | ||
102 | logger = logging.getLogger("selftest") | ||
103 | logger.info('Running bitbake to build aktualizr-native tools') | ||
104 | bitbake('aktualizr-native') | ||
105 | |||
106 | def test_implicit_writer_help(self): | ||
107 | akt_native_run(self, 'aktualizr_implicit_writer --help') | ||
108 | |||
109 | def test_cert_provider_help(self): | ||
110 | akt_native_run(self, 'aktualizr_cert_provider --help') | ||
111 | |||
112 | def test_cert_provider_local_output(self): | ||
113 | logger = logging.getLogger("selftest") | ||
114 | logger.info('Running bitbake to build aktualizr-implicit-prov') | ||
115 | bitbake('aktualizr-implicit-prov') | ||
116 | bb_vars = get_bb_vars(['SOTA_PACKED_CREDENTIALS', 'T'], 'aktualizr-native') | ||
117 | creds = bb_vars['SOTA_PACKED_CREDENTIALS'] | ||
118 | temp_dir = bb_vars['T'] | ||
119 | bb_vars_prov = get_bb_vars(['STAGING_DIR_NATIVE', 'libdir'], 'aktualizr-implicit-prov') | ||
120 | config = bb_vars_prov['STAGING_DIR_NATIVE'] + bb_vars_prov['libdir'] + '/sota/sota_implicit_prov.toml' | ||
121 | |||
122 | akt_native_run(self, 'aktualizr_cert_provider -c {creds} -r -l {temp} -g {config}' | ||
123 | .format(creds=creds, temp=temp_dir, config=config)) | ||
124 | |||
125 | # Might be nice if these names weren't hardcoded. | ||
126 | cert_path = temp_dir + '/client.pem' | ||
127 | self.assertTrue(os.path.isfile(cert_path), "Client certificate not found at %s." % cert_path) | ||
128 | self.assertTrue(os.path.getsize(cert_path) > 0, "Client certificate at %s is empty." % cert_path) | ||
129 | pkey_path = temp_dir + '/pkey.pem' | ||
130 | self.assertTrue(os.path.isfile(pkey_path), "Private key not found at %s." % pkey_path) | ||
131 | self.assertTrue(os.path.getsize(pkey_path) > 0, "Private key at %s is empty." % pkey_path) | ||
132 | ca_path = temp_dir + '/root.crt' | ||
133 | self.assertTrue(os.path.isfile(ca_path), "Client certificate not found at %s." % ca_path) | ||
134 | self.assertTrue(os.path.getsize(ca_path) > 0, "Client certificate at %s is empty." % ca_path) | ||
135 | |||
136 | |||
137 | class QemuTests(OESelftestTestCase): | ||
138 | |||
139 | @classmethod | ||
140 | def setUpClass(cls): | ||
141 | super(QemuTests, cls).setUpClass() | ||
142 | cls.qemu, cls.s = qemu_launch(machine='qemux86-64') | ||
143 | |||
144 | @classmethod | ||
145 | def tearDownClass(cls): | ||
146 | qemu_terminate(cls.s) | ||
147 | super(QemuTests, cls).tearDownClass() | ||
148 | |||
149 | def qemu_command(self, command): | ||
150 | return qemu_send_command(self.qemu.ssh_port, command) | ||
151 | |||
152 | def test_qemu(self): | ||
153 | print('Checking machine name (hostname) of device:') | ||
154 | stdout, stderr, retcode = self.qemu_command('hostname') | ||
155 | self.assertEqual(retcode, 0, "Unable to check hostname. " + | ||
156 | "Is an ssh daemon (such as dropbear or openssh) installed on the device?") | ||
157 | machine = get_bb_var('MACHINE', 'core-image-minimal') | ||
158 | self.assertEqual(stderr, b'', 'Error: ' + stderr.decode()) | ||
159 | # Strip off line ending. | ||
160 | value_str = stdout.decode()[:-1] | ||
161 | self.assertEqual(value_str, machine, | ||
162 | 'MACHINE does not match hostname: ' + machine + ', ' + value_str) | ||
163 | print(value_str) | ||
164 | print('Checking output of aktualizr-info:') | ||
165 | ran_ok = False | ||
166 | for delay in [0, 1, 2, 5, 10, 15]: | ||
167 | sleep(delay) | ||
168 | stdout, stderr, retcode = self.qemu_command('aktualizr-info') | ||
169 | if retcode == 0 and stderr == b'': | ||
170 | ran_ok = True | ||
171 | break | ||
172 | self.assertTrue(ran_ok, 'aktualizr-info failed: ' + stderr.decode() + stdout.decode()) | ||
173 | |||
174 | |||
175 | class GrubTests(OESelftestTestCase): | ||
176 | |||
177 | def setUpLocal(self): | ||
178 | # This is a bit of a hack but I can't see a better option. | ||
179 | path = os.path.abspath(os.path.dirname(__file__)) | ||
180 | metadir = path + "/../../../../../" | ||
181 | grub_config = 'OSTREE_BOOTLOADER = "grub"\nMACHINE = "intel-corei7-64"' | ||
182 | self.append_config(grub_config) | ||
183 | self.meta_intel = metadir + "meta-intel" | ||
184 | self.meta_minnow = metadir + "meta-updater-minnowboard" | ||
185 | runCmd('bitbake-layers add-layer "%s"' % self.meta_intel) | ||
186 | runCmd('bitbake-layers add-layer "%s"' % self.meta_minnow) | ||
187 | self.qemu, self.s = qemu_launch(efi=True, machine='intel-corei7-64') | ||
188 | |||
189 | def tearDownLocal(self): | ||
190 | qemu_terminate(self.s) | ||
191 | runCmd('bitbake-layers remove-layer "%s"' % self.meta_intel, ignore_status=True) | ||
192 | runCmd('bitbake-layers remove-layer "%s"' % self.meta_minnow, ignore_status=True) | ||
193 | |||
194 | def qemu_command(self, command): | ||
195 | return qemu_send_command(self.qemu.ssh_port, command) | ||
196 | |||
197 | def test_grub(self): | ||
198 | print('') | ||
199 | print('Checking machine name (hostname) of device:') | ||
200 | stdout, stderr, retcode = self.qemu_command('hostname') | ||
201 | self.assertEqual(retcode, 0, "Unable to check hostname. " + | ||
202 | "Is an ssh daemon (such as dropbear or openssh) installed on the device?") | ||
203 | machine = get_bb_var('MACHINE', 'core-image-minimal') | ||
204 | self.assertEqual(stderr, b'', 'Error: ' + stderr.decode()) | ||
205 | # Strip off line ending. | ||
206 | value = stdout.decode()[:-1] | ||
207 | self.assertEqual(value, machine, | ||
208 | 'MACHINE does not match hostname: ' + machine + ', ' + value + | ||
209 | '\nIs TianoCore ovmf installed on your host machine?') | ||
210 | print(value) | ||
211 | print('Checking output of aktualizr-info:') | ||
212 | ran_ok = False | ||
213 | for delay in [0, 1, 2, 5, 10, 15]: | ||
214 | sleep(delay) | ||
215 | stdout, stderr, retcode = self.qemu_command('aktualizr-info') | ||
216 | if retcode == 0 and stderr == b'': | ||
217 | ran_ok = True | ||
218 | break | ||
219 | self.assertTrue(ran_ok, 'aktualizr-info failed: ' + stderr.decode() + stdout.decode()) | ||
220 | |||
221 | |||
222 | class ImplProvTests(OESelftestTestCase): | ||
223 | |||
224 | def setUpLocal(self): | ||
225 | self.append_config('SOTA_CLIENT_PROV = " aktualizr-implicit-prov "') | ||
226 | # note: this will build aktualizr-native as a side-effect | ||
227 | self.qemu, self.s = qemu_launch(machine='qemux86-64') | ||
228 | |||
229 | def tearDownLocal(self): | ||
230 | qemu_terminate(self.s) | ||
231 | |||
232 | def qemu_command(self, command): | ||
233 | return qemu_send_command(self.qemu.ssh_port, command) | ||
234 | |||
235 | def test_provisioning(self): | ||
236 | print('Checking machine name (hostname) of device:') | ||
237 | stdout, stderr, retcode = self.qemu_command('hostname') | ||
238 | self.assertEqual(retcode, 0, "Unable to check hostname. " + | ||
239 | "Is an ssh daemon (such as dropbear or openssh) installed on the device?") | ||
240 | machine = get_bb_var('MACHINE', 'core-image-minimal') | ||
241 | self.assertEqual(stderr, b'', 'Error: ' + stderr.decode()) | ||
242 | # Strip off line ending. | ||
243 | value_str = stdout.decode()[:-1] | ||
244 | self.assertEqual(value_str, machine, | ||
245 | 'MACHINE does not match hostname: ' + machine + ', ' + value_str) | ||
246 | print(value_str) | ||
247 | print('Checking output of aktualizr-info:') | ||
248 | ran_ok = False | ||
249 | for delay in [0, 1, 2, 5, 10, 15]: | ||
250 | stdout, stderr, retcode = self.qemu_command('aktualizr-info') | ||
251 | if retcode == 0 and stderr == b'': | ||
252 | ran_ok = True | ||
253 | break | ||
254 | self.assertTrue(ran_ok, 'aktualizr-info failed: ' + stderr.decode() + stdout.decode()) | ||
255 | # Verify that device has NOT yet provisioned. | ||
256 | self.assertIn(b'Couldn\'t load device ID', stdout, | ||
257 | 'Device already provisioned!? ' + stderr.decode() + stdout.decode()) | ||
258 | self.assertIn(b'Couldn\'t load ECU serials', stdout, | ||
259 | 'Device already provisioned!? ' + stderr.decode() + stdout.decode()) | ||
260 | self.assertIn(b'Provisioned on server: no', stdout, | ||
261 | 'Device already provisioned!? ' + stderr.decode() + stdout.decode()) | ||
262 | self.assertIn(b'Fetched metadata: no', stdout, | ||
263 | 'Device already provisioned!? ' + stderr.decode() + stdout.decode()) | ||
264 | |||
265 | # Run cert_provider. | ||
266 | bb_vars = get_bb_vars(['SOTA_PACKED_CREDENTIALS'], 'aktualizr-native') | ||
267 | creds = bb_vars['SOTA_PACKED_CREDENTIALS'] | ||
268 | bb_vars_prov = get_bb_vars(['STAGING_DIR_NATIVE', 'libdir'], 'aktualizr-implicit-prov') | ||
269 | config = bb_vars_prov['STAGING_DIR_NATIVE'] + bb_vars_prov['libdir'] + '/sota/sota_implicit_prov.toml' | ||
270 | |||
271 | akt_native_run(self, 'aktualizr_cert_provider -c {creds} -t root@localhost -p {port} -s -g {config}' | ||
272 | .format(creds=creds, port=self.qemu.ssh_port, config=config)) | ||
273 | |||
274 | # Verify that device HAS provisioned. | ||
275 | ran_ok = False | ||
276 | for delay in [5, 5, 5, 5, 10]: | ||
277 | sleep(delay) | ||
278 | stdout, stderr, retcode = self.qemu_command('aktualizr-info') | ||
279 | if retcode == 0 and stderr == b'' and stdout.decode().find('Fetched metadata: yes') >= 0: | ||
280 | ran_ok = True | ||
281 | break | ||
282 | self.assertIn(b'Device ID: ', stdout, 'Provisioning failed: ' + stderr.decode() + stdout.decode()) | ||
283 | self.assertIn(b'Primary ecu hardware ID: qemux86-64', stdout, | ||
284 | 'Provisioning failed: ' + stderr.decode() + stdout.decode()) | ||
285 | self.assertIn(b'Fetched metadata: yes', stdout, 'Provisioning failed: ' + stderr.decode() + stdout.decode()) | ||
286 | p = re.compile(r'Device ID: ([a-z0-9-]*)\n') | ||
287 | m = p.search(stdout.decode()) | ||
288 | self.assertTrue(m, 'Device ID could not be read: ' + stderr.decode() + stdout.decode()) | ||
289 | self.assertGreater(m.lastindex, 0, 'Device ID could not be read: ' + stderr.decode() + stdout.decode()) | ||
290 | logger = logging.getLogger("selftest") | ||
291 | logger.info('Device successfully provisioned with ID: ' + m.group(1)) | ||
292 | |||
293 | |||
294 | class HsmTests(OESelftestTestCase): | ||
295 | |||
296 | def setUpLocal(self): | ||
297 | self.append_config('SOTA_CLIENT_PROV = "aktualizr-hsm-prov"') | ||
298 | self.append_config('SOTA_CLIENT_FEATURES = "hsm"') | ||
299 | # note: this will build aktualizr-native as a side-effect | ||
300 | self.qemu, self.s = qemu_launch(machine='qemux86-64') | ||
301 | |||
302 | def tearDownLocal(self): | ||
303 | qemu_terminate(self.s) | ||
304 | |||
305 | def qemu_command(self, command): | ||
306 | return qemu_send_command(self.qemu.ssh_port, command) | ||
307 | |||
308 | def test_provisioning(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_str = stdout.decode()[:-1] | ||
317 | self.assertEqual(value_str, machine, | ||
318 | 'MACHINE does not match hostname: ' + machine + ', ' + value_str + | ||
319 | '\nIs tianocore ovmf installed?') | ||
320 | print(value_str) | ||
321 | print('Checking output of aktualizr-info:') | ||
322 | ran_ok = False | ||
323 | for delay in [0, 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 | self.assertTrue(ran_ok, 'aktualizr-info failed: ' + stderr.decode() + stdout.decode()) | ||
329 | # Verify that device has NOT yet provisioned. | ||
330 | self.assertIn(b'Couldn\'t load device ID', stdout, | ||
331 | 'Device already provisioned!? ' + stderr.decode() + stdout.decode()) | ||
332 | self.assertIn(b'Couldn\'t load ECU serials', stdout, | ||
333 | 'Device already provisioned!? ' + stderr.decode() + stdout.decode()) | ||
334 | self.assertIn(b'Provisioned on server: no', stdout, | ||
335 | 'Device already provisioned!? ' + stderr.decode() + stdout.decode()) | ||
336 | self.assertIn(b'Fetched metadata: no', stdout, | ||
337 | 'Device already provisioned!? ' + stderr.decode() + stdout.decode()) | ||
338 | |||
339 | # Verify that HSM is not yet initialized. | ||
340 | pkcs11_command = 'pkcs11-tool --module=/usr/lib/softhsm/libsofthsm2.so -O' | ||
341 | stdout, stderr, retcode = self.qemu_command(pkcs11_command) | ||
342 | self.assertNotEqual(retcode, 0, 'pkcs11-tool succeeded before initialization: ' + | ||
343 | stdout.decode() + stderr.decode()) | ||
344 | softhsm2_command = 'softhsm2-util --show-slots' | ||
345 | stdout, stderr, retcode = self.qemu_command(softhsm2_command) | ||
346 | self.assertNotEqual(retcode, 0, 'softhsm2-tool succeeded before initialization: ' + | ||
347 | stdout.decode() + stderr.decode()) | ||
348 | |||
349 | # Run cert_provider. | ||
350 | bb_vars = get_bb_vars(['SOTA_PACKED_CREDENTIALS'], 'aktualizr-native') | ||
351 | creds = bb_vars['SOTA_PACKED_CREDENTIALS'] | ||
352 | bb_vars_prov = get_bb_vars(['STAGING_DIR_NATIVE', 'libdir'], 'aktualizr-hsm-prov') | ||
353 | config = bb_vars_prov['STAGING_DIR_NATIVE'] + bb_vars_prov['libdir'] + '/sota/sota_hsm_prov.toml' | ||
354 | |||
355 | akt_native_run(self, 'aktualizr_cert_provider -c {creds} -t root@localhost -p {port} -r -s -g {config}' | ||
356 | .format(creds=creds, port=self.qemu.ssh_port, config=config)) | ||
357 | |||
358 | # Verify that HSM is able to initialize. | ||
359 | ran_ok = False | ||
360 | for delay in [5, 5, 5, 5, 10]: | ||
361 | sleep(delay) | ||
362 | p11_out, p11_err, p11_ret = self.qemu_command(pkcs11_command) | ||
363 | hsm_out, hsm_err, hsm_ret = self.qemu_command(softhsm2_command) | ||
364 | if p11_ret == 0 and hsm_ret == 0 and hsm_err == b'': | ||
365 | ran_ok = True | ||
366 | break | ||
367 | self.assertTrue(ran_ok, 'pkcs11-tool or softhsm2-tool failed: ' + p11_err.decode() + | ||
368 | p11_out.decode() + hsm_err.decode() + hsm_out.decode()) | ||
369 | self.assertIn(b'present token', p11_err, 'pkcs11-tool failed: ' + p11_err.decode() + p11_out.decode()) | ||
370 | self.assertIn(b'X.509 cert', p11_out, 'pkcs11-tool failed: ' + p11_err.decode() + p11_out.decode()) | ||
371 | self.assertIn(b'Initialized: yes', hsm_out, 'softhsm2-tool failed: ' + | ||
372 | hsm_err.decode() + hsm_out.decode()) | ||
373 | self.assertIn(b'User PIN init.: yes', hsm_out, 'softhsm2-tool failed: ' + | ||
374 | hsm_err.decode() + hsm_out.decode()) | ||
375 | |||
376 | # Check that pkcs11 output matches sofhsm output. | ||
377 | p11_p = re.compile(r'Using slot [0-9] with a present token \((0x[0-9a-f]*)\)\s') | ||
378 | p11_m = p11_p.search(p11_err.decode()) | ||
379 | self.assertTrue(p11_m, 'Slot number not found with pkcs11-tool: ' + p11_err.decode() + p11_out.decode()) | ||
380 | self.assertGreater(p11_m.lastindex, 0, 'Slot number not found with pkcs11-tool: ' + | ||
381 | p11_err.decode() + p11_out.decode()) | ||
382 | hsm_p = re.compile(r'Description:\s*SoftHSM slot ID (0x[0-9a-f]*)\s') | ||
383 | hsm_m = hsm_p.search(hsm_out.decode()) | ||
384 | self.assertTrue(hsm_m, 'Slot number not found with softhsm2-tool: ' + hsm_err.decode() + hsm_out.decode()) | ||
385 | self.assertGreater(hsm_m.lastindex, 0, 'Slot number not found with softhsm2-tool: ' + | ||
386 | hsm_err.decode() + hsm_out.decode()) | ||
387 | self.assertEqual(p11_m.group(1), hsm_m.group(1), 'Slot number does not match: ' + | ||
388 | p11_err.decode() + p11_out.decode() + hsm_err.decode() + hsm_out.decode()) | ||
389 | |||
390 | # Verify that device HAS provisioned. | ||
391 | ran_ok = False | ||
392 | for delay in [5, 5, 5, 5, 10]: | ||
393 | sleep(delay) | ||
394 | stdout, stderr, retcode = self.qemu_command('aktualizr-info') | ||
395 | if retcode == 0 and stderr == b'' and stdout.decode().find('Fetched metadata: yes') >= 0: | ||
396 | ran_ok = True | ||
397 | break | ||
398 | self.assertIn(b'Device ID: ', stdout, 'Provisioning failed: ' + stderr.decode() + stdout.decode()) | ||
399 | self.assertIn(b'Primary ecu hardware ID: qemux86-64', stdout, | ||
400 | 'Provisioning failed: ' + stderr.decode() + stdout.decode()) | ||
401 | self.assertIn(b'Fetched metadata: yes', stdout, 'Provisioning failed: ' + stderr.decode() + stdout.decode()) | ||
402 | p = re.compile(r'Device ID: ([a-z0-9-]*)\n') | ||
403 | m = p.search(stdout.decode()) | ||
404 | self.assertTrue(m, 'Device ID could not be read: ' + stderr.decode() + stdout.decode()) | ||
405 | self.assertGreater(m.lastindex, 0, 'Device ID could not be read: ' + stderr.decode() + stdout.decode()) | ||
406 | logger = logging.getLogger("selftest") | ||
407 | logger.info('Device successfully provisioned with ID: ' + m.group(1)) | ||
408 | |||
409 | |||
410 | def qemu_launch(efi=False, machine=None): | ||
411 | logger = logging.getLogger("selftest") | ||
412 | logger.info('Running bitbake to build core-image-minimal') | ||
413 | bitbake('core-image-minimal') | ||
414 | # Create empty object. | ||
415 | args = type('', (), {})() | ||
416 | args.imagename = 'core-image-minimal' | ||
417 | args.mac = None | ||
418 | # Could use DEPLOY_DIR_IMAGE here but it's already in the machine | ||
419 | # subdirectory. | ||
420 | args.dir = 'tmp/deploy/images' | ||
421 | args.efi = efi | ||
422 | args.machine = machine | ||
423 | args.kvm = None # Autodetect | ||
424 | args.no_gui = True | ||
425 | args.gdb = False | ||
426 | args.pcap = None | ||
427 | args.overlay = None | ||
428 | args.dry_run = False | ||
429 | |||
430 | qemu = QemuCommand(args) | ||
431 | cmdline = qemu.command_line() | ||
432 | print('Booting image with run-qemu-ota...') | ||
433 | s = subprocess.Popen(cmdline) | ||
434 | sleep(10) | ||
435 | return qemu, s | ||
436 | |||
437 | |||
438 | def qemu_terminate(s): | ||
439 | try: | ||
440 | s.terminate() | ||
441 | except KeyboardInterrupt: | ||
442 | pass | ||
443 | |||
444 | |||
445 | def qemu_send_command(port, command): | ||
446 | command = ['ssh -q -o UserKnownHostsFile=/dev/null -o StrictHostKeyChecking=no root@localhost -p ' + | ||
447 | str(port) + ' "' + command + '"'] | ||
448 | s2 = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) | ||
449 | stdout, stderr = s2.communicate() | ||
450 | return stdout, stderr, s2.returncode | ||
451 | |||
452 | |||
453 | def akt_native_run(testInst, cmd, **kwargs): | ||
454 | # run a command supplied by aktualizr-native and checks that: | ||
455 | # - the executable exists | ||
456 | # - the command runs without error | ||
457 | # NOTE: the base test class must have built aktualizr-native (in | ||
458 | # setUpClass, for example) | ||
459 | bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'base_prefix', 'libdir', 'bindir'], | ||
460 | 'aktualizr-native') | ||
461 | sysroot = bb_vars['SYSROOT_DESTDIR'] + bb_vars['base_prefix'] | ||
462 | sysrootbin = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir'] | ||
463 | libdir = bb_vars['libdir'] | ||
464 | |||
465 | program, *_ = cmd.split(' ') | ||
466 | p = '{}/{}'.format(sysrootbin, program) | ||
467 | testInst.assertTrue(os.path.isfile(p), msg="No {} found ({})".format(program, p)) | ||
468 | env = dict(os.environ) | ||
469 | env['LD_LIBRARY_PATH'] = libdir | ||
470 | result = runCmd(cmd, env=env, native_sysroot=sysroot, ignore_status=True, **kwargs) | ||
471 | testInst.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output) | ||
472 | |||
473 | |||
474 | |||
475 | # vim:set ts=4 sw=4 sts=4 expandtab: | ||