diff options
Diffstat (limited to 'lib/oeqa/selftest/cases/updater_qemux86_64.py')
-rw-r--r-- | lib/oeqa/selftest/cases/updater_qemux86_64.py | 441 |
1 files changed, 441 insertions, 0 deletions
diff --git a/lib/oeqa/selftest/cases/updater_qemux86_64.py b/lib/oeqa/selftest/cases/updater_qemux86_64.py new file mode 100644 index 0000000..814e139 --- /dev/null +++ b/lib/oeqa/selftest/cases/updater_qemux86_64.py | |||
@@ -0,0 +1,441 @@ | |||
1 | # pylint: disable=C0111,C0325 | ||
2 | import os | ||
3 | import logging | ||
4 | import re | ||
5 | import unittest | ||
6 | from time import sleep | ||
7 | |||
8 | from oeqa.selftest.case import OESelftestTestCase | ||
9 | from oeqa.utils.commands import runCmd, bitbake, get_bb_var, get_bb_vars | ||
10 | from testutils import qemu_launch, qemu_send_command, qemu_terminate, \ | ||
11 | akt_native_run, verifyProvisioned | ||
12 | |||
13 | |||
14 | class GeneralTests(OESelftestTestCase): | ||
15 | def test_credentials(self): | ||
16 | logger = logging.getLogger("selftest") | ||
17 | logger.info('Running bitbake to build core-image-minimal') | ||
18 | self.append_config('SOTA_CLIENT_PROV = "aktualizr-auto-prov"') | ||
19 | bitbake('core-image-minimal') | ||
20 | credentials = get_bb_var('SOTA_PACKED_CREDENTIALS') | ||
21 | # skip the test if the variable SOTA_PACKED_CREDENTIALS is not set | ||
22 | if credentials is None: | ||
23 | raise unittest.SkipTest("Variable 'SOTA_PACKED_CREDENTIALS' not set.") | ||
24 | # Check if the file exists | ||
25 | self.assertTrue(os.path.isfile(credentials), "File %s does not exist" % credentials) | ||
26 | deploydir = get_bb_var('DEPLOY_DIR_IMAGE') | ||
27 | imagename = get_bb_var('IMAGE_LINK_NAME', 'core-image-minimal') | ||
28 | # Check if the credentials are included in the output image | ||
29 | result = runCmd('tar -jtvf %s/%s.tar.bz2 | grep sota_provisioning_credentials.zip' % | ||
30 | (deploydir, imagename), ignore_status=True) | ||
31 | self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output) | ||
32 | |||
33 | |||
34 | class AktualizrToolsTests(OESelftestTestCase): | ||
35 | |||
36 | @classmethod | ||
37 | def setUpClass(cls): | ||
38 | super(AktualizrToolsTests, cls).setUpClass() | ||
39 | logger = logging.getLogger("selftest") | ||
40 | logger.info('Running bitbake to build aktualizr-native tools') | ||
41 | bitbake('aktualizr-native') | ||
42 | |||
43 | def test_cert_provider_help(self): | ||
44 | akt_native_run(self, 'aktualizr-cert-provider --help') | ||
45 | |||
46 | def test_cert_provider_local_output(self): | ||
47 | logger = logging.getLogger("selftest") | ||
48 | logger.info('Running bitbake to build aktualizr-ca-implicit-prov') | ||
49 | bitbake('aktualizr-ca-implicit-prov') | ||
50 | bb_vars = get_bb_vars(['SOTA_PACKED_CREDENTIALS', 'T'], 'aktualizr-native') | ||
51 | creds = bb_vars['SOTA_PACKED_CREDENTIALS'] | ||
52 | temp_dir = bb_vars['T'] | ||
53 | bb_vars_prov = get_bb_vars(['STAGING_DIR_HOST', 'libdir'], 'aktualizr-ca-implicit-prov') | ||
54 | config = bb_vars_prov['STAGING_DIR_HOST'] + bb_vars_prov['libdir'] + '/sota/sota_implicit_prov_ca.toml' | ||
55 | |||
56 | akt_native_run(self, 'aktualizr-cert-provider -c {creds} -r -l {temp} -g {config}' | ||
57 | .format(creds=creds, temp=temp_dir, config=config)) | ||
58 | |||
59 | # Might be nice if these names weren't hardcoded. | ||
60 | cert_path = temp_dir + '/var/sota/import/client.pem' | ||
61 | self.assertTrue(os.path.isfile(cert_path), "Client certificate not found at %s." % cert_path) | ||
62 | self.assertTrue(os.path.getsize(cert_path) > 0, "Client certificate at %s is empty." % cert_path) | ||
63 | pkey_path = temp_dir + '/var/sota/import/pkey.pem' | ||
64 | self.assertTrue(os.path.isfile(pkey_path), "Private key not found at %s." % pkey_path) | ||
65 | self.assertTrue(os.path.getsize(pkey_path) > 0, "Private key at %s is empty." % pkey_path) | ||
66 | ca_path = temp_dir + '/var/sota/import/root.crt' | ||
67 | self.assertTrue(os.path.isfile(ca_path), "Client certificate not found at %s." % ca_path) | ||
68 | self.assertTrue(os.path.getsize(ca_path) > 0, "Client certificate at %s is empty." % ca_path) | ||
69 | |||
70 | |||
71 | class AutoProvTests(OESelftestTestCase): | ||
72 | |||
73 | def setUpLocal(self): | ||
74 | layer = "meta-updater-qemux86-64" | ||
75 | result = runCmd('bitbake-layers show-layers') | ||
76 | if re.search(layer, result.output) is None: | ||
77 | # Assume the directory layout for finding other layers. We could also | ||
78 | # make assumptions by using 'show-layers', but either way, if the | ||
79 | # layers we need aren't where we expect them, we are out of luck. | ||
80 | path = os.path.abspath(os.path.dirname(__file__)) | ||
81 | metadir = path + "/../../../../../" | ||
82 | self.meta_qemu = metadir + layer | ||
83 | runCmd('bitbake-layers add-layer "%s"' % self.meta_qemu) | ||
84 | else: | ||
85 | self.meta_qemu = None | ||
86 | self.append_config('MACHINE = "qemux86-64"') | ||
87 | self.append_config('SOTA_CLIENT_PROV = " aktualizr-auto-prov "') | ||
88 | self.qemu, self.s = qemu_launch(machine='qemux86-64') | ||
89 | |||
90 | def tearDownLocal(self): | ||
91 | qemu_terminate(self.s) | ||
92 | if self.meta_qemu: | ||
93 | runCmd('bitbake-layers remove-layer "%s"' % self.meta_qemu, ignore_status=True) | ||
94 | |||
95 | def qemu_command(self, command): | ||
96 | return qemu_send_command(self.qemu.ssh_port, command) | ||
97 | |||
98 | def test_provisioning(self): | ||
99 | print('Checking machine name (hostname) of device:') | ||
100 | stdout, stderr, retcode = self.qemu_command('hostname') | ||
101 | self.assertEqual(retcode, 0, "Unable to check hostname. " + | ||
102 | "Is an ssh daemon (such as dropbear or openssh) installed on the device?") | ||
103 | machine = get_bb_var('MACHINE', 'core-image-minimal') | ||
104 | self.assertEqual(stderr, b'', 'Error: ' + stderr.decode()) | ||
105 | # Strip off line ending. | ||
106 | value = stdout.decode()[:-1] | ||
107 | self.assertEqual(value, machine, | ||
108 | 'MACHINE does not match hostname: ' + machine + ', ' + value) | ||
109 | print(value) | ||
110 | print('Checking output of aktualizr-info:') | ||
111 | ran_ok = False | ||
112 | for delay in [1, 2, 5, 10, 15]: | ||
113 | stdout, stderr, retcode = self.qemu_command('aktualizr-info') | ||
114 | if retcode == 0 and stderr == b'': | ||
115 | ran_ok = True | ||
116 | break | ||
117 | sleep(delay) | ||
118 | self.assertTrue(ran_ok, 'aktualizr-info failed: ' + stderr.decode() + stdout.decode()) | ||
119 | |||
120 | verifyProvisioned(self, machine) | ||
121 | |||
122 | |||
123 | class ManualControlTests(OESelftestTestCase): | ||
124 | |||
125 | def setUpLocal(self): | ||
126 | layer = "meta-updater-qemux86-64" | ||
127 | result = runCmd('bitbake-layers show-layers') | ||
128 | if re.search(layer, result.output) is None: | ||
129 | # Assume the directory layout for finding other layers. We could also | ||
130 | # make assumptions by using 'show-layers', but either way, if the | ||
131 | # layers we need aren't where we expect them, we are out of like. | ||
132 | path = os.path.abspath(os.path.dirname(__file__)) | ||
133 | metadir = path + "/../../../../../" | ||
134 | self.meta_qemu = metadir + layer | ||
135 | runCmd('bitbake-layers add-layer "%s"' % self.meta_qemu) | ||
136 | else: | ||
137 | self.meta_qemu = None | ||
138 | self.append_config('MACHINE = "qemux86-64"') | ||
139 | self.append_config('SOTA_CLIENT_PROV = " aktualizr-auto-prov "') | ||
140 | self.append_config('SYSTEMD_AUTO_ENABLE_aktualizr = "disable"') | ||
141 | self.qemu, self.s = qemu_launch(machine='qemux86-64') | ||
142 | |||
143 | def tearDownLocal(self): | ||
144 | qemu_terminate(self.s) | ||
145 | if self.meta_qemu: | ||
146 | runCmd('bitbake-layers remove-layer "%s"' % self.meta_qemu, ignore_status=True) | ||
147 | |||
148 | def qemu_command(self, command): | ||
149 | return qemu_send_command(self.qemu.ssh_port, command) | ||
150 | |||
151 | def test_manual_run_mode_once(self): | ||
152 | """ | ||
153 | Disable the systemd service then run aktualizr manually | ||
154 | """ | ||
155 | sleep(20) | ||
156 | stdout, stderr, retcode = self.qemu_command('aktualizr-info') | ||
157 | self.assertIn(b'Can\'t open database', stderr, | ||
158 | 'Aktualizr should not have run yet' + stderr.decode() + stdout.decode()) | ||
159 | |||
160 | stdout, stderr, retcode = self.qemu_command('aktualizr once') | ||
161 | |||
162 | stdout, stderr, retcode = self.qemu_command('aktualizr-info') | ||
163 | self.assertIn(b'Fetched metadata: yes', stdout, | ||
164 | 'Aktualizr should have run' + stderr.decode() + stdout.decode()) | ||
165 | |||
166 | |||
167 | class ImplProvTests(OESelftestTestCase): | ||
168 | |||
169 | def setUpLocal(self): | ||
170 | layer = "meta-updater-qemux86-64" | ||
171 | result = runCmd('bitbake-layers show-layers') | ||
172 | if re.search(layer, result.output) is None: | ||
173 | # Assume the directory layout for finding other layers. We could also | ||
174 | # make assumptions by using 'show-layers', but either way, if the | ||
175 | # layers we need aren't where we expect them, we are out of luck. | ||
176 | path = os.path.abspath(os.path.dirname(__file__)) | ||
177 | metadir = path + "/../../../../../" | ||
178 | self.meta_qemu = metadir + layer | ||
179 | runCmd('bitbake-layers add-layer "%s"' % self.meta_qemu) | ||
180 | else: | ||
181 | self.meta_qemu = None | ||
182 | self.append_config('MACHINE = "qemux86-64"') | ||
183 | self.append_config('SOTA_CLIENT_PROV = " aktualizr-ca-implicit-prov "') | ||
184 | self.append_config('SOTA_DEPLOY_CREDENTIALS = "0"') | ||
185 | runCmd('bitbake -c cleanall aktualizr aktualizr-ca-implicit-prov') | ||
186 | self.qemu, self.s = qemu_launch(machine='qemux86-64') | ||
187 | |||
188 | def tearDownLocal(self): | ||
189 | qemu_terminate(self.s) | ||
190 | if self.meta_qemu: | ||
191 | runCmd('bitbake-layers remove-layer "%s"' % self.meta_qemu, ignore_status=True) | ||
192 | |||
193 | def qemu_command(self, command): | ||
194 | return qemu_send_command(self.qemu.ssh_port, command) | ||
195 | |||
196 | def test_provisioning(self): | ||
197 | print('Checking machine name (hostname) of device:') | ||
198 | stdout, stderr, retcode = self.qemu_command('hostname') | ||
199 | self.assertEqual(retcode, 0, "Unable to check hostname. " + | ||
200 | "Is an ssh daemon (such as dropbear or openssh) installed on the device?") | ||
201 | machine = get_bb_var('MACHINE', 'core-image-minimal') | ||
202 | self.assertEqual(stderr, b'', 'Error: ' + stderr.decode()) | ||
203 | # Strip off line ending. | ||
204 | value = stdout.decode()[:-1] | ||
205 | self.assertEqual(value, machine, | ||
206 | 'MACHINE does not match hostname: ' + machine + ', ' + value) | ||
207 | print(value) | ||
208 | print('Checking output of aktualizr-info:') | ||
209 | ran_ok = False | ||
210 | for delay in [1, 2, 5, 10, 15]: | ||
211 | stdout, stderr, retcode = self.qemu_command('aktualizr-info') | ||
212 | if retcode == 0 and stderr == b'': | ||
213 | ran_ok = True | ||
214 | break | ||
215 | sleep(delay) | ||
216 | self.assertTrue(ran_ok, 'aktualizr-info failed: ' + stderr.decode() + stdout.decode()) | ||
217 | # Verify that device has NOT yet provisioned. | ||
218 | self.assertIn(b'Couldn\'t load device ID', stdout, | ||
219 | 'Device already provisioned!? ' + stderr.decode() + stdout.decode()) | ||
220 | self.assertIn(b'Couldn\'t load ECU serials', stdout, | ||
221 | 'Device already provisioned!? ' + stderr.decode() + stdout.decode()) | ||
222 | self.assertIn(b'Provisioned on server: no', stdout, | ||
223 | 'Device already provisioned!? ' + stderr.decode() + stdout.decode()) | ||
224 | self.assertIn(b'Fetched metadata: no', stdout, | ||
225 | 'Device already provisioned!? ' + stderr.decode() + stdout.decode()) | ||
226 | |||
227 | # Run aktualizr-cert-provider. | ||
228 | bb_vars = get_bb_vars(['SOTA_PACKED_CREDENTIALS'], 'aktualizr-native') | ||
229 | creds = bb_vars['SOTA_PACKED_CREDENTIALS'] | ||
230 | bb_vars_prov = get_bb_vars(['STAGING_DIR_HOST', 'libdir'], 'aktualizr-ca-implicit-prov') | ||
231 | config = bb_vars_prov['STAGING_DIR_HOST'] + bb_vars_prov['libdir'] + '/sota/sota_implicit_prov_ca.toml' | ||
232 | |||
233 | print('Provisining at root@localhost:%d' % self.qemu.ssh_port) | ||
234 | akt_native_run(self, 'aktualizr-cert-provider -c {creds} -t root@localhost -p {port} -s -u -r -g {config}' | ||
235 | .format(creds=creds, port=self.qemu.ssh_port, config=config)) | ||
236 | |||
237 | verifyProvisioned(self, machine) | ||
238 | |||
239 | |||
240 | class HsmTests(OESelftestTestCase): | ||
241 | |||
242 | def setUpLocal(self): | ||
243 | layer = "meta-updater-qemux86-64" | ||
244 | result = runCmd('bitbake-layers show-layers') | ||
245 | if re.search(layer, result.output) is None: | ||
246 | # Assume the directory layout for finding other layers. We could also | ||
247 | # make assumptions by using 'show-layers', but either way, if the | ||
248 | # layers we need aren't where we expect them, we are out of luck. | ||
249 | path = os.path.abspath(os.path.dirname(__file__)) | ||
250 | metadir = path + "/../../../../../" | ||
251 | self.meta_qemu = metadir + layer | ||
252 | runCmd('bitbake-layers add-layer "%s"' % self.meta_qemu) | ||
253 | else: | ||
254 | self.meta_qemu = None | ||
255 | self.append_config('MACHINE = "qemux86-64"') | ||
256 | self.append_config('SOTA_CLIENT_PROV = "aktualizr-hsm-prov"') | ||
257 | self.append_config('SOTA_DEPLOY_CREDENTIALS = "0"') | ||
258 | self.append_config('SOTA_CLIENT_FEATURES = "hsm"') | ||
259 | self.append_config('IMAGE_INSTALL_append = " softhsm-testtoken"') | ||
260 | runCmd('bitbake -c cleanall aktualizr aktualizr-hsm-prov') | ||
261 | self.qemu, self.s = qemu_launch(machine='qemux86-64') | ||
262 | |||
263 | def tearDownLocal(self): | ||
264 | qemu_terminate(self.s) | ||
265 | if self.meta_qemu: | ||
266 | runCmd('bitbake-layers remove-layer "%s"' % self.meta_qemu, ignore_status=True) | ||
267 | |||
268 | def qemu_command(self, command): | ||
269 | return qemu_send_command(self.qemu.ssh_port, command) | ||
270 | |||
271 | def test_provisioning(self): | ||
272 | print('Checking machine name (hostname) of device:') | ||
273 | stdout, stderr, retcode = self.qemu_command('hostname') | ||
274 | self.assertEqual(retcode, 0, "Unable to check hostname. " + | ||
275 | "Is an ssh daemon (such as dropbear or openssh) installed on the device?") | ||
276 | machine = get_bb_var('MACHINE', 'core-image-minimal') | ||
277 | self.assertEqual(stderr, b'', 'Error: ' + stderr.decode()) | ||
278 | # Strip off line ending. | ||
279 | value = stdout.decode()[:-1] | ||
280 | self.assertEqual(value, machine, | ||
281 | 'MACHINE does not match hostname: ' + machine + ', ' + value) | ||
282 | print(value) | ||
283 | print('Checking output of aktualizr-info:') | ||
284 | ran_ok = False | ||
285 | for delay in [1, 2, 5, 10, 15]: | ||
286 | stdout, stderr, retcode = self.qemu_command('aktualizr-info') | ||
287 | if retcode == 0 and stderr == b'': | ||
288 | ran_ok = True | ||
289 | break | ||
290 | sleep(delay) | ||
291 | self.assertTrue(ran_ok, 'aktualizr-info failed: ' + stderr.decode() + stdout.decode()) | ||
292 | # Verify that device has NOT yet provisioned. | ||
293 | self.assertIn(b'Couldn\'t load device ID', stdout, | ||
294 | 'Device already provisioned!? ' + stderr.decode() + stdout.decode()) | ||
295 | self.assertIn(b'Couldn\'t load ECU serials', stdout, | ||
296 | 'Device already provisioned!? ' + stderr.decode() + stdout.decode()) | ||
297 | self.assertIn(b'Provisioned on server: no', stdout, | ||
298 | 'Device already provisioned!? ' + stderr.decode() + stdout.decode()) | ||
299 | self.assertIn(b'Fetched metadata: no', stdout, | ||
300 | 'Device already provisioned!? ' + stderr.decode() + stdout.decode()) | ||
301 | |||
302 | # Verify that HSM is not yet initialized. | ||
303 | pkcs11_command = 'pkcs11-tool --module=/usr/lib/softhsm/libsofthsm2.so -O' | ||
304 | stdout, stderr, retcode = self.qemu_command(pkcs11_command) | ||
305 | self.assertNotEqual(retcode, 0, 'pkcs11-tool succeeded before initialization: ' + | ||
306 | stdout.decode() + stderr.decode()) | ||
307 | softhsm2_command = 'softhsm2-util --show-slots' | ||
308 | stdout, stderr, retcode = self.qemu_command(softhsm2_command) | ||
309 | self.assertNotEqual(retcode, 0, 'softhsm2-tool succeeded before initialization: ' + | ||
310 | stdout.decode() + stderr.decode()) | ||
311 | |||
312 | # Run aktualizr-cert-provider. | ||
313 | bb_vars = get_bb_vars(['SOTA_PACKED_CREDENTIALS'], 'aktualizr-native') | ||
314 | creds = bb_vars['SOTA_PACKED_CREDENTIALS'] | ||
315 | bb_vars_prov = get_bb_vars(['STAGING_DIR_NATIVE', 'libdir'], 'aktualizr-hsm-prov') | ||
316 | config = bb_vars_prov['STAGING_DIR_NATIVE'] + bb_vars_prov['libdir'] + '/sota/sota_hsm_prov.toml' | ||
317 | |||
318 | akt_native_run(self, 'aktualizr-cert-provider -c {creds} -t root@localhost -p {port} -r -s -u -g {config}' | ||
319 | .format(creds=creds, port=self.qemu.ssh_port, config=config)) | ||
320 | |||
321 | # Verify that HSM is able to initialize. | ||
322 | ran_ok = False | ||
323 | for delay in [5, 5, 5, 5, 10]: | ||
324 | sleep(delay) | ||
325 | p11_out, p11_err, p11_ret = self.qemu_command(pkcs11_command) | ||
326 | hsm_out, hsm_err, hsm_ret = self.qemu_command(softhsm2_command) | ||
327 | if p11_ret == 0 and hsm_ret == 0 and hsm_err == b'': | ||
328 | ran_ok = True | ||
329 | break | ||
330 | self.assertTrue(ran_ok, 'pkcs11-tool or softhsm2-tool failed: ' + p11_err.decode() + | ||
331 | p11_out.decode() + hsm_err.decode() + hsm_out.decode()) | ||
332 | self.assertIn(b'present token', p11_err, 'pkcs11-tool failed: ' + p11_err.decode() + p11_out.decode()) | ||
333 | self.assertIn(b'X.509 cert', p11_out, 'pkcs11-tool failed: ' + p11_err.decode() + p11_out.decode()) | ||
334 | self.assertIn(b'Initialized: yes', hsm_out, 'softhsm2-tool failed: ' + | ||
335 | hsm_err.decode() + hsm_out.decode()) | ||
336 | self.assertIn(b'User PIN init.: yes', hsm_out, 'softhsm2-tool failed: ' + | ||
337 | hsm_err.decode() + hsm_out.decode()) | ||
338 | |||
339 | # Check that pkcs11 output matches sofhsm output. | ||
340 | p11_p = re.compile(r'Using slot [0-9] with a present token \((0x[0-9a-f]*)\)\s') | ||
341 | p11_m = p11_p.search(p11_err.decode()) | ||
342 | self.assertTrue(p11_m, 'Slot number not found with pkcs11-tool: ' + p11_err.decode() + p11_out.decode()) | ||
343 | self.assertGreater(p11_m.lastindex, 0, 'Slot number not found with pkcs11-tool: ' + | ||
344 | p11_err.decode() + p11_out.decode()) | ||
345 | hsm_p = re.compile(r'Description:\s*SoftHSM slot ID (0x[0-9a-f]*)\s') | ||
346 | hsm_m = hsm_p.search(hsm_out.decode()) | ||
347 | self.assertTrue(hsm_m, 'Slot number not found with softhsm2-tool: ' + hsm_err.decode() + hsm_out.decode()) | ||
348 | self.assertGreater(hsm_m.lastindex, 0, 'Slot number not found with softhsm2-tool: ' + | ||
349 | hsm_err.decode() + hsm_out.decode()) | ||
350 | self.assertEqual(p11_m.group(1), hsm_m.group(1), 'Slot number does not match: ' + | ||
351 | p11_err.decode() + p11_out.decode() + hsm_err.decode() + hsm_out.decode()) | ||
352 | |||
353 | verifyProvisioned(self, machine) | ||
354 | |||
355 | |||
356 | class SecondaryTests(OESelftestTestCase): | ||
357 | @classmethod | ||
358 | def setUpClass(cls): | ||
359 | super(SecondaryTests, cls).setUpClass() | ||
360 | logger = logging.getLogger("selftest") | ||
361 | logger.info('Running bitbake to build secondary-image') | ||
362 | bitbake('secondary-image') | ||
363 | |||
364 | def setUpLocal(self): | ||
365 | layer = "meta-updater-qemux86-64" | ||
366 | result = runCmd('bitbake-layers show-layers') | ||
367 | if re.search(layer, result.output) is None: | ||
368 | # Assume the directory layout for finding other layers. We could also | ||
369 | # make assumptions by using 'show-layers', but either way, if the | ||
370 | # layers we need aren't where we expect them, we are out of luck. | ||
371 | path = os.path.abspath(os.path.dirname(__file__)) | ||
372 | metadir = path + "/../../../../../" | ||
373 | self.meta_qemu = metadir + layer | ||
374 | runCmd('bitbake-layers add-layer "%s"' % self.meta_qemu) | ||
375 | else: | ||
376 | self.meta_qemu = None | ||
377 | self.append_config('MACHINE = "qemux86-64"') | ||
378 | self.append_config('SOTA_CLIENT_PROV = " aktualizr-auto-prov "') | ||
379 | self.qemu, self.s = qemu_launch(machine='qemux86-64', imagename='secondary-image') | ||
380 | |||
381 | def tearDownLocal(self): | ||
382 | qemu_terminate(self.s) | ||
383 | if self.meta_qemu: | ||
384 | runCmd('bitbake-layers remove-layer "%s"' % self.meta_qemu, ignore_status=True) | ||
385 | |||
386 | def qemu_command(self, command): | ||
387 | return qemu_send_command(self.qemu.ssh_port, command) | ||
388 | |||
389 | def test_secondary_present(self): | ||
390 | print('Checking aktualizr-secondary is present') | ||
391 | stdout, stderr, retcode = self.qemu_command('aktualizr-secondary --help') | ||
392 | self.assertEqual(retcode, 0, "Unable to run aktualizr-secondary --help") | ||
393 | self.assertEqual(stderr, b'', 'Error: ' + stderr.decode()) | ||
394 | |||
395 | def test_secondary_listening(self): | ||
396 | print('Checking aktualizr-secondary service is listening') | ||
397 | stdout, stderr, retcode = self.qemu_command('aktualizr-check-discovery') | ||
398 | self.assertEqual(retcode, 0, "Unable to connect to secondary") | ||
399 | |||
400 | |||
401 | class PrimaryTests(OESelftestTestCase): | ||
402 | @classmethod | ||
403 | def setUpClass(cls): | ||
404 | super(PrimaryTests, cls).setUpClass() | ||
405 | logger = logging.getLogger("selftest") | ||
406 | logger.info('Running bitbake to build primary-image') | ||
407 | bitbake('primary-image') | ||
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-auto-prov "') | ||
424 | self.append_config('SOTA_CLIENT_FEATURES = "secondary-network"') | ||
425 | self.qemu, self.s = qemu_launch(machine='qemux86-64', imagename='primary-image') | ||
426 | |||
427 | def tearDownLocal(self): | ||
428 | qemu_terminate(self.s) | ||
429 | if self.meta_qemu: | ||
430 | runCmd('bitbake-layers remove-layer "%s"' % self.meta_qemu, ignore_status=True) | ||
431 | |||
432 | def qemu_command(self, command): | ||
433 | return qemu_send_command(self.qemu.ssh_port, command) | ||
434 | |||
435 | def test_aktualizr_present(self): | ||
436 | print('Checking aktualizr is present') | ||
437 | stdout, stderr, retcode = self.qemu_command('aktualizr --help') | ||
438 | self.assertEqual(retcode, 0, "Unable to run aktualizr --help") | ||
439 | self.assertEqual(stderr, b'', 'Error: ' + stderr.decode()) | ||
440 | |||
441 | # vim:set ts=4 sw=4 sts=4 expandtab: | ||