Refactor oe selftests

Factorize the call of aktualizr-native commands in each testing class

1 files changed, 70 insertions, 51 deletions

diff --git a/lib/oeqa/selftest/updater.py b/lib/oeqa/selftest/updater.py
index 703df91..8563e3b 100644
--- a/lib/oeqa/selftest/updater.py
+++ b/lib/oeqa/selftest/updater.py
@@ -16,32 +16,32 @@ class SotaToolsTests(oeSelfTest):
     @classmethod
     def setUpClass(cls):
         logger = logging.getLogger("selftest")
+        bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'base_prefix', 'libdir', 'bindir'],
+                              'aktualizr-native')
+        cls.sysroot = bb_vars['SYSROOT_DESTDIR'] + bb_vars['base_prefix']
+        cls.sysrootbin = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir']
+        cls.libdir = bb_vars['libdir']
+
         logger.info('Running bitbake to build aktualizr-native tools')
         bitbake('aktualizr-native')
 
-    def test_push_help(self):
-        bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'bindir', 'libdir'], 'aktualizr-native')
-        l = bb_vars['libdir']
-        p = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir'] + "/garage-push"
-        self.assertTrue(os.path.isfile(p), msg = "No garage-push found (%s)" % p)
-        result = runCmd('LD_LIBRARY_PATH=%s %s --help' % (l, p), ignore_status=True)
+    def runNativeCmd(self, cmd, **kwargs):
+        program, *_ = cmd.split(' ')
+        p = '{}/{}'.format(self.sysrootbin, program)
+        self.assertTrue(os.path.isfile(p), msg="No {} found ({})".format(program, p))
+        env = dict(os.environ)
+        env['LD_LIBRARY_PATH'] = self.libdir
+        result = runCmd(cmd, env=env, native_sysroot=self.sysroot, ignore_status=True, **kwargs)
         self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output)
 
+    def test_push_help(self):
+        self.runNativeCmd('garage-push --help')
+
     def test_deploy_help(self):
-        bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'bindir', 'libdir'], 'aktualizr-native')
-        l = bb_vars['libdir']
-        p = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir'] + "/garage-deploy"
-        self.assertTrue(os.path.isfile(p), msg = "No garage-deploy found (%s)" % p)
-        result = runCmd('LD_LIBRARY_PATH=%s %s --help' % (l, p), ignore_status=True)
-        self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output)
+        self.runNativeCmd('garage-deploy --help')
 
     def test_garagesign_help(self):
-        bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'bindir', 'libdir'], 'aktualizr-native')
-        l = bb_vars['libdir']
-        p = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir'] + "/garage-sign"
-        self.assertTrue(os.path.isfile(p), msg = "No garage-sign found (%s)" % p)
-        result = runCmd('LD_LIBRARY_PATH=%s %s --help' % (l, p), ignore_status=True)
-        self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output)
+        self.runNativeCmd('garage-sign --help')
 
 
 class GeneralTests(oeSelfTest):
@@ -113,24 +113,29 @@ class AktualizrToolsTests(oeSelfTest):
     @classmethod
     def setUpClass(cls):
         logger = logging.getLogger("selftest")
+        bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'base_prefix', 'libdir', 'bindir'],
+                              'aktualizr-native')
+        cls.sysroot = bb_vars['SYSROOT_DESTDIR'] + bb_vars['base_prefix']
+        cls.sysrootbin = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir']
+        cls.libdir = bb_vars['libdir']
+
         logger.info('Running bitbake to build aktualizr-native tools')
         bitbake('aktualizr-native')
 
-    def test_implicit_writer_help(self):
-        bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'bindir', 'libdir'], 'aktualizr-native')
-        l = bb_vars['libdir']
-        p = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir'] + "/aktualizr_implicit_writer"
-        self.assertTrue(os.path.isfile(p), msg = "No aktualizr_implicit_writer found (%s)" % p)
-        result = runCmd('LD_LIBRARY_PATH=%s %s --help' % (l, p), ignore_status=True)
+    def runNativeCmd(self, cmd, **kwargs):
+        program, *_ = cmd.split(' ')
+        p = '{}/{}'.format(self.sysrootbin, program)
+        self.assertTrue(os.path.isfile(p), msg="No {} found ({})".format(program, p))
+        env = dict(os.environ)
+        env['LD_LIBRARY_PATH'] = self.libdir
+        result = runCmd(cmd, env=env, native_sysroot=self.sysroot, ignore_status=True, **kwargs)
         self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output)
 
+    def test_implicit_writer_help(self):
+        self.runNativeCmd('aktualizr_implicit_writer --help')
+
     def test_cert_provider_help(self):
-        bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'bindir', 'libdir'], 'aktualizr-native')
-        l = bb_vars['libdir']
-        p = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir'] + "/aktualizr_cert_provider"
-        self.assertTrue(os.path.isfile(p), msg = "No aktualizr_cert_provider found (%s)" % p)
-        result = runCmd('LD_LIBRARY_PATH=%s %s --help' % (l, p), ignore_status=True)
-        self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output)
+        self.runNativeCmd('aktualizr_cert_provider --help')
 
     def test_cert_provider_local_output(self):
         logger = logging.getLogger("selftest")
@@ -138,17 +143,14 @@ class AktualizrToolsTests(oeSelfTest):
         bitbake('aktualizr-implicit-prov')
         bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'bindir', 'libdir',
                                'SOTA_PACKED_CREDENTIALS', 'T'], 'aktualizr-native')
-        l = bb_vars['libdir']
-        p = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir'] + "/aktualizr_cert_provider"
         creds = bb_vars['SOTA_PACKED_CREDENTIALS']
         temp_dir = bb_vars['T']
         bb_vars_prov = get_bb_vars(['STAGING_DIR_NATIVE', 'libdir'], 'aktualizr-implicit-prov')
         config = bb_vars_prov['STAGING_DIR_NATIVE'] + bb_vars_prov['libdir'] + '/sota/sota_implicit_prov.toml'
-        self.assertTrue(os.path.isfile(p), msg = "No aktualizr_cert_provider found (%s)" % p)
-        command = 'LD_LIBRARY_PATH=' + l + ' ' + p + ' -c ' + creds + ' -r -l ' + temp_dir + ' -g ' + config
-        # logger.info('Checking output of: ' + command)
-        result = runCmd(command, ignore_status=True)
-        self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output)
+
+        self.runNativeCmd('aktualizr_cert_provider -c {creds} -r -l {temp} -g {config}'
+                          .format(creds=creds, temp=temp_dir, config=config))
+
         # Might be nice if these names weren't hardcoded.
         cert_path = temp_dir + '/client.pem'
         self.assertTrue(os.path.isfile(cert_path), "Client certificate not found at %s." % cert_path)
@@ -178,7 +180,7 @@ class QemuTests(oeSelfTest):
         print('Checking machine name (hostname) of device:')
         stdout, stderr, retcode = self.qemu_command('hostname')
         self.assertEqual(retcode, 0, "Unable to check hostname. " +
-                                                 "Is an ssh daemon (such as dropbear or openssh) installed on the device?")
+                         "Is an ssh daemon (such as dropbear or openssh) installed on the device?")
         machine = get_bb_var('MACHINE', 'core-image-minimal')
         self.assertEqual(stderr, b'', 'Error: ' + stderr.decode())
         # Strip off line ending.
@@ -224,7 +226,7 @@ class GrubTests(oeSelfTest):
         print('Checking machine name (hostname) of device:')
         stdout, stderr, retcode = self.qemu_command('hostname')
         self.assertEqual(retcode, 0, "Unable to check hostname. " +
-                                                 "Is an ssh daemon (such as dropbear or openssh) installed on the device?")
+                         "Is an ssh daemon (such as dropbear or openssh) installed on the device?")
         machine = get_bb_var('MACHINE', 'core-image-minimal')
         self.assertEqual(stderr, b'', 'Error: ' + stderr.decode())
         # Strip off line ending.
@@ -246,6 +248,18 @@ class GrubTests(oeSelfTest):
 
 class HsmTests(oeSelfTest):
 
+    @classmethod
+    def setUpClass(cls):
+        logger = logging.getLogger("selftest")
+        bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'base_prefix', 'libdir', 'bindir'],
+                              'aktualizr-native')
+        cls.sysroot = bb_vars['SYSROOT_DESTDIR'] + bb_vars['base_prefix']
+        cls.sysrootbin = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir']
+        cls.libdir = bb_vars['libdir']
+
+        logger.info('Running bitbake to build aktualizr-native tools')
+        bitbake('aktualizr-native')
+
     def setUpLocal(self):
         self.append_config('SOTA_CLIENT_PROV = "aktualizr-hsm-prov"')
         self.append_config('SOTA_CLIENT_FEATURES = "hsm"')
@@ -254,6 +268,15 @@ class HsmTests(oeSelfTest):
     def tearDownLocal(self):
         qemu_terminate(self.s)
 
+    def runNativeCmd(self, cmd, **kwargs):
+        program, *_ = cmd.split(' ')
+        p = '{}/{}'.format(self.sysrootbin, program)
+        self.assertTrue(os.path.isfile(p), msg="No {} found ({})".format(program, p))
+        env = dict(os.environ)
+        env['LD_LIBRARY_PATH'] = self.libdir
+        result = runCmd(cmd, env=env, native_sysroot=self.sysroot, ignore_status=True, **kwargs)
+        self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output)
+
     def qemu_command(self, command):
         return qemu_send_command(self.qemu.ssh_port, command)
 
@@ -261,7 +284,7 @@ class HsmTests(oeSelfTest):
         print('Checking machine name (hostname) of device:')
         stdout, stderr, retcode = self.qemu_command('hostname')
         self.assertEqual(retcode, 0, "Unable to check hostname. " +
-                                                 "Is an ssh daemon (such as dropbear or openssh) installed on the device?")
+                         "Is an ssh daemon (such as dropbear or openssh) installed on the device?")
         machine = get_bb_var('MACHINE', 'core-image-minimal')
         self.assertEqual(stderr, b'', 'Error: ' + stderr.decode())
         # Strip off line ending.
@@ -291,7 +314,7 @@ class HsmTests(oeSelfTest):
         pkcs11_command = 'pkcs11-tool --module=/usr/lib/softhsm/libsofthsm2.so -O'
         stdout, stderr, retcode = self.qemu_command(pkcs11_command)
         self.assertNotEqual(retcode, 0, 'pkcs11-tool succeeded before initialization: ' +
-                        stdout.decode() + stderr.decode())
+                            stdout.decode() + stderr.decode())
         softhsm2_command = 'softhsm2-util --show-slots'
         stdout, stderr, retcode = self.qemu_command(softhsm2_command)
         self.assertNotEqual(retcode, 0, 'softhsm2-tool succeeded before initialization: ' +
@@ -300,18 +323,12 @@ class HsmTests(oeSelfTest):
         # Run cert_provider.
         bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'bindir', 'libdir',
                                'SOTA_PACKED_CREDENTIALS'], 'aktualizr-native')
-        l = bb_vars['libdir']
-        p = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir'] + "/aktualizr_cert_provider"
         creds = bb_vars['SOTA_PACKED_CREDENTIALS']
         bb_vars_prov = get_bb_vars(['STAGING_DIR_NATIVE', 'libdir'], 'aktualizr-hsm-prov')
         config = bb_vars_prov['STAGING_DIR_NATIVE'] + bb_vars_prov['libdir'] + '/sota/sota_implicit_prov.toml'
-        self.assertTrue(os.path.isfile(p), msg = "No aktualizr_cert_provider found (%s)" % p)
-        command = ('LD_LIBRARY_PATH=' + l + ' ' + p + ' -c ' + creds + ' -t root@localhost -p ' +
-                   str(self.qemu.ssh_port) + ' -r -s -g ' + config)
-        logger = logging.getLogger("selftest")
-        # logger.info('Checking output of: ' + command)
-        result = runCmd(command, ignore_status=True)
-        self.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output)
+
+        self.runNativeCmd('aktualizr_cert_provider -c {creds} -t root@localhost -p {port} -r -s -g {config}'
+                          .format(creds=creds, port=self.qemu.ssh_port, config=config))
 
         # Verify that HSM is able to initialize.
         ran_ok = False
@@ -361,7 +378,7 @@ class HsmTests(oeSelfTest):
         m = p.search(stdout.decode())
         self.assertTrue(m, 'Device ID could not be read: ' + stderr.decode() + stdout.decode())
         self.assertGreater(m.lastindex, 0, 'Device ID could not be read: ' + stderr.decode() + stdout.decode())
-        logger.info('Device successfully provisioned with ID: ' + m.group(1))
+        self.logger.info('Device successfully provisioned with ID: ' + m.group(1))
 
 
 def qemu_launch(efi=False, machine=None):
@@ -391,12 +408,14 @@ def qemu_launch(efi=False, machine=None):
     sleep(10)
     return qemu, s
 
+
 def qemu_terminate(s):
     try:
         s.terminate()
     except KeyboardInterrupt:
         pass
 
+
 def qemu_send_command(port, command):
     command = ['ssh -q -o UserKnownHostsFile=/dev/null -o StrictHostKeyChecking=no root@localhost -p ' +
                str(port) + ' "' + command + '"']