path: root/lava_dispatcher/test/test_uboot.py

# Copyright (C) 2014 Linaro Limited
#
# Author: Neil Williams <neil.williams@linaro.org>
#
# This file is part of LAVA Dispatcher.
#
# LAVA Dispatcher is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# LAVA Dispatcher is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along
# with this program; if not, see <http://www.gnu.org/licenses>.


import os
import yaml
import logging
import unittest
from lava_dispatcher.device import NewDevice
from lava_dispatcher.parser import JobParser
from lava_dispatcher.actions.boot.u_boot import (
    UBootAction,
    UBootSecondaryMedia
)
from lava_dispatcher.actions.boot import BootloaderCommandOverlay
from lava_dispatcher.actions.deploy.apply_overlay import CompressRamdisk
from lava_dispatcher.actions.deploy.tftp import TftpAction
from lava_dispatcher.job import Job
from lava_dispatcher.action import Pipeline, JobError
from lava_dispatcher.test.test_basic import Factory, StdoutTestCase
from lava_dispatcher.test.utils import DummyLogger, infrastructure_error
from lava_dispatcher.utils.network import dispatcher_ip
from lava_dispatcher.utils.filesystem import tftpd_dir
from lava_dispatcher.utils.strings import substitute


class UBootFactory(Factory):  # pylint: disable=too-few-public-methods
    """
    Not Model based, this is not a Django factory.
    Factory objects are dispatcher based classes, independent
    of any database objects.
    """
    def create_bbb_job(self, filename):
        return self.create_job('bbb-03.jinja2', filename)

    def create_x15_job(self, filename):
        return self.create_job('x15-01.jinja2', filename)

    def create_juno_job(self, filename):
        return self.create_job('juno-r2-01.jinja2', filename)

    def create_zcu102_job(self, filename):
        return self.create_job('zcu102.jinja2', filename)


class TestUbootAction(StdoutTestCase):  # pylint: disable=too-many-public-methods

    def setUp(self):
        super().setUp()
        self.factory = UBootFactory()

    @unittest.skipIf(infrastructure_error('mkimage'), "u-boot-tools not installed")
    def test_simulated_action(self):
        job = self.factory.create_bbb_job('sample_jobs/uboot-ramdisk.yaml')
        self.assertIsNotNone(job)

        # uboot and uboot-ramdisk have the same pipeline structure
        description_ref = self.pipeline_reference('uboot.yaml', job=job)
        self.assertEqual(description_ref, job.pipeline.describe(False))

        self.assertIsNone(job.validate())

    def test_tftp_pipeline(self):
        job = self.factory.create_bbb_job('sample_jobs/uboot-ramdisk.yaml')
        self.assertEqual(
            [action.name for action in job.pipeline.actions],
            ['tftp-deploy', 'uboot-action', 'lava-test-retry', 'finalize']
        )
        tftp = [action for action in job.pipeline.actions if action.name == 'tftp-deploy'][0]
        self.assertTrue(tftp.get_namespace_data(action=tftp.name, label='tftp', key='ramdisk'))
        self.assertIsNotNone(tftp.internal_pipeline)
        self.assertEqual(
            [action.name for action in tftp.internal_pipeline.actions],
            ['download-retry', 'download-retry', 'download-retry', 'prepare-tftp-overlay', 'lxc-create-udev-rule-action', 'deploy-device-env']
        )
        self.assertIn('ramdisk', [action.key for action in tftp.internal_pipeline.actions if hasattr(action, 'key')])
        self.assertIn('kernel', [action.key for action in tftp.internal_pipeline.actions if hasattr(action, 'key')])
        self.assertIn('dtb', [action.key for action in tftp.internal_pipeline.actions if hasattr(action, 'key')])
        self.assertNotIn('=', tftpd_dir())
        job.validate()
        tftp.validate()
        self.assertEqual([], tftp.errors)

    def test_device_bbb(self):
        job = self.factory.create_bbb_job('sample_jobs/uboot.yaml')
        self.assertEqual(
            job.device['commands']['connections']['uart0']['connect'],
            'telnet localhost 6000'
        )
        self.assertEqual(job.device['commands'].get('interrupt', ' '), ' ')
        methods = job.device['actions']['boot']['methods']
        self.assertIn('u-boot', methods)
        self.assertEqual(methods['u-boot']['parameters'].get('bootloader_prompt'), 'U-Boot')

    @unittest.skipIf(infrastructure_error('mkimage'), "u-boot-tools not installed")
    def test_uboot_action(self):
        job = self.factory.create_bbb_job('sample_jobs/uboot-ramdisk.yaml')
        job.validate()
        self.assertEqual(job.pipeline.errors, [])
        self.assertIn('u-boot', job.device['actions']['boot']['methods'])
        params = job.device['actions']['deploy']['parameters']
        self.assertIn('mkimage_arch', params)
        boot_message = params.get('boot_message',
                                  job.device.get_constant('kernel-start-message'))
        self.assertIsNotNone(boot_message)
        for action in job.pipeline.actions:
            action.validate()
            if isinstance(action, UBootAction):
                self.assertIn('method', action.parameters)
                self.assertEqual('u-boot', action.parameters['method'])
                self.assertEqual(
                    'reboot: Restarting system',
                    action.parameters.get('parameters', {}).get('shutdown-message', job.device.get_constant('shutdown-message'))
                )
            if isinstance(action, TftpAction):
                self.assertIn('ramdisk', action.parameters)
                self.assertIn('kernel', action.parameters)
                self.assertIn('to', action.parameters)
                self.assertEqual('tftp', action.parameters['to'])
            if isinstance(action, CompressRamdisk):
                self.assertEqual(action.mkimage_arch, 'arm')
            self.assertTrue(action.valid)

    def test_fastboot_uboot(self):  # pylint: disable=too-many-locals
        job = self.factory.create_x15_job('sample_jobs/x15-uboot.yaml')
        job.validate()
        description_ref = self.pipeline_reference('x15-uboot.yaml', job=job)
        self.assertEqual(description_ref, job.pipeline.describe(False))
        deploy = [action for action in job.pipeline.actions if action.name == 'fastboot-deploy'][0]
        enter = [action for action in deploy.internal_pipeline.actions if action.name == 'uboot-enter-fastboot'][0]
        interrupt = [action for action in enter.internal_pipeline.actions if action.name == 'bootloader-interrupt'][0]
        self.assertIsNotNone(interrupt.params)
        self.assertNotEqual(interrupt.params, {})
        self.assertEqual('u-boot', interrupt.method)

    def test_x15_uboot_nfs(self):  # pylint: disable=too-many-locals
        job = self.factory.create_x15_job('sample_jobs/x15-nfs.yaml')
        job.validate()
        description_ref = self.pipeline_reference('x15-nfs.yaml', job=job)
        self.assertEqual(description_ref, job.pipeline.describe(False))
        tftp_deploy = [action for action in job.pipeline.actions if action.name == 'tftp-deploy'][0]
        prepare = [action for action in tftp_deploy.internal_pipeline.actions if action.name == 'prepare-tftp-overlay'][0]
        nfs = [action for action in prepare.internal_pipeline.actions if action.name == 'extract-nfsrootfs'][0]
        self.assertIn('compression', nfs.parameters['nfsrootfs'])
        self.assertEqual(nfs.parameters['nfsrootfs']['compression'], 'gz')

    def test_juno_uboot_nfs(self):
        job = self.factory.create_juno_job('sample_jobs/juno-uboot-nfs.yaml')
        job.validate()
        description_ref = self.pipeline_reference('juno-uboot-nfs.yaml', job=job)
        self.assertEqual(description_ref, job.pipeline.describe(False))

    def test_overlay_action(self):  # pylint: disable=too-many-locals
        parameters = {
            'device_type': 'beaglebone-black',
            'job_name': 'uboot-pipeline',
            'job_timeout': '15m',
            'action_timeout': '5m',
            'priority': 'medium',
            'actions': {
                'boot': {
                    'method': 'u-boot',
                    'commands': 'ramdisk',
                    'type': 'bootz',
                    'prompts': ['linaro-test', 'root@debian:~#']
                },
                'deploy': {
                    'ramdisk': 'initrd.gz',
                    'kernel': 'zImage',
                    'dtb': 'broken.dtb'
                }
            }
        }
        device = NewDevice(os.path.join(os.path.dirname(__file__), '../devices/bbb-01.yaml'))
        job = Job(4212, parameters, None)
        job.device = device
        pipeline = Pipeline(job=job, parameters=parameters['actions']['boot'])
        job.pipeline = pipeline
        overlay = BootloaderCommandOverlay()
        pipeline.add_action(overlay)
        ip_addr = dispatcher_ip(None)
        parsed = []
        kernel_addr = job.device['parameters'][overlay.parameters['type']]['ramdisk']
        ramdisk_addr = job.device['parameters'][overlay.parameters['type']]['ramdisk']
        dtb_addr = job.device['parameters'][overlay.parameters['type']]['dtb']
        kernel = parameters['actions']['deploy']['kernel']
        ramdisk = parameters['actions']['deploy']['ramdisk']
        dtb = parameters['actions']['deploy']['dtb']

        substitution_dictionary = {
            '{SERVER_IP}': ip_addr,
            # the addresses need to be hexadecimal
            '{KERNEL_ADDR}': kernel_addr,
            '{DTB_ADDR}': dtb_addr,
            '{RAMDISK_ADDR}': ramdisk_addr,
            '{BOOTX}': "%s %s %s %s" % (
                overlay.parameters['type'], kernel_addr, ramdisk_addr, dtb_addr),
            '{RAMDISK}': ramdisk,
            '{KERNEL}': kernel,
            '{DTB}': dtb
        }
        params = device['actions']['boot']['methods']
        params['u-boot']['ramdisk']['commands'] = substitute(params['u-boot']['ramdisk']['commands'], substitution_dictionary)

        commands = params['u-boot']['ramdisk']['commands']
        self.assertIs(type(commands), list)
        self.assertIn("setenv loadkernel 'tftp ${kernel_addr_r} zImage'", commands)
        self.assertIn("setenv loadinitrd 'tftp ${initrd_addr_r} initrd.gz; setenv initrd_size ${filesize}'", commands)
        self.assertIn("setenv loadfdt 'tftp ${fdt_addr_r} broken.dtb'", commands)
        self.assertNotIn("setenv kernel_addr_r '{KERNEL_ADDR}'", commands)
        self.assertNotIn("setenv initrd_addr_r '{RAMDISK_ADDR}'", commands)
        self.assertNotIn("setenv fdt_addr_r '{DTB_ADDR}'", commands)

        for line in params['u-boot']['ramdisk']['commands']:
            line = line.replace('{SERVER_IP}', ip_addr)
            # the addresses need to be hexadecimal
            line = line.replace('{KERNEL_ADDR}', kernel_addr)
            line = line.replace('{DTB_ADDR}', dtb_addr)
            line = line.replace('{RAMDISK_ADDR}', ramdisk_addr)
            line = line.replace('{BOOTX}', "%s %s %s %s" % (
                overlay.parameters['type'], kernel_addr, ramdisk_addr, dtb_addr))
            line = line.replace('{RAMDISK}', ramdisk)
            line = line.replace('{KERNEL}', kernel)
            line = line.replace('{DTB}', dtb)
            parsed.append(line)
        self.assertIn("setenv loadkernel 'tftp ${kernel_addr_r} zImage'", parsed)
        self.assertIn("setenv loadinitrd 'tftp ${initrd_addr_r} initrd.gz; setenv initrd_size ${filesize}'", parsed)
        self.assertIn("setenv loadfdt 'tftp ${fdt_addr_r} broken.dtb'", parsed)
        self.assertNotIn("setenv kernel_addr_r '{KERNEL_ADDR}'", parsed)
        self.assertNotIn("setenv initrd_addr_r '{RAMDISK_ADDR}'", parsed)
        self.assertNotIn("setenv fdt_addr_r '{DTB_ADDR}'", parsed)

    def test_boot_commands(self):
        job = self.factory.create_bbb_job('sample_jobs/uboot-ramdisk-inline-commands.yaml')
        job.validate()
        uboot = [action for action in job.pipeline.actions if action.name == 'uboot-action'][0]
        overlay = [action for action in uboot.internal_pipeline.actions if action.name == 'bootloader-overlay'][0]
        self.assertEqual(overlay.commands, ['a list', 'of commands', 'with a {KERNEL_ADDR} substitution'])

    @unittest.skipIf(infrastructure_error('xnbd-server'), "xnbd-server not installed")
    def test_nbd_boot(self):
        job = self.factory.create_bbb_job('sample_jobs/bbb-initrd-nbd.yaml')
        job.validate()
        self.assertEqual(job.pipeline.errors, [])
        description_ref = self.pipeline_reference('bbb-initrd-nbd.yaml', job=job)
        self.assertEqual(description_ref, job.pipeline.describe(False))
        # Fixme: more asserts
        self.assertIn('u-boot', job.device['actions']['boot']['methods'])
        params = job.device['actions']['deploy']['parameters']
        self.assertIsNotNone(params)
        for action in job.pipeline.actions:
            action.validate()
            if isinstance(action, UBootAction):
                self.assertIn('method', action.parameters)
                self.assertEqual('u-boot', action.parameters['method'])
            elif isinstance(action, TftpAction):
                self.assertIn('initrd', action.parameters)
                self.assertIn('kernel', action.parameters)
                self.assertIn('nbdroot', action.parameters)
                self.assertIn('to', action.parameters)
                self.assertEqual('nbd', action.parameters['to'])
            self.assertTrue(action.valid)
        uboot = [action for action in job.pipeline.actions if action.name == 'uboot-action'][0]
        overlay = [action for action in uboot.internal_pipeline.actions if action.name == 'bootloader-overlay'][0]
        for setenv in overlay.commands:
            if 'setenv nbdbasekargs' in setenv:
                self.assertIn('rw', setenv.split("'")[1])
                self.assertIn('${extraargs}', setenv.split("'")[1])
                self.assertEqual(3, len(setenv.split("'")))

    def test_transfer_media(self):
        """
        Test adding the overlay to existing rootfs
        """
        job = self.factory.create_bbb_job('sample_jobs/uboot-ramdisk-inline-commands.yaml')
        job.validate()
        description_ref = self.pipeline_reference('uboot-ramdisk-inline-commands.yaml', job=job)
        self.assertEqual(description_ref, job.pipeline.describe(False))
        uboot = [action for action in job.pipeline.actions if action.name == 'uboot-action'][0]
        retry = [action for action in uboot.internal_pipeline.actions if action.name == 'uboot-retry'][0]
        transfer = [action for action in retry.internal_pipeline.actions if action.name == 'overlay-unpack'][0]
        self.assertIn('transfer_overlay', transfer.parameters)
        self.assertIn('download_command', transfer.parameters['transfer_overlay'])
        self.assertIn('unpack_command', transfer.parameters['transfer_overlay'])

    def test_download_action(self):
        job = self.factory.create_bbb_job('sample_jobs/uboot.yaml')
        for action in job.pipeline.actions:
            action.validate()
            self.assertTrue(action.valid)
        job.validate()
        self.assertEqual(job.pipeline.errors, [])
        deploy = None
        overlay = None
        extract = None
        for action in job.pipeline.actions:
            if action.name == 'tftp-deploy':
                deploy = action
        if deploy:
            for action in deploy.internal_pipeline.actions:
                if action.name == 'prepare-tftp-overlay':
                    overlay = action
        if overlay:
            for action in overlay.internal_pipeline.actions:
                if action.name == 'extract-nfsrootfs':
                    extract = action
        test_dir = overlay.get_namespace_data(action='test', label='results', key='lava_test_results_dir')
        self.assertIsNotNone(test_dir)
        self.assertIn('/lava-', test_dir)
        self.assertIsNotNone(extract)
        self.assertEqual(extract.timeout.duration, 240)

    def test_reset_actions(self):
        job = self.factory.create_bbb_job('sample_jobs/uboot.yaml')
        uboot_action = None
        uboot_retry = None
        reset_action = None
        for action in job.pipeline.actions:
            action.validate()
            self.assertTrue(action.valid)
            if action.name == 'uboot-action':
                uboot_action = action
        names = [r_action.name for r_action in uboot_action.internal_pipeline.actions]
        self.assertIn('connect-device', names)
        self.assertIn('uboot-retry', names)
        for action in uboot_action.internal_pipeline.actions:
            if action.name == 'uboot-retry':
                uboot_retry = action
        names = [r_action.name for r_action in uboot_retry.internal_pipeline.actions]
        self.assertIn('reset-device', names)
        self.assertIn('bootloader-interrupt', names)
        self.assertIn('expect-shell-connection', names)
        self.assertIn('bootloader-commands', names)
        for action in uboot_retry.internal_pipeline.actions:
            if action.name == 'reset-device':
                reset_action = action
        names = [r_action.name for r_action in reset_action.internal_pipeline.actions]
        self.assertIn('pdu-reboot', names)

    def test_secondary_media(self):
        """
        Test UBootSecondaryMedia validation
        """
        job_parser = JobParser()
        (rendered, _) = self.factory.create_device('cubie1.jinja2')
        cubie = NewDevice(yaml.load(rendered))
        sample_job_file = os.path.join(os.path.dirname(__file__), 'sample_jobs/cubietruck-removable.yaml')
        sample_job_data = open(sample_job_file)
        job = job_parser.parse(sample_job_data, cubie, 4212, None, "")
        job.logger = DummyLogger()
        job.validate()
        sample_job_data.close()
        uboot_action = [action for action in job.pipeline.actions if action.name == 'uboot-action' and action.parameters['namespace'] == 'boot2'][0]
        u_boot_media = [action for action in uboot_action.internal_pipeline.actions if action.name == 'uboot-from-media' and action.parameters['namespace'] == 'boot2'][0]
        self.assertIsInstance(u_boot_media, UBootSecondaryMedia)
        self.assertEqual([], u_boot_media.errors)
        self.assertEqual(u_boot_media.parameters['kernel'], '/boot/vmlinuz-3.16.0-4-armmp-lpae')
        self.assertEqual(u_boot_media.parameters['kernel'], u_boot_media.get_namespace_data(
            action='download-action', label='file', key='kernel'))
        self.assertEqual(u_boot_media.parameters['ramdisk'], u_boot_media.get_namespace_data(
            action='compress-ramdisk', label='file', key='ramdisk'))
        self.assertEqual(u_boot_media.parameters['dtb'], u_boot_media.get_namespace_data(
            action='download-action', label='file', key='dtb'))
        # use the base class name so that uboot-from-media can pick up the value reliably.
        self.assertEqual(u_boot_media.parameters['root_uuid'], u_boot_media.get_namespace_data(
            action='bootloader-from-media', label='uuid', key='root'))
        device = u_boot_media.get_namespace_data(action='storage-deploy', label='u-boot', key='device')
        self.assertIsNotNone(device)
        part_reference = '%s:%s' % (
            job.device['parameters']['media']['usb'][device]['device_id'],
            u_boot_media.parameters['boot_part']
        )
        self.assertEqual(part_reference, u_boot_media.get_namespace_data(
            action=u_boot_media.name, label='uuid', key='boot_part'))
        self.assertEqual(part_reference, "0:1")

    def test_xz_nfs(self):
        job = self.factory.create_bbb_job('sample_jobs/uboot-nfs.yaml')
        # this job won't validate as the .xz nfsrootfs URL is a fiction
        self.assertRaises(JobError, job.validate)
        tftp_deploy = [action for action in job.pipeline.actions if action.name == 'tftp-deploy'][0]
        prepare = [action for action in tftp_deploy.internal_pipeline.actions if action.name == 'prepare-tftp-overlay'][0]
        nfs = [action for action in prepare.internal_pipeline.actions if action.name == 'extract-nfsrootfs'][0]
        self.assertIn('compression', nfs.parameters['nfsrootfs'])
        self.assertEqual(nfs.parameters['nfsrootfs']['compression'], 'xz')

    def test_prefix(self):
        job = self.factory.create_bbb_job('sample_jobs/bbb-skip-install.yaml')
        job.validate()
        tftp_deploy = [action for action in job.pipeline.actions if action.name == 'tftp-deploy'][0]
        prepare = [action for action in tftp_deploy.internal_pipeline.actions if action.name == 'prepare-tftp-overlay'][0]
        nfs = [action for action in prepare.internal_pipeline.actions if action.name == 'extract-nfsrootfs'][0]
        self.assertIn('prefix', nfs.parameters['nfsrootfs'])
        self.assertEqual(nfs.parameters['nfsrootfs']['prefix'], 'jessie/')
        self.assertEqual(nfs.param_key, 'nfsrootfs')

    def test_zcu102(self):
        job = self.factory.create_zcu102_job('sample_jobs/zcu102-ramdisk.yaml')
        job.validate()
        self.assertEqual(job.pipeline.errors, [])
        description_ref = self.pipeline_reference('zcu102-ramdisk.yaml', job=job)
        self.assertEqual(description_ref, job.pipeline.describe(False))

    def test_imx8m(self):
        job = self.factory.create_job('imx8m-01.jinja2', 'sample_jobs/imx8m.yaml')
        self.assertIsNotNone(job)
        job.validate()
        self.assertEqual(job.pipeline.errors, [])
        description_ref = self.pipeline_reference('imx8m.yaml', job=job)
        self.assertEqual(description_ref, job.pipeline.describe(False))
        deploy = [action for action in job.pipeline.actions if action.name == 'fastboot-deploy'][0]
        fastboot = [action for action in deploy.internal_pipeline.actions if action.name == 'uboot-enter-fastboot'][0]
        bootloader = [action for action in fastboot.internal_pipeline.actions if action.name == 'bootloader-interrupt'][0]
        self.assertEqual('u-boot', bootloader.method)


class TestKernelConversion(StdoutTestCase):

    def setUp(self):
        logger = logging.getLogger('dispatcher')
        logger.disabled = True
        logger.propagate = False
        logger = logging.getLogger('lava-dispatcher')
        logger.disabled = True
        logger.propagate = False
        self.device = NewDevice(os.path.join(os.path.dirname(__file__), '../devices/bbb-01.yaml'))
        bbb_yaml = os.path.join(os.path.dirname(__file__), 'sample_jobs/uboot-ramdisk.yaml')
        with open(bbb_yaml) as sample_job_data:
            self.base_data = yaml.load(sample_job_data)
        self.deploy_block = [block for block in self.base_data['actions'] if 'deploy' in block][0]['deploy']
        self.boot_block = [block for block in self.base_data['actions'] if 'boot' in block][0]['boot']
        self.parser = JobParser()

    def test_zimage_bootz(self):
        self.deploy_block['kernel']['type'] = 'zimage'
        job = self.parser.parse(yaml.dump(self.base_data), self.device, 4212, None, "")
        job.logger = DummyLogger()
        job.validate()
        deploy = [action for action in job.pipeline.actions if action.name == 'tftp-deploy'][0]
        overlay = [action for action in deploy.internal_pipeline.actions if action.name == 'prepare-tftp-overlay'][0]
        prepare = [action for action in overlay.internal_pipeline.actions if action.name == 'prepare-kernel'][0]
        uboot_prepare = [action for action in prepare.internal_pipeline.actions if action.name == 'uboot-prepare-kernel'][0]
        self.assertEqual('zimage', uboot_prepare.kernel_type)
        self.assertEqual('bootz', uboot_prepare.bootcommand)
        self.assertFalse(uboot_prepare.mkimage_conversion)

    def test_image(self):
        self.deploy_block['kernel']['type'] = 'image'
        job = self.parser.parse(yaml.dump(self.base_data), self.device, 4212, None, "")
        job.logger = DummyLogger()
        job.validate()
        deploy = [action for action in job.pipeline.actions if action.name == 'tftp-deploy'][0]
        overlay = [action for action in deploy.internal_pipeline.actions if action.name == 'prepare-tftp-overlay'][0]
        prepare = [action for action in overlay.internal_pipeline.actions if action.name == 'prepare-kernel'][0]
        uboot_prepare = [action for action in prepare.internal_pipeline.actions if action.name == 'uboot-prepare-kernel'][0]
        self.assertEqual('image', uboot_prepare.kernel_type)
        # bbb-01.yaml does not contain booti parameters, try to convert to a uImage
        self.assertEqual('bootm', uboot_prepare.bootcommand)
        self.assertTrue(uboot_prepare.mkimage_conversion)

    def test_uimage(self):
        self.deploy_block['kernel']['type'] = 'uimage'
        job = self.parser.parse(yaml.dump(self.base_data), self.device, 4212, None, "")
        job.logger = DummyLogger()
        job.validate()
        deploy = [action for action in job.pipeline.actions if action.name == 'tftp-deploy'][0]
        overlay = [action for action in deploy.internal_pipeline.actions if action.name == 'prepare-tftp-overlay'][0]
        prepare = [action for action in overlay.internal_pipeline.actions if action.name == 'prepare-kernel'][0]
        uboot_prepare = [action for action in prepare.internal_pipeline.actions if action.name == 'uboot-prepare-kernel'][0]
        self.assertEqual('uimage', uboot_prepare.kernel_type)
        self.assertEqual('bootm', uboot_prepare.bootcommand)
        self.assertFalse(uboot_prepare.mkimage_conversion)

    def test_zimage_nobootz(self):
        # drop bootz from the device for this part of the test
        del self.device['parameters']['bootz']
        self.deploy_block['kernel']['type'] = 'zimage'
        job = self.parser.parse(yaml.dump(self.base_data), self.device, 4212, None, "")
        job.logger = DummyLogger()
        job.validate()
        deploy = [action for action in job.pipeline.actions if action.name == 'tftp-deploy'][0]
        overlay = [action for action in deploy.internal_pipeline.actions if action.name == 'prepare-tftp-overlay'][0]
        prepare = [action for action in overlay.internal_pipeline.actions if action.name == 'prepare-kernel'][0]
        uboot_prepare = [action for action in prepare.internal_pipeline.actions if action.name == 'uboot-prepare-kernel'][0]
        self.assertEqual('zimage', uboot_prepare.kernel_type)
        self.assertEqual('bootm', uboot_prepare.bootcommand)
        self.assertTrue(uboot_prepare.mkimage_conversion)

    def test_uimage_boot_type(self):
        # uimage in boot type
        del self.deploy_block['kernel']['type']
        self.boot_block['type'] = 'bootm'
        job = self.parser.parse(yaml.dump(self.base_data), self.device, 4212, None, "")
        job.logger = DummyLogger()
        job.validate()
        deploy = [action for action in job.pipeline.actions if action.name == 'tftp-deploy'][0]
        overlay = [action for action in deploy.internal_pipeline.actions if action.name == 'prepare-tftp-overlay'][0]
        self.assertNotIn('uboot-prepare-kernel', [action.name for action in overlay.internal_pipeline.actions])


class TestOverlayCommands(StdoutTestCase):  # pylint: disable=too-many-public-methods

    def setUp(self):
        super().setUp()
        self.factory = UBootFactory()

    def test_combined_ramdisk_nfs(self):
        job = self.factory.create_bbb_job('sample_jobs/bbb-ramdisk-nfs.yaml')
        tftp_deploy = [action for action in job.pipeline.actions if action.name == 'tftp-deploy'][0]
        prepare = [action for action in tftp_deploy.internal_pipeline.actions if action.name == 'prepare-tftp-overlay'][0]
        nfs = [action for action in prepare.internal_pipeline.actions if action.name == 'extract-nfsrootfs'][0]
        ramdisk = [action for action in prepare.internal_pipeline.actions if action.name == 'extract-overlay-ramdisk'][0]
        modules = [action for action in prepare.internal_pipeline.actions if action.name == 'extract-modules'][0]
        overlay = [action for action in prepare.internal_pipeline.actions if action.name == 'apply-overlay-tftp'][0]
        self.assertIsNotNone(ramdisk.parameters.get('ramdisk'))
        self.assertIsNotNone(ramdisk.parameters['ramdisk'].get('url'))
        self.assertIsNotNone(ramdisk.parameters['ramdisk'].get('compression'))
        self.assertTrue(ramdisk.parameters['ramdisk'].get('install_modules', True))
        self.assertTrue(ramdisk.parameters['ramdisk'].get('install_overlay', True))
        self.assertIsNotNone(modules.parameters.get('ramdisk'))
        self.assertIsNotNone(modules.parameters.get('nfsrootfs'))
        self.assertIsNotNone(nfs.parameters.get('nfsrootfs'))
        self.assertIsNotNone(overlay.parameters.get('nfsrootfs'))
        self.assertIsNotNone(overlay.parameters.get('ramdisk'))

    def test_ramdisk_nfs_nomodules(self):
        job = self.factory.create_bbb_job('sample_jobs/bbb-uinitrd-nfs.yaml')
        tftp_deploy = [action for action in job.pipeline.actions if action.name == 'tftp-deploy'][0]
        prepare = [action for action in tftp_deploy.internal_pipeline.actions if action.name == 'prepare-tftp-overlay'][0]
        nfs = [action for action in prepare.internal_pipeline.actions if action.name == 'extract-nfsrootfs'][0]
        ramdisk = [action for action in prepare.internal_pipeline.actions if action.name == 'extract-overlay-ramdisk'][0]
        modules = [action for action in prepare.internal_pipeline.actions if action.name == 'extract-modules'][0]
        overlay = [action for action in prepare.internal_pipeline.actions if action.name == 'apply-overlay-tftp'][0]
        self.assertIsNotNone(ramdisk.parameters.get('ramdisk'))
        self.assertIsNotNone(ramdisk.parameters['ramdisk'].get('url'))
        self.assertIsNone(ramdisk.parameters['ramdisk'].get('compression'))
        self.assertFalse(ramdisk.parameters['ramdisk'].get('install_overlay', True))
        self.assertIsNotNone(modules.parameters.get('ramdisk'))
        self.assertIsNotNone(modules.parameters.get('nfsrootfs'))
        self.assertIsNotNone(nfs.parameters.get('nfsrootfs'))
        self.assertIsNotNone(overlay.parameters.get('nfsrootfs'))
        self.assertIsNotNone(overlay.parameters.get('ramdisk'))