aboutsummaryrefslogtreecommitdiff
path: root/lava_scheduler_app/tests/test_menus.py
blob: 5d20e1355e18eb6221609f99e64c97dd579d8189 (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
import os
import yaml
from lava_scheduler_app.tests.test_pipeline import YamlFactory
from lava_scheduler_app.tests.test_submission import TestCaseWithFactory


class YamlMenuFactory(YamlFactory):

    def make_fake_mustang_device(self, hostname='fakemustang1'):  # pylint: disable=no-self-use
        assert hostname == 'fakemustang1'

    def make_job_data(self, actions=None, **kw):
        sample_job_file = os.path.join(os.path.dirname(__file__), 'sample_jobs', 'mustang-menu-ramdisk.yaml')
        with open(sample_job_file, 'r') as test_support:
            data = yaml.safe_load(test_support)
        data.update(kw)
        return data


class TestPipelineMenu(TestCaseWithFactory):  # pylint: disable=too-many-ancestors
    """
    Test the building and override support of pipeline menus from submission YAML
    """
    def setUp(self):
        super().setUp()
        self.factory = YamlMenuFactory()
        self.device_type = self.factory.make_device_type(name='mustang-uefi')

    def test_make_job_yaml(self):
        data = yaml.safe_load(self.factory.make_job_yaml())
        self.assertIn('device_type', data)
        self.assertNotIn('timeout', data)
        self.assertIn('timeouts', data)
        self.assertIn('job', data['timeouts'])
        self.assertIn('priority', data)

    def test_menu_device(self):
        job_ctx = {}
        hostname = 'fakemustang2'
        device = self.factory.make_device(self.device_type, hostname)
        self.assertEqual(device.device_type.name, 'mustang-uefi')

        config = device.load_configuration()
        self.assertIsNotNone(config)
        self.assertNotIn('parameters', config)
        self.assertIsNotNone(config['actions']['boot']['methods']['uefi-menu']['nfs'])
        menu_data = config['actions']['boot']['methods']['uefi-menu']['nfs']
        tftp_menu = [item for item in menu_data if 'items' in item['select'] and 'TFTP' in item['select']['items'][0]][0]
        tftp_mac = "52:54:00:12:34:59"
        # value from device dictionary correctly replaces device type default
        self.assertIn(tftp_mac, tftp_menu['select']['items'][0])

    def test_menu_context(self):
        job_ctx = {
            'menu_early_printk': '',
            'menu_interrupt_prompt': 'The default boot selection will start in',
            'base_ip_args': 'ip=dhcp'
        }
        hostname = 'fakemustang2'
        device = self.factory.make_device(self.device_type, hostname)
        config = device.load_configuration(job_ctx)
        self.assertIsNotNone(config)
        self.assertIsNotNone(config['actions']['boot']['methods']['uefi-menu']['nfs'])
        menu_data = config['actions']['boot']['methods']['uefi-menu']
        # assert that menu_interrupt_prompt replaces the default 'The default boot selection will start in'
        self.assertEqual(
            menu_data['parameters']['interrupt_prompt'],
            job_ctx['menu_interrupt_prompt']
        )
        # assert that menu_early_printk replaces the default earlyprintk default
        self.assertEqual(
            [e for e in menu_data['nfs'] if 'enter' in e['select'] and 'new Entry' in e['select']['wait']][0]['select']['enter'],
            'console=ttyS0,115200  debug root=/dev/nfs rw nfsroot={NFS_SERVER_IP}:{NFSROOTFS},tcp,hard,intr ip=dhcp'
        )