summaryrefslogtreecommitdiffstats
path: root/lib/oeqa/selftest/cases/testutils.py
blob: 90ba65339d6fe3b77919ceccbab18dafa29fe725 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
import os
import logging
import re
import subprocess
from time import sleep

from oeqa.utils.commands import runCmd, bitbake, get_bb_var, get_bb_vars
from qemucommand import QemuCommand


def qemu_launch(efi=False, machine=None, imagename=None):
    logger = logging.getLogger("selftest")
    if imagename is None:
        imagename = 'core-image-minimal'
    logger.info('Running bitbake to build {}'.format(imagename))
    bitbake(imagename)
    # Create empty object.
    args = type('', (), {})()
    args.imagename = imagename
    args.mac = None
    # Could use DEPLOY_DIR_IMAGE here but it's already in the machine
    # subdirectory.
    args.dir = 'tmp/deploy/images'
    args.efi = efi
    args.machine = machine
    qemu_use_kvm = get_bb_var("QEMU_USE_KVM")
    if qemu_use_kvm and \
            (qemu_use_kvm == 'True' and 'x86' in machine or
             get_bb_var('MACHINE') in qemu_use_kvm.split()):
        args.kvm = True
    else:
        args.kvm = None  # Autodetect
    args.no_gui = True
    args.gdb = False
    args.pcap = None
    args.overlay = None
    args.dry_run = False
    args.secondary_network = False

    qemu = QemuCommand(args)
    cmdline = qemu.command_line()
    print('Booting image with run-qemu-ota...')
    s = subprocess.Popen(cmdline)
    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 + '"']
    s2 = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    stdout, stderr = s2.communicate(timeout=60)
    return stdout, stderr, s2.returncode


def akt_native_run(testInst, cmd, **kwargs):
    # run a command supplied by aktualizr-native and checks that:
    # - the executable exists
    # - the command runs without error
    # NOTE: the base test class must have built aktualizr-native (in
    # setUpClass, for example)
    bb_vars = get_bb_vars(['SYSROOT_DESTDIR', 'base_prefix', 'libdir', 'bindir'],
                          'aktualizr-native')
    sysroot = bb_vars['SYSROOT_DESTDIR'] + bb_vars['base_prefix']
    sysrootbin = bb_vars['SYSROOT_DESTDIR'] + bb_vars['bindir']
    libdir = bb_vars['libdir']

    program, *_ = cmd.split(' ')
    p = '{}/{}'.format(sysrootbin, program)
    testInst.assertTrue(os.path.isfile(p), msg="No {} found ({})".format(program, p))
    env = dict(os.environ)
    env['LD_LIBRARY_PATH'] = libdir
    result = runCmd(cmd, env=env, native_sysroot=sysroot, ignore_status=True, **kwargs)
    testInst.assertEqual(result.status, 0, "Status not equal to 0. output: %s" % result.output)


def verifyNotProvisioned(testInst, machine):
    print('Checking output of aktualizr-info:')
    ran_ok = False
    for delay in [5, 5, 5, 5, 10, 10, 10, 10]:
        stdout, stderr, retcode = testInst.qemu_command('aktualizr-info')
        if retcode == 0 and stderr == b'':
            ran_ok = True
            break
        sleep(delay)
    testInst.assertTrue(ran_ok, 'aktualizr-info failed: ' + stderr.decode() + stdout.decode())

    # Verify that device has NOT yet provisioned.
    testInst.assertIn(b'Couldn\'t load device ID', stdout,
                      'Device already provisioned!? ' + stderr.decode() + stdout.decode())
    testInst.assertIn(b'Couldn\'t load ECU serials', stdout,
                      'Device already provisioned!? ' + stderr.decode() + stdout.decode())
    testInst.assertIn(b'Provisioned on server: no', stdout,
                      'Device already provisioned!? ' + stderr.decode() + stdout.decode())
    testInst.assertIn(b'Fetched metadata: no', stdout,
                      'Device already provisioned!? ' + stderr.decode() + stdout.decode())


def verifyProvisioned(testInst, machine):
    # Verify that device HAS provisioned.
    ran_ok = False
    for delay in [5, 5, 5, 5, 10, 10, 10, 10]:
        stdout, stderr, retcode = testInst.qemu_command('aktualizr-info')
        if retcode == 0 and stderr == b'' and stdout.decode().find('Fetched metadata: yes') >= 0:
            ran_ok = True
            break
        sleep(delay)
    testInst.assertTrue(ran_ok, 'aktualizr-info failed: ' + stderr.decode() + stdout.decode())

    testInst.assertIn(b'Device ID: ', stdout, 'Provisioning failed: ' + stderr.decode() + stdout.decode())
    testInst.assertIn(b'Primary ecu hardware ID: ' + machine.encode(), stdout,
                      'Provisioning failed: ' + stderr.decode() + stdout.decode())
    testInst.assertIn(b'Fetched metadata: yes', stdout, 'Provisioning failed: ' + stderr.decode() + stdout.decode())
    p = re.compile(r'Device ID: ([a-z0-9-]*)\n')
    m = p.search(stdout.decode())
    testInst.assertTrue(m, 'Device ID could not be read: ' + stderr.decode() + stdout.decode())
    testInst.assertGreater(m.lastindex, 0, 'Device ID could not be read: ' + stderr.decode() + stdout.decode())
    logger = logging.getLogger("selftest")
    logger.info('Device successfully provisioned with ID: ' + m.group(1))

# vim:set ts=4 sw=4 sts=4 expandtab: