summaryrefslogtreecommitdiffstats
path: root/lib
diff options
context:
space:
mode:
Diffstat (limited to 'lib')
-rw-r--r--lib/oeqa/selftest/cases/updater.py345
1 files changed, 292 insertions, 53 deletions
diff --git a/lib/oeqa/selftest/cases/updater.py b/lib/oeqa/selftest/cases/updater.py
index 91ac9fc..b544762 100644
--- a/lib/oeqa/selftest/cases/updater.py
+++ b/lib/oeqa/selftest/cases/updater.py
@@ -1,6 +1,7 @@
1# pylint: disable=C0111,C0325 1# pylint: disable=C0111,C0325
2import os 2import os
3import logging 3import logging
4import re
4import subprocess 5import subprocess
5import unittest 6import unittest
6from time import sleep 7from time import sleep
@@ -20,32 +21,13 @@ class SotaToolsTests(OESelftestTestCase):
20 bitbake('aktualizr-native') 21 bitbake('aktualizr-native')
21 22
22 def test_push_help(self): 23 def test_push_help(self):
23 bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'bindir'], 'aktualizr-native') 24 akt_native_run(self, 'garage-push --help')
24 p = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir'] + "/" + "garage-push"
25 self.assertTrue(os.path.isfile(p), msg = "No garage-push found (%s)" % p)
26 result = runCmd('%s --help' % p, ignore_status=True)
27 self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output)
28 25
29 def test_deploy_help(self): 26 def test_deploy_help(self):
30 bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'bindir'], 'aktualizr-native') 27 akt_native_run(self, 'garage-deploy --help')
31 p = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir'] + "/" + "garage-deploy"
32 self.assertTrue(os.path.isfile(p), msg = "No garage-deploy found (%s)" % p)
33 result = runCmd('%s --help' % p, ignore_status=True)
34 self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output)
35 28
36 def test_garagesign_help(self): 29 def test_garagesign_help(self):
37 bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'bindir'], 'aktualizr-native') 30 akt_native_run(self, 'garage-sign --help')
38 p = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir'] + "/" + "garage-sign"
39 self.assertTrue(os.path.isfile(p), msg = "No garage-sign found (%s)" % p)
40 result = runCmd('%s --help' % p, ignore_status=True)
41 self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output)
42
43
44class HsmTests(OESelftestTestCase):
45
46 def test_hsm(self):
47 self.write_config('SOTA_CLIENT_FEATURES="hsm"')
48 bitbake('core-image-minimal')
49 31
50 32
51class GeneralTests(OESelftestTestCase): 33class GeneralTests(OESelftestTestCase):
@@ -59,6 +41,9 @@ class GeneralTests(OESelftestTestCase):
59 self.assertNotEqual(result, -1, 'Feature "systemd" not set at DISTRO_FEATURES') 41 self.assertNotEqual(result, -1, 'Feature "systemd" not set at DISTRO_FEATURES')
60 42
61 def test_credentials(self): 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"')
62 bitbake('core-image-minimal') 47 bitbake('core-image-minimal')
63 credentials = get_bb_var('SOTA_PACKED_CREDENTIALS') 48 credentials = get_bb_var('SOTA_PACKED_CREDENTIALS')
64 # skip the test if the variable SOTA_PACKED_CREDENTIALS is not set 49 # skip the test if the variable SOTA_PACKED_CREDENTIALS is not set
@@ -75,7 +60,8 @@ class GeneralTests(OESelftestTestCase):
75 60
76 def test_java(self): 61 def test_java(self):
77 result = runCmd('which java', ignore_status=True) 62 result = runCmd('which java', ignore_status=True)
78 self.assertEqual(result.status, 0, "Java not found.") 63 self.assertEqual(result.status, 0,
64 "Java not found. Do you have a JDK installed on your host machine?")
79 65
80 def test_add_package(self): 66 def test_add_package(self):
81 print('') 67 print('')
@@ -85,7 +71,7 @@ class GeneralTests(OESelftestTestCase):
85 logger = logging.getLogger("selftest") 71 logger = logging.getLogger("selftest")
86 72
87 logger.info('Running bitbake with man in the image package list') 73 logger.info('Running bitbake with man in the image package list')
88 self.write_config('IMAGE_INSTALL_append = " man "') 74 self.append_config('IMAGE_INSTALL_append = " man "')
89 bitbake('-c cleanall man') 75 bitbake('-c cleanall man')
90 bitbake('core-image-minimal') 76 bitbake('core-image-minimal')
91 result = runCmd('oe-pkgdata-util find-path /usr/bin/man') 77 result = runCmd('oe-pkgdata-util find-path /usr/bin/man')
@@ -95,7 +81,7 @@ class GeneralTests(OESelftestTestCase):
95 logger.info('First image %s has size %i' % (path1, size1)) 81 logger.info('First image %s has size %i' % (path1, size1))
96 82
97 logger.info('Running bitbake without man in the image package list') 83 logger.info('Running bitbake without man in the image package list')
98 self.write_config('IMAGE_INSTALL_remove = " man "') 84 self.append_config('IMAGE_INSTALL_remove = " man "')
99 bitbake('-c cleanall man') 85 bitbake('-c cleanall man')
100 bitbake('core-image-minimal') 86 bitbake('core-image-minimal')
101 result = runCmd('oe-pkgdata-util find-path /usr/bin/man', ignore_status=True) 87 result = runCmd('oe-pkgdata-util find-path /usr/bin/man', ignore_status=True)
@@ -108,6 +94,46 @@ class GeneralTests(OESelftestTestCase):
108 self.assertNotEqual(size1, size2, "Image sizes are identical; image was not rebuilt.") 94 self.assertNotEqual(size1, size2, "Image sizes are identical; image was not rebuilt.")
109 95
110 96
97class 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
111class QemuTests(OESelftestTestCase): 137class QemuTests(OESelftestTestCase):
112 138
113 @classmethod 139 @classmethod
@@ -118,14 +144,16 @@ class QemuTests(OESelftestTestCase):
118 @classmethod 144 @classmethod
119 def tearDownClass(cls): 145 def tearDownClass(cls):
120 qemu_terminate(cls.s) 146 qemu_terminate(cls.s)
147 super(QemuTests, cls).tearDownClass()
121 148
122 def run_command(self, command): 149 def qemu_command(self, command):
123 return qemu_send_command(self.qemu.ssh_port, command) 150 return qemu_send_command(self.qemu.ssh_port, command)
124 151
125 def test_hostname(self): 152 def test_qemu(self):
126 print('')
127 print('Checking machine name (hostname) of device:') 153 print('Checking machine name (hostname) of device:')
128 stdout, stderr, retcode = self.run_command('hostname') 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?")
129 machine = get_bb_var('MACHINE', 'core-image-minimal') 157 machine = get_bb_var('MACHINE', 'core-image-minimal')
130 self.assertEqual(stderr, b'', 'Error: ' + stderr.decode()) 158 self.assertEqual(stderr, b'', 'Error: ' + stderr.decode())
131 # Strip off line ending. 159 # Strip off line ending.
@@ -133,30 +161,15 @@ class QemuTests(OESelftestTestCase):
133 self.assertEqual(value_str, machine, 161 self.assertEqual(value_str, machine,
134 'MACHINE does not match hostname: ' + machine + ', ' + value_str) 162 'MACHINE does not match hostname: ' + machine + ', ' + value_str)
135 print(value_str) 163 print(value_str)
136
137 def test_var_sota(self):
138 print('')
139 print('Checking contents of /var/sota:')
140 stdout, stderr, retcode = self.run_command('ls /var/sota')
141 self.assertEqual(stderr, b'', 'Error: ' + stderr.decode())
142 self.assertEqual(retcode, 0)
143 print(stdout.decode())
144
145 def test_aktualizr_info(self):
146 print('Checking output of aktualizr-info:') 164 print('Checking output of aktualizr-info:')
147 ran_ok = False 165 ran_ok = False
148 for delay in [0, 1, 2, 5, 10, 15]: 166 for delay in [0, 1, 2, 5, 10, 15]:
149 sleep(delay) 167 sleep(delay)
150 try: 168 stdout, stderr, retcode = self.qemu_command('aktualizr-info')
151 stdout, stderr, retcode = self.run_command('aktualizr-info') 169 if retcode == 0 and stderr == b'':
152 if retcode == 0 and stderr == b'': 170 ran_ok = True
153 ran_ok = True 171 break
154 break 172 self.assertTrue(ran_ok, 'aktualizr-info failed: ' + stderr.decode() + stdout.decode())
155 except IOError as e:
156 print(e)
157 if not ran_ok:
158 print(stdout.decode())
159 print(stderr.decode())
160 173
161 174
162class GrubTests(OESelftestTestCase): 175class GrubTests(OESelftestTestCase):
@@ -178,19 +191,220 @@ class GrubTests(OESelftestTestCase):
178 runCmd('bitbake-layers remove-layer "%s"' % self.meta_intel, ignore_status=True) 191 runCmd('bitbake-layers remove-layer "%s"' % self.meta_intel, ignore_status=True)
179 runCmd('bitbake-layers remove-layer "%s"' % self.meta_minnow, ignore_status=True) 192 runCmd('bitbake-layers remove-layer "%s"' % self.meta_minnow, ignore_status=True)
180 193
194 def qemu_command(self, command):
195 return qemu_send_command(self.qemu.ssh_port, command)
196
181 def test_grub(self): 197 def test_grub(self):
182 print('') 198 print('')
183 print('Checking machine name (hostname) of device:') 199 print('Checking machine name (hostname) of device:')
184 value, err, retcode = qemu_send_command(self.qemu.ssh_port, 'hostname') 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
222class 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
294class 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?")
185 machine = get_bb_var('MACHINE', 'core-image-minimal') 313 machine = get_bb_var('MACHINE', 'core-image-minimal')
186 self.assertEqual(err, b'', 'Error: ' + err.decode()) 314 self.assertEqual(stderr, b'', 'Error: ' + stderr.decode())
187 self.assertEqual(retcode, 0)
188 # Strip off line ending. 315 # Strip off line ending.
189 value_str = value.decode()[:-1] 316 value_str = stdout.decode()[:-1]
190 self.assertEqual(value_str, machine, 317 self.assertEqual(value_str, machine,
191 'MACHINE does not match hostname: ' + machine + ', ' + value_str + 318 'MACHINE does not match hostname: ' + machine + ', ' + value_str +
192 '\nIs tianocore ovmf installed?') 319 '\nIs tianocore ovmf installed?')
193 print(value_str) 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))
194 408
195 409
196def qemu_launch(efi=False, machine=None): 410def qemu_launch(efi=False, machine=None):
@@ -220,12 +434,14 @@ def qemu_launch(efi=False, machine=None):
220 sleep(10) 434 sleep(10)
221 return qemu, s 435 return qemu, s
222 436
437
223def qemu_terminate(s): 438def qemu_terminate(s):
224 try: 439 try:
225 s.terminate() 440 s.terminate()
226 except KeyboardInterrupt: 441 except KeyboardInterrupt:
227 pass 442 pass
228 443
444
229def qemu_send_command(port, command): 445def qemu_send_command(port, command):
230 command = ['ssh -q -o UserKnownHostsFile=/dev/null -o StrictHostKeyChecking=no root@localhost -p ' + 446 command = ['ssh -q -o UserKnownHostsFile=/dev/null -o StrictHostKeyChecking=no root@localhost -p ' +
231 str(port) + ' "' + command + '"'] 447 str(port) + ' "' + command + '"']
@@ -233,4 +449,27 @@ def qemu_send_command(port, command):
233 stdout, stderr = s2.communicate() 449 stdout, stderr = s2.communicate()
234 return stdout, stderr, s2.returncode 450 return stdout, stderr, s2.returncode
235 451
452
453def 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
236# vim:set ts=4 sw=4 sts=4 expandtab: 475# vim:set ts=4 sw=4 sts=4 expandtab: