diff options
| author | Neil Williams <neil.williams@linaro.org> | 2018-09-28 11:09:48 +0100 |
|---|---|---|
| committer | Rémi Duraffort <remi.duraffort@linaro.org> | 2018-10-01 06:33:56 +0000 |
| commit | 146e143d4a98aede60c68e975d37ec05f6947c29 (patch) | |
| tree | 97c6e25e87cd80ac718d8a161f8b3a561e6d1807 | |
| parent | 01a3b0c0b6fc00fb67525fed814b4e8a6d6fb83d (diff) | |
Apply yaml.safe_load to unit tests
Ensure unit tests are testing the same behaviour
as the rest of the source code and
also reduce the amount of noise in the SAST report
so that genuine issues are easier to find.
Signed-off-by: Neil Williams <neil.williams@linaro.org>
31 files changed, 264 insertions, 402 deletions
diff --git a/doc/v2/device-integration.rst b/doc/v2/device-integration.rst index 7dfe901b3..62be2ac82 100644 --- a/doc/v2/device-integration.rst +++ b/doc/v2/device-integration.rst @@ -625,9 +625,7 @@ files to see how this is done. e.g. for QEMU from {% set mac_addr = 'DE:AD:BE:EF:28:01' %} {% set memory = 512 %}""" job_ctx = {'arch': 'amd64'} - test_template = prepare_jinja_template('staging-qemu-01', data) - rendered = test_template.render(**job_ctx) - template_dict = yaml.safe_load(rendered) + template_dict = prepare_jinja_template('staging-qemu-01', data, job_ctx=job_ctx, raw=False) self.assertEqual( 'c', template_dict['actions']['boot']['methods']['qemu']['parameters']['boot_options']['boot_order'] diff --git a/doc/v2/examples/source/query-results.py b/doc/v2/examples/source/query-results.py index baee5264b..a01f8c9af 100755 --- a/doc/v2/examples/source/query-results.py +++ b/doc/v2/examples/source/query-results.py @@ -63,7 +63,7 @@ def main(args): return 0 print("Job, Type, Message, Time") for result in data: - job_lava = yaml.load(connection.results.get_testcase_results_yaml(result['id'], 'lava', 'job'))[0] + job_lava = yaml.safe_load(connection.results.get_testcase_results_yaml(result['id'], 'lava', 'job'))[0] job_id = job_lava['job'] logged = job_lava['logged'] if result['status'] == INCOMPLETE: diff --git a/lava_dispatcher/test/test_basic.py b/lava_dispatcher/test/test_basic.py index dab7f1c2c..751ac745f 100644 --- a/lava_dispatcher/test/test_basic.py +++ b/lava_dispatcher/test/test_basic.py @@ -66,7 +66,7 @@ class StdoutTestCase(unittest.TestCase): # pylint: disable=too-many-public-meth with open(y_file, 'w') as describe: yaml.dump(job.pipeline.describe(False), describe) with open(y_file, 'r') as f_ref: - return yaml.load(f_ref) + return yaml.safe_load(f_ref) class TestAction(StdoutTestCase): # pylint: disable=too-many-public-methods @@ -191,7 +191,7 @@ class Factory: """ rendered = self.render_device_dictionary(hostname, data, job_ctx) try: - ret = validate_device(yaml.load(rendered)) + ret = validate_device(yaml.safe_load(rendered)) except (SubmissionException, ConfigurationError) as exc: print('#######') print(rendered) @@ -217,7 +217,7 @@ class Factory: def create_custom_job(self, template, job_data): job_ctx = job_data.get('context') (data, device_dict) = self.create_device(template, job_ctx) - device = NewDevice(yaml.load(data)) + device = NewDevice(yaml.safe_load(data)) if self.debug: print('####### Device configuration #######') print(data) @@ -236,7 +236,7 @@ class Factory: def create_job(self, template, filename): y_file = os.path.join(os.path.dirname(__file__), filename) with open(y_file) as sample_job_data: - job_data = yaml.load(sample_job_data.read()) + job_data = yaml.safe_load(sample_job_data.read()) return self.create_custom_job(template, job_data) def create_fake_qemu_job(self): @@ -247,7 +247,7 @@ class Factory: Custom function to allow for extra exception handling. """ (data, device_dict) = self.create_device('kvm01.jinja2') - device = NewDevice(yaml.load(data)) + device = NewDevice(yaml.safe_load(data)) if self.debug: print('####### Device configuration #######') print(data) @@ -430,11 +430,11 @@ class TestPipeline(StdoutTestCase): # pylint: disable=too-many-public-methods self.assertEqual(job.compatibility, DeployImages.compatibility) kvm_yaml = os.path.join(os.path.dirname(__file__), 'sample_jobs/kvm.yaml') with open(kvm_yaml, 'r') as kvm_yaml: - job_def = yaml.load(kvm_yaml) + job_def = yaml.safe_load(kvm_yaml) job_def['compatibility'] = job.compatibility parser = JobParser() (rendered, data) = factory.create_device('kvm01.jinja2') - device = yaml.load(rendered) + device = yaml.safe_load(rendered) try: job = parser.parse(yaml.dump(job_def), device, 4212, None, "") except NotImplementedError: diff --git a/lava_dispatcher/test/test_connections.py b/lava_dispatcher/test/test_connections.py index 2df1854fe..e9f5f18e0 100644 --- a/lava_dispatcher/test/test_connections.py +++ b/lava_dispatcher/test/test_connections.py @@ -288,7 +288,7 @@ commands: uart0: connect: telnet localhost 7020 """ - data = yaml.load(device_data) + data = yaml.safe_load(device_data) self.assertIn('commands', data) self.assertIn('connections', data['commands']) self.assertNotIn('connect', data['commands']) @@ -354,7 +354,7 @@ class TestTimeouts(StdoutTestCase): """ y_file = os.path.join(os.path.dirname(__file__), './sample_jobs/uboot-ramdisk.yaml') with open(y_file, 'r') as uboot_ramdisk: - data = yaml.load(uboot_ramdisk) + data = yaml.safe_load(uboot_ramdisk) data['timeouts']['connection'] = {'seconds': 20} job = self.factory.create_custom_job('bbb-01.jinja2', data) for action in job.pipeline.actions: @@ -380,7 +380,7 @@ class TestTimeouts(StdoutTestCase): """ y_file = os.path.join(os.path.dirname(__file__), './sample_jobs/uboot-ramdisk.yaml') with open(y_file, 'r') as uboot_ramdisk: - data = yaml.load(uboot_ramdisk) + data = yaml.safe_load(uboot_ramdisk) connection_timeout = Timeout.parse(data['timeouts']['connections']['lava-test-shell']) data['timeouts']['actions']['uboot-retry'] = {} data['timeouts']['actions']['uboot-retry']['seconds'] = 90 diff --git a/lava_dispatcher/test/test_defs.py b/lava_dispatcher/test/test_defs.py index 535fa9e65..a5fd7a1e6 100644 --- a/lava_dispatcher/test/test_defs.py +++ b/lava_dispatcher/test/test_defs.py @@ -129,11 +129,11 @@ class TestDefinitionHandlers(StdoutTestCase): # pylint: disable=too-many-public testdef.validate() self.assertEqual([], testdef.errors) (rendered, _) = self.factory.create_device('kvm01.jinja2') - device = yaml.load(rendered) + device = yaml.safe_load(rendered) kvm_yaml = os.path.join(os.path.dirname(__file__), 'sample_jobs/kvm.yaml') parser = JobParser() with open(kvm_yaml, 'r') as sample_job_data: - content = yaml.load(sample_job_data) + content = yaml.safe_load(sample_job_data) data = [block['test'] for block in content['actions'] if 'test' in block][0] definitions = [block for block in data['definitions'] if 'path' in block][0] definitions['name'] = 'smoke tests' @@ -412,7 +412,7 @@ class TestDefinitions(StdoutTestCase): def test_pattern(self): self.assertTrue(os.path.exists(self.testdef)) with open(self.testdef, 'r') as par: - params = yaml.load(par) + params = yaml.safe_load(par) self.assertIn('parse', params.keys()) line = 'test1a: pass' self.assertEqual( @@ -429,7 +429,7 @@ class TestDefinitions(StdoutTestCase): # without a name from a testdef, the pattern is not valid. self.assertFalse(pattern.valid()) with open(self.testdef, 'r') as par: - params = yaml.load(par) + params = yaml.safe_load(par) pattern = PatternFixup(testdef=params, count=0) self.assertTrue(pattern.valid()) @@ -469,7 +469,7 @@ class TestDefinitions(StdoutTestCase): ) # fake up a run step with open(self.testdef, 'r') as par: - params = yaml.load(par) + params = yaml.safe_load(par) self.assertEqual( r'(?P<test_case_id>.*-*):\s+(?P<result>(pass|fail))', params['parse']['pattern']) @@ -509,7 +509,7 @@ test3a: skip \"test4a:\" \"unknown\" """ with open(self.testdef, 'r') as par: - params = yaml.load(par) + params = yaml.safe_load(par) pattern = params['parse']['pattern'] re_pat = re.compile(pattern, re.M) match = re.search(re_pat, data) diff --git a/lava_dispatcher/test/test_devices.py b/lava_dispatcher/test/test_devices.py index 5d3914a23..b440a20a9 100644 --- a/lava_dispatcher/test/test_devices.py +++ b/lava_dispatcher/test/test_devices.py @@ -39,7 +39,7 @@ class TestDeviceParser(StdoutTestCase): # pylint: disable=too-many-public-metho def test_new_device(self): factory = Factory() (rendered, _) = factory.create_device('kvm01.jinja2') - kvm01 = yaml.load(rendered) + kvm01 = yaml.safe_load(rendered) try: self.assertIsNotNone(kvm01['actions']) except Exception: @@ -113,16 +113,16 @@ class TestJobDeviceParameters(StdoutTestCase): # pylint: disable=too-many-publi def test_device_power(self): factory = Factory() (rendered, _) = factory.create_device('bbb-01.jinja2') - device = yaml.load(rendered) + device = yaml.safe_load(rendered) self.assertNotEqual(device['commands'].get('hard_reset', ''), '') (rendered, _) = factory.create_device('kvm01.jinja2') - device = yaml.load(rendered) + device = yaml.safe_load(rendered) self.assertNotIn('commands', device) def test_device_constants(self): factory = Factory() (rendered, _) = factory.create_device('bbb-01.jinja2') - device = NewDevice(yaml.load(rendered)) + device = NewDevice(yaml.safe_load(rendered)) self.assertIn('constants', device) self.assertEqual(device.get_constant('kernel-start-message'), "Linux version [0-9]") self.assertRaises(ConfigurationError, @@ -139,7 +139,7 @@ class TestDeviceEnvironment(StdoutTestCase): # pylint: disable=too-many-public- data = None job_parser = JobParser() (rendered, _) = factory.create_device('bbb-01.jinja2') - device = NewDevice(yaml.load(rendered)) + device = NewDevice(yaml.safe_load(rendered)) sample_job_file = os.path.join(os.path.dirname(__file__), 'sample_jobs/uboot-ramdisk.yaml') with open(sample_job_file) as sample_job_data: job = job_parser.parse( @@ -165,7 +165,7 @@ overrides: factory = Factory() job_parser = JobParser() (rendered, _) = factory.create_device('bbb-01.jinja2') - device = NewDevice(yaml.load(rendered)) + device = NewDevice(yaml.safe_load(rendered)) sample_job_file = os.path.join(os.path.dirname(__file__), 'sample_jobs/uboot-ramdisk.yaml') with open(sample_job_file) as sample_job_data: job = job_parser.parse( @@ -190,7 +190,7 @@ overrides: factory = Factory() job_parser = JobParser() (rendered, _) = factory.create_device('bbb-01.jinja2') - device = NewDevice(yaml.load(rendered)) + device = NewDevice(yaml.safe_load(rendered)) sample_job_file = os.path.join(os.path.dirname(__file__), 'sample_jobs/uboot-ramdisk.yaml') with open(sample_job_file) as sample_job_data: job = job_parser.parse( diff --git a/lava_dispatcher/test/test_grub.py b/lava_dispatcher/test/test_grub.py index 8bac6eb70..929cf1369 100644 --- a/lava_dispatcher/test/test_grub.py +++ b/lava_dispatcher/test/test_grub.py @@ -19,12 +19,10 @@ # with this program; if not, see <http://www.gnu.org/licenses>. -import os import yaml import unittest from lava_dispatcher.device import NewDevice -from lava_dispatcher.parser import JobParser -from lava_dispatcher.test.utils import DummyLogger, infrastructure_error, infrastructure_error_multi_paths +from lava_dispatcher.test.utils import infrastructure_error, infrastructure_error_multi_paths from lava_dispatcher.actions.boot.grub import GrubMainAction from lava_dispatcher.actions.boot import BootloaderCommandOverlay, BootloaderInterruptAction from lava_dispatcher.actions.deploy.tftp import TftpAction @@ -32,7 +30,7 @@ from lava_dispatcher.job import Job from lava_dispatcher.action import JobError, Pipeline from lava_dispatcher.test.test_basic import Factory, StdoutTestCase from lava_dispatcher.utils.network import dispatcher_ip -from lava_dispatcher.utils.filesystem import mkdtemp, tftpd_dir +from lava_dispatcher.utils.filesystem import tftpd_dir from lava_dispatcher.utils.strings import substitute @@ -145,7 +143,7 @@ class TestGrubAction(StdoutTestCase): # pylint: disable=too-many-public-methods } } (rendered, _) = self.factory.create_device('d02-01.jinja2') - device = NewDevice(yaml.load(rendered)) + device = NewDevice(yaml.safe_load(rendered)) job = Job(4212, parameters, None) job.device = device pipeline = Pipeline(job=job, parameters=parameters['actions']['boot']) diff --git a/lava_dispatcher/test/test_ipxe.py b/lava_dispatcher/test/test_ipxe.py index 35a1d009e..f2d1eb54b 100644 --- a/lava_dispatcher/test/test_ipxe.py +++ b/lava_dispatcher/test/test_ipxe.py @@ -20,7 +20,6 @@ import os -import sys import yaml import unittest from lava_dispatcher.device import NewDevice @@ -129,7 +128,7 @@ class TestBootloaderAction(StdoutTestCase): # pylint: disable=too-many-public-m } } (rendered, _) = self.factory.create_device('x86-01.jinja2') - device = NewDevice(yaml.load(rendered)) + device = NewDevice(yaml.safe_load(rendered)) job = Job(4212, parameters, None) job.device = device pipeline = Pipeline(job=job, parameters=parameters['actions']['boot']) @@ -229,12 +228,12 @@ class TestBootloaderAction(StdoutTestCase): # pylint: disable=too-many-public-m if action.name == 'expect-shell-connection'][0] check = expect.parameters (rendered, _) = self.factory.create_device('x86-01.jinja2') - device = NewDevice(yaml.load(rendered)) + device = NewDevice(yaml.safe_load(rendered)) extra_yaml = os.path.join(os.path.dirname(__file__), 'sample_jobs/ipxe.yaml') with open(extra_yaml) as data: sample_job_string = data.read() parser = JobParser() - sample_job_data = yaml.load(sample_job_string) + sample_job_data = yaml.safe_load(sample_job_string) boot = [item['boot'] for item in sample_job_data['actions'] if 'boot' in item][0] self.assertIsNotNone(boot) sample_job_string = yaml.dump(sample_job_data) diff --git a/lava_dispatcher/test/test_iso.py b/lava_dispatcher/test/test_iso.py index e10e7848e..90a54b868 100644 --- a/lava_dispatcher/test/test_iso.py +++ b/lava_dispatcher/test/test_iso.py @@ -33,7 +33,7 @@ class InstallerFactory(Factory): # pylint: disable=too-few-public-methods def create_qemu_installer_job(self): (rendered, _) = self.create_device('kvm01.jinja2') - device = NewDevice(yaml.load(rendered)) + device = NewDevice(yaml.safe_load(rendered)) sample_job_file = os.path.join(os.path.dirname(__file__), 'sample_jobs/qemu-debian-installer.yaml') parser = JobParser() try: @@ -109,7 +109,7 @@ class TestIsoJob(StdoutTestCase): test_retry = [action for action in self.job.pipeline.actions if action.name == 'lava-test-retry'][0] sample_job_file = os.path.join(os.path.dirname(__file__), 'sample_jobs/qemu-debian-installer.yaml') with open(sample_job_file, 'r') as jobdef: - data = yaml.load(jobdef) + data = yaml.safe_load(jobdef) testdata = [block['test'] for block in data['actions'] if 'test' in block][0] duration = (Timeout.parse(testdata['timeout'])) self.assertEqual( diff --git a/lava_dispatcher/test/test_kvm.py b/lava_dispatcher/test/test_kvm.py index b4dacb235..4b6548f05 100644 --- a/lava_dispatcher/test/test_kvm.py +++ b/lava_dispatcher/test/test_kvm.py @@ -329,11 +329,11 @@ class TestKVMInlineTestDeploy(StdoutTestCase): # pylint: disable=too-many-publi def test_extra_options(self): (rendered, _) = self.factory.create_device('kvm01.jinja2') - device = NewDevice(yaml.load(rendered)) + device = NewDevice(yaml.safe_load(rendered)) kvm_yaml = os.path.join(os.path.dirname(__file__), 'sample_jobs/kvm-inline.yaml') with open(kvm_yaml) as sample_job_data: - job_data = yaml.load(sample_job_data) - device['actions']['boot']['methods']['qemu']['parameters']['extra'] = yaml.load(""" + job_data = yaml.safe_load(sample_job_data) + device['actions']['boot']['methods']['qemu']['parameters']['extra'] = yaml.safe_load(""" - -smp - 1 - -global @@ -383,7 +383,7 @@ class TestKVMInlineTestDeploy(StdoutTestCase): # pylint: disable=too-many-publi yaml_file = os.path.join(location, '0/tests/0_smoke-tests-inline/inline/smoke-tests-basic.yaml') self.assertTrue(os.path.exists(yaml_file)) with open(yaml_file, 'r') as f_in: - testdef = yaml.load(f_in) + testdef = yaml.safe_load(f_in) expected_testdef = {'metadata': {'description': 'Basic system test command for Linaro Ubuntu images', 'devices': ['panda', 'panda-es', 'arndale', 'vexpress-a9', 'vexpress-tc2'], @@ -667,7 +667,7 @@ class TestChecksum(StdoutTestCase): def test_uboot_checksum(self): (rendered, _) = self.factory.create_device('bbb-01.jinja2') - device = NewDevice(yaml.load(rendered)) + device = NewDevice(yaml.safe_load(rendered)) bbb_yaml = os.path.join(os.path.dirname(__file__), 'sample_jobs/bbb-ramdisk-nfs.yaml') with open(bbb_yaml) as sample_job_data: parser = JobParser() diff --git a/lava_dispatcher/test/test_lavashell.py b/lava_dispatcher/test/test_lavashell.py index 2e402890c..9b1a66327 100644 --- a/lava_dispatcher/test/test_lavashell.py +++ b/lava_dispatcher/test/test_lavashell.py @@ -69,11 +69,11 @@ class TestDefinitionHandlers(StdoutTestCase): # pylint: disable=too-many-public def test_missing_handler(self): (rendered, _) = self.factory.create_device('kvm01.jinja2') - device = NewDevice(yaml.load(rendered)) + device = NewDevice(yaml.safe_load(rendered)) kvm_yaml = os.path.join(os.path.dirname(__file__), 'sample_jobs/kvm.yaml') parser = JobParser() with open(kvm_yaml) as sample_job_data: - data = yaml.load(sample_job_data) + data = yaml.safe_load(sample_job_data) data['actions'][2]['test']['definitions'][0]['from'] = 'unusable-handler' try: job = parser.parse(yaml.dump(data), device, 4212, None, "") diff --git a/lava_dispatcher/test/test_lxc.py b/lava_dispatcher/test/test_lxc.py index 68867a6b4..1a984aa37 100644 --- a/lava_dispatcher/test/test_lxc.py +++ b/lava_dispatcher/test/test_lxc.py @@ -146,7 +146,7 @@ class TestLxcWithDevices(StdoutTestCase): self.job.validate() lxc_yaml = os.path.join(os.path.dirname(__file__), 'sample_jobs/bbb-lxc.yaml') with open(lxc_yaml) as sample_job_data: - data = yaml.load(sample_job_data) + data = yaml.safe_load(sample_job_data) lxc_deploy = [action for action in self.job.pipeline.actions if action.name == 'lxc-deploy'][0] overlay = [action for action in lxc_deploy.internal_pipeline.actions if action.name == 'lava-overlay'][0] test_def = [action for action in overlay.internal_pipeline.actions if action.name == 'test-definition'][0] @@ -182,7 +182,7 @@ class TestLxcWithDevices(StdoutTestCase): self.assertEqual(test_actions[0]['test']['namespace'], 'probe') parser = JobParser() (rendered, _) = self.factory.create_device('bbb-01.jinja2') - device = NewDevice(yaml.load(rendered)) + device = NewDevice(yaml.safe_load(rendered)) job = parser.parse(yaml.dump(data), device, 4577, None, "") job.logger = DummyLogger() job.validate() @@ -211,10 +211,10 @@ class TestLxcWithDevices(StdoutTestCase): def test_lxc_without_lxctest(self): # pylint: disable=too-many-locals lxc_yaml = os.path.join(os.path.dirname(__file__), 'sample_jobs/bbb-lxc-notest.yaml') with open(lxc_yaml) as sample_job_data: - data = yaml.load(sample_job_data) + data = yaml.safe_load(sample_job_data) parser = JobParser() (rendered, _) = self.factory.create_device('bbb-01.jinja2') - device = NewDevice(yaml.load(rendered)) + device = NewDevice(yaml.safe_load(rendered)) job = parser.parse(yaml.dump(data), device, 4577, None, "") job.logger = DummyLogger() job.validate() diff --git a/lava_dispatcher/test/test_multi.py b/lava_dispatcher/test/test_multi.py index 5cce8ebe4..66a2a01b2 100644 --- a/lava_dispatcher/test/test_multi.py +++ b/lava_dispatcher/test/test_multi.py @@ -158,7 +158,7 @@ class TestMultiDefinition(StdoutTestCase): # pylint: disable=too-many-public-me 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-nfs.yaml') with open(bbb_yaml) as sample_job_data: - self.job_data = yaml.load(sample_job_data) + self.job_data = yaml.safe_load(sample_job_data) def test_multidefinition(self): block = [testblock['test'] for testblock in self.job_data['actions'] if 'test' in testblock][0] diff --git a/lava_dispatcher/test/test_multinode.py b/lava_dispatcher/test/test_multinode.py index b7d0ab8eb..14249d0c9 100644 --- a/lava_dispatcher/test/test_multinode.py +++ b/lava_dispatcher/test/test_multinode.py @@ -211,7 +211,7 @@ class TestMultinode(StdoutTestCase): # pylint: disable=too-many-public-methods for action in self.client_job.pipeline.actions: data = action.explode() data_str = yaml.dump(data) - yaml.load(data_str) + yaml.load(data_str) # nosec not suitable for safe_load def test_multinode_timeout(self): """ diff --git a/lava_dispatcher/test/test_recovery.py b/lava_dispatcher/test/test_recovery.py index 7137061f8..71cd0b047 100644 --- a/lava_dispatcher/test/test_recovery.py +++ b/lava_dispatcher/test/test_recovery.py @@ -53,7 +53,7 @@ class FastBootFactory(Factory): # pylint: disable=too-few-public-methods def create_hikey_bl_job(self, filename): (data, device_dict) = self.create_hikey_bl_device('hi6220-hikey-01') - device = NewDevice(yaml.load(data)) + device = NewDevice(yaml.safe_load(data)) self.validate_data('hi6220-hikey-01', device_dict) fastboot_yaml = os.path.join(os.path.dirname(__file__), filename) with open(fastboot_yaml) as sample_job_data: diff --git a/lava_dispatcher/test/test_removable.py b/lava_dispatcher/test/test_removable.py index 9d4ce8c8a..49fb16a52 100644 --- a/lava_dispatcher/test/test_removable.py +++ b/lava_dispatcher/test/test_removable.py @@ -41,8 +41,8 @@ class RemovableFactory(Factory): # pylint: disable=too-few-public-methods def create_job(self, sample_job, device_file): # pylint: disable=no-self-use device = NewDevice(os.path.join(os.path.dirname(__file__), device_file)) - yaml = os.path.join(os.path.dirname(__file__), sample_job) - with open(yaml) as sample_job_data: + j_yaml = os.path.join(os.path.dirname(__file__), sample_job) + with open(j_yaml) as sample_job_data: parser = JobParser() job = parser.parse(sample_job_data, device, 4212, None, "") job.logger = DummyLogger() @@ -60,7 +60,7 @@ class TestRemovable(StdoutTestCase): # pylint: disable=too-many-public-methods Test that the correct parameters have been set for the device """ (rendered, _) = self.factory.create_device('cubie2.jinja2') - cubie = NewDevice(yaml.load(rendered)) + cubie = NewDevice(yaml.safe_load(rendered)) self.assertIsNotNone(cubie['parameters']['media'].get('usb')) self.assertIsNotNone(cubie.get('commands')) self.assertIsNotNone(cubie.get('actions')) @@ -113,7 +113,7 @@ class TestRemovable(StdoutTestCase): # pylint: disable=too-many-public-methods Test that the job parameters match expected structure """ (rendered, _) = self.factory.create_device('cubie1.jinja2') - cubie = NewDevice(yaml.load(rendered)) + cubie = NewDevice(yaml.safe_load(rendered)) job = self._check_valid_job(cubie, 'cubietruck-removable.yaml') self._check_job_parameters(cubie, job, 'download') @@ -122,7 +122,7 @@ class TestRemovable(StdoutTestCase): # pylint: disable=too-many-public-methods Test that the job parameters with a writer tool match expected structure """ (rendered, _) = self.factory.create_device('cubie1.jinja2') - cubie = NewDevice(yaml.load(rendered)) + cubie = NewDevice(yaml.safe_load(rendered)) job = self._check_valid_job(cubie, 'cubietruck-removable-with-writer.yaml') self._check_job_parameters(cubie, job, 'writer') @@ -159,12 +159,12 @@ class TestRemovable(StdoutTestCase): # pylint: disable=too-many-public-methods def test_deployment(self): (rendered, _) = self.factory.create_device('cubie1.jinja2') - cubie = NewDevice(yaml.load(rendered)) + cubie = NewDevice(yaml.safe_load(rendered)) self._check_deployment(cubie, 'cubietruck-removable.yaml') def test_writer_deployment(self): (rendered, _) = self.factory.create_device('cubie1.jinja2') - cubie = NewDevice(yaml.load(rendered)) + cubie = NewDevice(yaml.safe_load(rendered)) self._check_deployment(cubie, 'cubietruck-removable-with-writer.yaml') def test_juno_deployment(self): @@ -243,7 +243,7 @@ class TestRemovable(StdoutTestCase): # pylint: disable=too-many-public-methods """ job_parser = JobParser() (rendered, _) = self.factory.create_device('bbb-01.jinja2') - bbb = NewDevice(yaml.load(rendered)) + bbb = NewDevice(yaml.safe_load(rendered)) sample_job_file = os.path.join(os.path.dirname(__file__), 'sample_jobs/uboot-ramdisk.yaml') with open(sample_job_file) as sample_job_data: job = job_parser.parse(sample_job_data, bbb, 4212, None, "") @@ -262,7 +262,7 @@ class TestRemovable(StdoutTestCase): # pylint: disable=too-many-public-methods """ job_parser = JobParser() (rendered, _) = self.factory.create_device('cubie1.jinja2') - cubie = NewDevice(yaml.load(rendered)) + cubie = NewDevice(yaml.safe_load(rendered)) sample_job_file = os.path.join(os.path.dirname(__file__), 'sample_jobs/cubietruck-removable.yaml') with open(sample_job_file) as sample_job_data: job = job_parser.parse(sample_job_data, cubie, 4212, None, "") diff --git a/lava_dispatcher/test/test_test_shell.py b/lava_dispatcher/test/test_test_shell.py index 472a69dd0..95b138b80 100644 --- a/lava_dispatcher/test/test_test_shell.py +++ b/lava_dispatcher/test/test_test_shell.py @@ -52,7 +52,7 @@ class TestPatterns(StdoutTestCase): self.assertEqual([], self.job.pipeline.errors) self.assertTrue(os.path.exists(self.testdef)) with open(self.testdef, 'r') as par: - params = yaml.load(par) + params = yaml.safe_load(par) self.assertIn('parse', params.keys()) line = 'test1a: pass' diff --git a/lava_dispatcher/test/test_uboot.py b/lava_dispatcher/test/test_uboot.py index 3aebcb1f1..0cf1f79d4 100644 --- a/lava_dispatcher/test/test_uboot.py +++ b/lava_dispatcher/test/test_uboot.py @@ -360,7 +360,7 @@ class TestUbootAction(StdoutTestCase): # pylint: disable=too-many-public-method """ job_parser = JobParser() (rendered, _) = self.factory.create_device('cubie1.jinja2') - cubie = NewDevice(yaml.load(rendered)) + cubie = NewDevice(yaml.safe_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, "") @@ -443,7 +443,7 @@ class TestKernelConversion(StdoutTestCase): 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.base_data = yaml.safe_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() diff --git a/lava_dispatcher/test/test_vland.py b/lava_dispatcher/test/test_vland.py index c30e9da1e..c57cf7473 100644 --- a/lava_dispatcher/test/test_vland.py +++ b/lava_dispatcher/test/test_vland.py @@ -42,12 +42,12 @@ class TestVland(StdoutTestCase): # pylint: disable=too-many-public-methods self.beta_filename = os.path.join(os.path.dirname(__file__), 'sample_jobs/bbb-group-vland-beta.yaml') self.factory = Factory() (rendered, _) = self.factory.create_device('bbb-01.jinja2') - self.device = NewDevice(yaml.load(rendered)) + self.device = NewDevice(yaml.safe_load(rendered)) self.job_id = "100" def test_file_structure(self): with open(self.filename) as yaml_data: - alpha_data = yaml.load(yaml_data) + alpha_data = yaml.safe_load(yaml_data) self.assertIn('protocols', alpha_data) self.assertTrue(VlandProtocol.accepts(alpha_data)) level_tuple = Protocol.select_all(alpha_data) @@ -126,7 +126,7 @@ class TestVland(StdoutTestCase): # pylint: disable=too-many-public-methods def test_configure(self): with open(self.filename) as yaml_data: - alpha_data = yaml.load(yaml_data) + alpha_data = yaml.safe_load(yaml_data) self.assertIn('protocols', alpha_data) self.assertTrue(VlandProtocol.accepts(alpha_data)) vprotocol = VlandProtocol(alpha_data, self.job_id) @@ -152,7 +152,7 @@ class TestVland(StdoutTestCase): # pylint: disable=too-many-public-methods self.assertIsNotNone(vprotocol.multinode_protocol) (rendered, _) = self.factory.create_device('bbb-01.jinja2') - bbb2 = NewDevice(yaml.load(rendered)) + bbb2 = NewDevice(yaml.safe_load(rendered)) bbb2['parameters']['interfaces']['eth0']['switch'] = '192.168.0.2' bbb2['parameters']['interfaces']['eth0']['port'] = '6' bbb2['parameters']['interfaces']['eth1']['switch'] = '192.168.0.2' @@ -175,7 +175,7 @@ class TestVland(StdoutTestCase): # pylint: disable=too-many-public-methods def test_job(self): with open(self.filename) as yaml_data: - alpha_data = yaml.load(yaml_data) + alpha_data = yaml.safe_load(yaml_data) self.assertIn('protocols', alpha_data) self.assertIn(VlandProtocol.name, alpha_data['protocols']) with open(self.filename) as sample_job_data: @@ -216,7 +216,7 @@ class TestVland(StdoutTestCase): # pylint: disable=too-many-public-methods def test_vland_overlay(self): with open(self.filename) as yaml_data: - alpha_data = yaml.load(yaml_data) + alpha_data = yaml.safe_load(yaml_data) for vlan_key, _ in alpha_data['protocols'][VlandProtocol.name].items(): alpha_data['protocols'][VlandProtocol.name][vlan_key] = {'tags': []} # removed tags from original job to simulate job where any interface tags will be acceptable @@ -240,7 +240,7 @@ class TestVland(StdoutTestCase): # pylint: disable=too-many-public-methods def test_job_no_tags(self): with open(self.filename) as yaml_data: - alpha_data = yaml.load(yaml_data) + alpha_data = yaml.safe_load(yaml_data) for vlan_key, _ in alpha_data['protocols'][VlandProtocol.name].items(): alpha_data['protocols'][VlandProtocol.name][vlan_key] = {'tags': []} # removed tags from original job to simulate job where any interface tags will be acceptable @@ -268,7 +268,7 @@ class TestVland(StdoutTestCase): # pylint: disable=too-many-public-methods def test_job_bad_tags(self): with open(self.filename) as yaml_data: - alpha_data = yaml.load(yaml_data) + alpha_data = yaml.safe_load(yaml_data) for vlan_key, _ in alpha_data['protocols'][VlandProtocol.name].items(): alpha_data['protocols'][VlandProtocol.name][vlan_key] = {'tags': ['spurious']} # replaced tags from original job to simulate job where an unsupported tag is specified @@ -283,7 +283,7 @@ class TestVland(StdoutTestCase): # pylint: disable=too-many-public-methods def test_primary_interface(self): with open(self.filename) as yaml_data: - alpha_data = yaml.load(yaml_data) + alpha_data = yaml.safe_load(yaml_data) for interface in self.device['parameters']['interfaces']: # jinja2 processing of tags: [] results in tags: if self.device['parameters']['interfaces'][interface]['tags'] == []: @@ -301,7 +301,7 @@ class TestVland(StdoutTestCase): # pylint: disable=too-many-public-methods # pylint: disable=protected-access def demo(self): with open(self.filename) as yaml_data: - alpha_data = yaml.load(yaml_data) + alpha_data = yaml.safe_load(yaml_data) vprotocol = VlandProtocol(alpha_data, 422) vprotocol.settings = vprotocol.read_settings() self.assertIn('port', vprotocol.settings) diff --git a/lava_scheduler_app/tests/test_api.py b/lava_scheduler_app/tests/test_api.py index 54c87729d..e2be72abc 100644 --- a/lava_scheduler_app/tests/test_api.py +++ b/lava_scheduler_app/tests/test_api.py @@ -6,18 +6,14 @@ from nose.tools import nottest from io import BytesIO as StringIO import xmlrpc.client from django.test.client import Client -from django.contrib.auth.models import Permission, User -from django.utils import timezone +from django.contrib.auth.models import Permission from lava_scheduler_app.dbutils import validate_yaml from lava_scheduler_app.models import ( - Device, DeviceType, - Tag, - TestJob, Alias, ) from lava_scheduler_app.schema import validate_submission, validate_device, SubmissionException -from lava_scheduler_app.tests.test_submission import ModelFactory, TestCaseWithFactory +from lava_scheduler_app.tests.test_submission import TestCaseWithFactory # pylint: disable=invalid-name @@ -139,7 +135,7 @@ class TestVoluptuous(unittest.TestCase): for filename in files: # some files are dispatcher-level test files, e.g. after the multinode split try: - yaml_data = yaml.load(open(os.path.join(path, filename), 'r')) + yaml_data = yaml.safe_load(open(os.path.join(path, filename), 'r')) except yaml.YAMLError as exc: raise RuntimeError("Decoding YAML job submission failed: %s." % exc) if filename in device_files: @@ -163,9 +159,9 @@ timeouts: action: minutes: 5 """ - self.assertRaises(SubmissionException, validate_submission, yaml.load(bad_submission)) + self.assertRaises(SubmissionException, validate_submission, yaml.safe_load(bad_submission)) try: - validate_submission(yaml.load(bad_submission)) + validate_submission(yaml.safe_load(bad_submission)) except SubmissionException as exc: # with more than one omission, which one gets mentioned is undefined self.assertIn('required key not provided', str(exc)) @@ -174,9 +170,9 @@ actions: - deploy: to: tmpfs """ - self.assertRaises(SubmissionException, validate_submission, yaml.load(bad_submission)) + self.assertRaises(SubmissionException, validate_submission, yaml.safe_load(bad_submission)) try: - validate_submission(yaml.load(bad_submission)) + validate_submission(yaml.safe_load(bad_submission)) except SubmissionException as exc: self.assertIn('required key not provided', str(exc)) # with more than one omission, which one gets mentioned is undefined @@ -184,20 +180,20 @@ actions: bad_submission += """ visibility: public """ - self.assertRaises(SubmissionException, validate_submission, yaml.load(bad_submission)) + self.assertRaises(SubmissionException, validate_submission, yaml.safe_load(bad_submission)) try: - validate_submission(yaml.load(bad_submission)) + validate_submission(yaml.safe_load(bad_submission)) except SubmissionException as exc: self.assertIn('required key not provided', str(exc)) self.assertIn('job_name', str(exc)) bad_submission += """ job_name: qemu-pipeline """ - self.assertTrue(validate_submission(yaml.load(bad_submission))) - bad_yaml = yaml.load(bad_submission) + self.assertTrue(validate_submission(yaml.safe_load(bad_submission))) + bad_yaml = yaml.safe_load(bad_submission) del bad_yaml['timeouts']['job'] try: - validate_submission(yaml.load(bad_submission)) + validate_submission(yaml.safe_load(bad_submission)) except SubmissionException as exc: self.assertIn('required key not provided', str(exc)) self.assertIn('job', str(exc)) @@ -207,14 +203,14 @@ notify: criteria: status: complete """ - self.assertTrue(validate_submission(yaml.load(bad_submission))) + self.assertTrue(validate_submission(yaml.safe_load(bad_submission))) bad_submission += """ compare: query: entity: testrunfilter """ self.assertRaises(SubmissionException, validate_yaml, - yaml.load(bad_submission)) + yaml.safe_load(bad_submission)) invalid_monitors_name_char_yaml_def = """ # Zephyr JOB definition @@ -256,7 +252,7 @@ actions: """ self.assertRaises(SubmissionException, validate_submission, - yaml.load(invalid_monitors_name_char_yaml_def)) + yaml.safe_load(invalid_monitors_name_char_yaml_def)) def test_compression_change(self): @@ -284,7 +280,7 @@ actions: location: http://test.com/baz """ try: - validate_submission(yaml.load(bad_submission)) + validate_submission(yaml.safe_load(bad_submission)) except SubmissionException as exc: self.assertIn('required key not provided', str(exc)) self.assertIn('dtb', str(exc)) @@ -313,7 +309,7 @@ actions: dtb: http://test.com/baz """ try: - validate_submission(yaml.load(bad_submission)) + validate_submission(yaml.safe_load(bad_submission)) except SubmissionException as exc: self.assertIn('expected a dictionary for dictionary value', str(exc)) self.assertIn('dtb', str(exc)) @@ -340,7 +336,7 @@ secrets: foo: bar username: secret """ - self.assertTrue(validate_submission(yaml.load(secrets))) + self.assertTrue(validate_submission(yaml.safe_load(secrets))) secrets = """ job_name: kvm-test visibility: public @@ -361,7 +357,7 @@ secrets: foo: bar username: secret """ - self.assertRaises(SubmissionException, validate_submission, yaml.load(secrets)) + self.assertRaises(SubmissionException, validate_submission, yaml.safe_load(secrets)) def test_multinode(self): # Without protocols @@ -375,7 +371,7 @@ timeouts: minutes: 5 actions: [] """ - self.assertTrue(validate_submission(yaml.load(data))) + self.assertTrue(validate_submission(yaml.safe_load(data))) data = """ job_name: test @@ -388,7 +384,7 @@ timeouts: actions: [] protocols: {} """ - self.assertTrue(validate_submission(yaml.load(data))) + self.assertTrue(validate_submission(yaml.safe_load(data))) # With a valid multinode protocol data = """ @@ -406,7 +402,7 @@ protocols: guest: {} host: {} """ - self.assertTrue(validate_submission(yaml.load(data))) + self.assertTrue(validate_submission(yaml.safe_load(data))) data = """ job_name: test @@ -425,7 +421,7 @@ protocols: expect_role: host host: {} """ - self.assertTrue(validate_submission(yaml.load(data))) + self.assertTrue(validate_submission(yaml.safe_load(data))) # invalid host_role or expect_role data = """ @@ -444,7 +440,7 @@ protocols: host_role: server host: {} """ - self.assertRaises(SubmissionException, validate_submission, yaml.load(data)) + self.assertRaises(SubmissionException, validate_submission, yaml.safe_load(data)) data = """ job_name: test @@ -463,7 +459,7 @@ protocols: expect_role: server host: {} """ - self.assertRaises(SubmissionException, validate_submission, yaml.load(data)) + self.assertRaises(SubmissionException, validate_submission, yaml.safe_load(data)) # host_role without expect_role data = """ @@ -482,7 +478,7 @@ protocols: host_role: host host: {} """ - self.assertRaises(SubmissionException, validate_submission, yaml.load(data)) + self.assertRaises(SubmissionException, validate_submission, yaml.safe_load(data)) # expect_role without host_role data = """ @@ -501,4 +497,4 @@ protocols: expect_role: host host: {} """ - self.assertRaises(SubmissionException, validate_submission, yaml.load(data)) + self.assertRaises(SubmissionException, validate_submission, yaml.safe_load(data)) diff --git a/lava_scheduler_app/tests/test_base_templates.py b/lava_scheduler_app/tests/test_base_templates.py index bf86f538e..632030ee1 100644 --- a/lava_scheduler_app/tests/test_base_templates.py +++ b/lava_scheduler_app/tests/test_base_templates.py @@ -19,14 +19,22 @@ CONFIG_PATH = os.path.abspath( "lava_scheduler_app", "tests", "devices")) -def prepare_jinja_template(hostname, jinja_data): +def prepare_jinja_template(hostname, jinja_data, job_ctx=None, raw=True): + if not job_ctx: + job_ctx = {} string_loader = jinja2.DictLoader({'%s.jinja2' % hostname: jinja_data}) path = os.path.dirname(CONFIG_PATH) type_loader = jinja2.FileSystemLoader([os.path.join(path, 'device-types')]) env = jinja2.Environment( # nosec - YAML, not HTML, no XSS scope. loader=jinja2.ChoiceLoader([string_loader, type_loader]), trim_blocks=True, autoescape=False) - return env.get_template("%s.jinja2" % hostname) + test_template = env.get_template("%s.jinja2" % hostname) + if raw: + return test_template + rendered = test_template.render(**job_ctx) + if not rendered: + return None + return yaml.safe_load(rendered) class BaseTemplate: @@ -36,7 +44,7 @@ class BaseTemplate: debug = False # set to True to see the YAML device config output system = False # set to True to debug the system templates - def render_device_dictionary_file(self, filename, job_ctx=None): + def render_device_dictionary_file(self, filename, job_ctx=None, raw=True): device = filename.replace('.jinja2', '') with open(os.path.join(os.path.dirname(__file__), 'devices', filename)) as input: data = input.read() @@ -46,9 +54,12 @@ class BaseTemplate: test_template = prepare_jinja_template(device, data) if not job_ctx: job_ctx = {} - return test_template.render(**job_ctx) + rendered = test_template.render(**job_ctx) + if raw: + return rendered + return yaml.safe_load(rendered) - def render_device_dictionary(self, hostname, data, job_ctx=None): + def render_device_dictionary(self, hostname, data, job_ctx=None, raw=True): if not job_ctx: job_ctx = {} test_template = prepare_jinja_template(hostname, data) @@ -57,12 +68,14 @@ class BaseTemplate: print('#######') print(rendered) print('#######') - return rendered + if raw: + return rendered + return yaml.safe_load(rendered) def validate_data(self, hostname, data, job_ctx=None): - rendered = self.render_device_dictionary(hostname, data, job_ctx) + rendered = self.render_device_dictionary(hostname, data, job_ctx, raw=True) try: - ret = validate_device(yaml.load(rendered, Loader=yaml.CLoader)) + ret = validate_device(yaml.load(rendered, Loader=yaml.CSafeLoader)) # nosec - safe_load implemented directly except SubmissionException as exc: print('#######') print(rendered) @@ -88,7 +101,7 @@ class TestBaseTemplates(BaseTemplate.BaseTemplateCases): try: test_template = env.from_string(data) rendered = test_template.render() - template_dict = yaml.load(rendered, Loader=yaml.CLoader) + template_dict = yaml.load(rendered, Loader=yaml.CLoader) # nosec - safe_load implemented directly validate_device(template_dict) except AssertionError as exc: self.fail("Template %s failed: %s" % (os.path.basename(template), exc)) @@ -109,7 +122,7 @@ class TestBaseTemplates(BaseTemplate.BaseTemplateCases): data += "{% set connection_command = 'telnet calvin 6080' %}" test_template = env.from_string(data) rendered = test_template.render() - template_dict = yaml.load(rendered, Loader=yaml.CLoader) + template_dict = yaml.load(rendered, Loader=yaml.CSafeLoader) # nosec - safe_load implemented directly validate_device(template_dict) self.assertIn('connect', template_dict['commands']) self.assertNotIn( @@ -121,7 +134,7 @@ class TestBaseTemplates(BaseTemplate.BaseTemplateCases): data += "{% set connection_tags = {'uart1': ['primary']} %}" test_template = env.from_string(data) rendered = test_template.render() - template_dict = yaml.load(rendered, Loader=yaml.CLoader) + template_dict = yaml.load(rendered, Loader=yaml.CSafeLoader) # nosec - safe_load implemented directly validate_device(template_dict) self.assertNotIn('connect', template_dict['commands']) self.assertIn( @@ -151,9 +164,7 @@ class TestBaseTemplates(BaseTemplate.BaseTemplateCases): {% set fastboot_serial_number = 'R42D300FRYP' %} {% set connection_command = 'adb -s ' + adb_serial_number +' shell' %} """ - test_template = prepare_jinja_template('nexus4-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('nexus4-01', data, raw=False) self.assertEqual( 'adb -s R42D300FRYP shell', template_dict['commands']['connect'] @@ -169,9 +180,7 @@ class TestBaseTemplates(BaseTemplate.BaseTemplateCases): def test_console_baud(self): data = """{% extends 'beaglebone-black.jinja2' %}""" - test_template = prepare_jinja_template('bbb-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('bbb-01', data, raw=False) self.assertIn('u-boot', template_dict['actions']['boot']['methods']) self.assertIn('nfs', template_dict['actions']['boot']['methods']['u-boot']) commands = template_dict['actions']['boot']['methods']['u-boot']['nfs']['commands'] @@ -180,9 +189,7 @@ class TestBaseTemplates(BaseTemplate.BaseTemplateCases): continue self.assertIn('console=ttyO0,115200n8', command) data = """{% extends 'base-uboot.jinja2' %}""" - test_template = prepare_jinja_template('base-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('base-01', data, raw=False) self.assertIn('u-boot', template_dict['actions']['boot']['methods']) self.assertIn('nfs', template_dict['actions']['boot']['methods']['u-boot']) commands = template_dict['actions']['boot']['methods']['u-boot']['nfs']['commands'] @@ -206,7 +213,7 @@ class TestBaseTemplates(BaseTemplate.BaseTemplateCases): self.assertRaises( SubmissionException, validate_device, - yaml.load(device_dict) + yaml.safe_load(device_dict) ) def test_primary_connection_power_commands_empty_ssh_host(self): # pylint: disable=invalid-name @@ -217,7 +224,7 @@ class TestBaseTemplates(BaseTemplate.BaseTemplateCases): {% set connection_command = 'telnet localhost 7302' %} {% set ssh_host = '' %}""" device_dict = self.render_device_dictionary('staging-x86-01', data) - self.assertTrue(validate_device(yaml.load(device_dict))) + self.assertTrue(validate_device(yaml.safe_load(device_dict))) def test_primary_connection_power_commands(self): # pylint: disable=invalid-name data = """{% extends 'x86.jinja2' %} @@ -226,11 +233,10 @@ class TestBaseTemplates(BaseTemplate.BaseTemplateCases): {% set power_on_command = '/usr/bin/pduclient --command on' %} {% set connection_command = 'telnet localhost 7302' %}""" device_dict = self.render_device_dictionary('staging-x86-01', data) - self.assertTrue(validate_device(yaml.load(device_dict))) + self.assertTrue(validate_device(yaml.safe_load(device_dict))) def test_pexpect_spawn_window(self): - rendered = self.render_device_dictionary_file('hi6220-hikey-01.jinja2') - template_dict = yaml.load(rendered) + template_dict = self.render_device_dictionary_file('hi6220-hikey-01.jinja2', raw=False) self.assertIsNotNone(template_dict['constants']) self.assertIn('spawn_maxread', template_dict['constants']) self.assertIsInstance(template_dict['constants']['spawn_maxread'], str) @@ -238,8 +244,7 @@ class TestBaseTemplates(BaseTemplate.BaseTemplateCases): def test_test_shell_constants(self): job_ctx = {} - rendered = self.render_device_dictionary_file('hi6220-hikey-01.jinja2', job_ctx=job_ctx) - template_dict = yaml.load(rendered) + template_dict = self.render_device_dictionary_file('hi6220-hikey-01.jinja2', job_ctx=job_ctx, raw=False) self.assertIsNotNone(template_dict['constants']) self.assertIn('posix', template_dict['constants']) self.assertEqual( @@ -249,8 +254,7 @@ class TestBaseTemplates(BaseTemplate.BaseTemplateCases): job_ctx = { 'lava_test_results_dir': '/sysroot/lava-%s', } - rendered = self.render_device_dictionary_file('hi6220-hikey-01.jinja2', job_ctx=job_ctx) - template_dict = yaml.load(rendered) + template_dict = self.render_device_dictionary_file('hi6220-hikey-01.jinja2', job_ctx=job_ctx, raw=False) self.assertIsNotNone(template_dict['constants']) self.assertIn('posix', template_dict['constants']) self.assertEqual( diff --git a/lava_scheduler_app/tests/test_connections.py b/lava_scheduler_app/tests/test_connections.py index f52546e55..db80feef2 100644 --- a/lava_scheduler_app/tests/test_connections.py +++ b/lava_scheduler_app/tests/test_connections.py @@ -18,7 +18,7 @@ class YamlSshFactory(YamlFactory): def make_job_data(self, actions=None, **kw): sample_job_file = os.path.join(os.path.dirname(__file__), 'devices', 'qemu-ssh-guest.yaml') with open(sample_job_file, 'r') as test_support: - data = yaml.load(test_support) + data = yaml.safe_load(test_support) data.update(kw) return data @@ -31,7 +31,7 @@ class SecondaryConnections(TestCaseWithFactory): self.device_type = self.factory.make_device_type() def test_ssh_job_data(self): - data = yaml.load(self.factory.make_job_yaml()) + data = yaml.safe_load(self.factory.make_job_yaml()) self.assertNotIn('context', data) self.assertNotIn('timeout', data) self.assertIn('timeouts', data) @@ -52,7 +52,7 @@ class SecondaryConnections(TestCaseWithFactory): path = os.path.join(os.path.dirname(os.path.join(__file__)), 'devices') host_role = [] for job in jobs: - data = yaml.load(job.definition) + data = yaml.safe_load(job.definition) params = data['protocols']['lava-multinode'] params['target_group'] = 'replaced' if not group_size: @@ -62,7 +62,7 @@ class SecondaryConnections(TestCaseWithFactory): self.assertEqual(job.requested_device_type.name, device.device_type.name) self.assertIn(params['sub_id'], [0, 1, 2]) sub_id.append(params['sub_id']) - comparison = yaml.load(open(os.path.join(path, 'qemu-ssh-parent.yaml'), 'r').read()) + comparison = yaml.safe_load(open(os.path.join(path, 'qemu-ssh-parent.yaml'), 'r').read()) self.assertIn('protocols', data) self.assertIn('lava-multinode', data['protocols']) self.assertIn('sub_id', data['protocols']['lava-multinode']) @@ -85,7 +85,7 @@ class SecondaryConnections(TestCaseWithFactory): del(data['protocols']['lava-multinode']['sub_id']) self.assertEqual( data, - yaml.load(open(os.path.join(path, 'qemu-ssh-guest-1.yaml'), 'r').read()) + yaml.safe_load(open(os.path.join(path, 'qemu-ssh-guest-1.yaml'), 'r').read()) ) self.assertIsNone(job.requested_device_type) self.assertIsNone(job.actual_device) @@ -102,7 +102,7 @@ class SecondaryConnections(TestCaseWithFactory): # create a new device to allow the submission to reach the multinode YAML test. hostname = 'fakeqemu4' self.factory.make_device(self.device_type, hostname) - data = yaml.load(self.factory.make_job_yaml()) + data = yaml.safe_load(self.factory.make_job_yaml()) data['protocols']['lava-multinode']['roles']['host']['count'] = 2 self.assertRaises( SubmissionException, TestJob.from_yaml_and_user, @@ -114,7 +114,7 @@ class SecondaryConnections(TestCaseWithFactory): # create a new device to allow the submission to reach the multinode YAML test. hostname = 'fakeqemu4' self.factory.make_device(self.device_type, hostname) - data = yaml.load(self.factory.make_job_yaml()) + data = yaml.safe_load(self.factory.make_job_yaml()) deploy = [action['deploy'] for action in data['actions'] if 'deploy' in action] # replace working image with a broken URL for block in deploy: diff --git a/lava_scheduler_app/tests/test_device.py b/lava_scheduler_app/tests/test_device.py index b1987c85a..096fa26c9 100644 --- a/lava_scheduler_app/tests/test_device.py +++ b/lava_scheduler_app/tests/test_device.py @@ -4,8 +4,6 @@ import os import yaml import jinja2 import logging -from django.contrib.contenttypes.models import ContentType -from django.contrib.admin.models import LogEntry from django.db.models import Q from lava_scheduler_app.models import ( Device, @@ -133,7 +131,7 @@ class DeviceTypeTest(TestCaseWithFactory): data = None try: data = template.render() - yaml_data = yaml.load(data, Loader=yaml.CLoader) + yaml_data = yaml.load(data, Loader=yaml.CSafeLoader) # nosec - safe_load implemented directly except yaml.YAMLError as exc: print(data) # for easier debugging - use the online yaml parser self.fail("%s: %s" % (template_name, exc)) diff --git a/lava_scheduler_app/tests/test_fastboot_templates.py b/lava_scheduler_app/tests/test_fastboot_templates.py index d01fd6a95..486ebcb89 100644 --- a/lava_scheduler_app/tests/test_fastboot_templates.py +++ b/lava_scheduler_app/tests/test_fastboot_templates.py @@ -36,9 +36,7 @@ class TestFastbootTemplates(BaseTemplate.BaseTemplateCases): {% set static_info = [{'board_id': 'R32D300FRYP'}] %} """ self.assertTrue(self.validate_data('nexus4-01', data)) - test_template = prepare_jinja_template('nexus4-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('nexus4-01', data, raw=False) self.assertEqual('R32D300FRYP', template_dict['adb_serial_number']) self.assertEqual('R32D300FRYP', template_dict['fastboot_serial_number']) self.assertEqual([], template_dict['fastboot_options']) @@ -53,9 +51,7 @@ class TestFastbootTemplates(BaseTemplate.BaseTemplateCases): {% set interrupt_prompt = "interrupt bootloader" %} """ self.assertTrue(self.validate_data('x15-01', data)) - test_template = prepare_jinja_template('x15-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('x15-01', data, raw=False) self.assertEqual('R32D300FRYP', template_dict['adb_serial_number']) self.assertEqual('R32D300FRYP', template_dict['fastboot_serial_number']) self.assertEqual([], template_dict['fastboot_options']) @@ -85,9 +81,7 @@ class TestFastbootTemplates(BaseTemplate.BaseTemplateCases): {% set fastboot_serial_number = 'R32D300FRYP' %} """ self.assertTrue(self.validate_data('sharkl2-01', data)) - test_template = prepare_jinja_template('sharkl2-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('sharkl2-01', data, raw=False) self.assertEqual('R32D300FRYP', template_dict['adb_serial_number']) self.assertEqual('R32D300FRYP', template_dict['fastboot_serial_number']) self.assertEqual([], template_dict['fastboot_options']) @@ -108,10 +102,8 @@ class TestFastbootTemplates(BaseTemplate.BaseTemplateCases): {% set static_info = [{'board_id': 'R32D300FRYP'}] %} """ self.assertTrue(self.validate_data('staging-nexus10-01', data)) - test_template = prepare_jinja_template( - 'staging-pixel-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template( + 'staging-pixel-01', data, raw=False) self.assertIsNotNone(template_dict) self.assertIsInstance(template_dict['device_info'], list) self.assertIsInstance(template_dict['static_info'], list) @@ -121,9 +113,7 @@ class TestFastbootTemplates(BaseTemplate.BaseTemplateCases): data = hikey.read() self.assertIsNotNone(data) self.assertTrue(self.validate_data('hi6220-hikey-01', data)) - test_template = prepare_jinja_template('staging-hikey-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-hikey-01', data, raw=False) self.assertIsNotNone(template_dict) self.assertIsInstance(template_dict['device_info'], list) self.assertIsInstance(template_dict['static_info'], list) @@ -143,9 +133,7 @@ class TestFastbootTemplates(BaseTemplate.BaseTemplateCases): data = hikey.read() self.assertIsNotNone(data) self.assertTrue(self.validate_data('hi6220-hikey-01', data)) - test_template = prepare_jinja_template('staging-hikey-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-hikey-01', data, raw=False) self.assertIsNotNone(template_dict) self.assertIsInstance(template_dict['device_info'], list) self.assertIsInstance(template_dict['static_info'], list) @@ -165,9 +153,7 @@ class TestFastbootTemplates(BaseTemplate.BaseTemplateCases): # test support for retreiving MAC from device. data += "{% set device_mac = '00:E0:4C:53:44:58' %}" self.assertTrue(self.validate_data('hi6220-hikey-01', data)) - test_template = prepare_jinja_template('staging-hikey-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-hikey-01', data, raw=False) self.assertIn('parameters', template_dict) self.assertIn('interfaces', template_dict['parameters']) self.assertIn('target', template_dict['parameters']['interfaces']) @@ -186,9 +172,7 @@ class TestFastbootTemplates(BaseTemplate.BaseTemplateCases): 'devicetree': 'hi6220-hikey.dtb' } self.assertTrue(self.validate_data('hi6220-hikey-01', data)) - test_template = prepare_jinja_template('staging-hikey-01', data) - rendered = test_template.render(**job_ctx) - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-hikey-01', data, job_ctx=job_ctx, raw=False) self.assertIsNotNone(template_dict) self.assertIsNotNone(template_dict['actions']['boot']['methods']) self.assertIn('grub-efi', template_dict['actions']['boot']['methods']) @@ -221,9 +205,7 @@ class TestFastbootTemplates(BaseTemplate.BaseTemplateCases): self.assertIsNotNone(data) job_ctx = {} self.assertTrue(self.validate_data('hi6220-hikey-01', data)) - test_template = prepare_jinja_template('staging-hikey-01', data) - rendered = test_template.render(**job_ctx) - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-hikey-01', data, job_ctx=job_ctx, raw=False) validate_device(template_dict) self.assertIsNotNone(template_dict) self.assertIn('commands', template_dict) @@ -250,9 +232,7 @@ class TestFastbootTemplates(BaseTemplate.BaseTemplateCases): 'devicetree': 'hi960-hikey.dtb' } # self.assertTrue(self.validate_data('hi960-hikey-01', data)) - test_template = prepare_jinja_template('staging-hikey-01', data) - rendered = test_template.render(**job_ctx) - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-hikey-01', data, job_ctx=job_ctx, raw=False) self.assertIsNotNone(template_dict) self.assertIsNotNone(template_dict['actions']['boot']['methods']) self.assertNotIn('menu_options', template_dict['actions']['boot']['methods']['grub']) @@ -271,9 +251,7 @@ class TestFastbootTemplates(BaseTemplate.BaseTemplateCases): # test support for retreiving MAC from device using base-fastboot. data += "{% set device_mac = '00:E0:4C:53:44:58' %}" self.assertTrue(self.validate_data('hi6220-hikey-01', data)) - test_template = prepare_jinja_template('staging-hikey-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-hikey-01', data, job_ctx=None, raw=False) self.assertIn('parameters', template_dict) self.assertIn('interfaces', template_dict['parameters']) self.assertIn('target', template_dict['parameters']['interfaces']) @@ -292,10 +270,8 @@ class TestFastbootTemplates(BaseTemplate.BaseTemplateCases): {% set static_info = [{'board_id': '10de1214adae123'}] %} """ self.assertTrue(self.validate_data('staging-nexus5x-01', data)) - test_template = prepare_jinja_template( - 'staging-pixel-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template( + 'staging-pixel-01', data, raw=False) self.assertIsNotNone(template_dict) self.assertIsInstance(template_dict['device_info'], list) self.assertIsInstance(template_dict['static_info'], list) @@ -308,10 +284,8 @@ class TestFastbootTemplates(BaseTemplate.BaseTemplateCases): {% set static_info = [{'board_id': 'FDAC1231DAD'}] %} """ self.assertTrue(self.validate_data('staging-pixel-01', data)) - test_template = prepare_jinja_template( - 'staging-pixel-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template( + 'staging-pixel-01', data, raw=False) self.assertIsNotNone(template_dict) self.assertIsInstance(template_dict['device_info'], list) self.assertIsInstance(template_dict['static_info'], list) @@ -327,10 +301,8 @@ class TestFastbootTemplates(BaseTemplate.BaseTemplateCases): {% set fastboot_serial_number = 'e080c212' %} """ self.assertTrue(self.validate_data('staging-ifc6410-01', data)) - test_template = prepare_jinja_template('staging-ifc6410-01', - data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template( + 'staging-ifc6410-01', data, raw=False) self.assertEqual('e080c212', template_dict['adb_serial_number']) self.assertEqual('e080c212', template_dict['fastboot_serial_number']) self.assertEqual([], template_dict['fastboot_options']) @@ -346,9 +318,7 @@ class TestFastbootTemplates(BaseTemplate.BaseTemplateCases): {% set fastboot_serial_number = '3083f595' %} """ self.assertTrue(self.validate_data('db820c-01', data)) - test_template = prepare_jinja_template('db820c-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('db820c-01', data, raw=False) self.assertEqual('3083f595', template_dict['adb_serial_number']) self.assertEqual('3083f595', template_dict['fastboot_serial_number']) self.assertEqual([], template_dict['fastboot_options']) @@ -357,9 +327,7 @@ class TestFastbootTemplates(BaseTemplate.BaseTemplateCases): with open(os.path.join(os.path.dirname(__file__), 'devices', 'hi6220-hikey-bl-01.jinja2')) as hikey: data = hikey.read() self.assertTrue(self.validate_data('hi620-bl-01', data)) - test_template = prepare_jinja_template('hi620-bl-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('hi620-bl-01', data, raw=False) recovery = template_dict['actions']['deploy']['methods'] self.assertIsNotNone('recovery', recovery) self.assertIn('recovery', recovery) diff --git a/lava_scheduler_app/tests/test_grub_templates.py b/lava_scheduler_app/tests/test_grub_templates.py index 9c0f6f549..e8d1e7c6a 100644 --- a/lava_scheduler_app/tests/test_grub_templates.py +++ b/lava_scheduler_app/tests/test_grub_templates.py @@ -1,4 +1,3 @@ -import yaml # pylint: disable=superfluous-parens,ungrouped-imports from lava_scheduler_app.tests.test_base_templates import ( BaseTemplate, @@ -33,9 +32,7 @@ class TestGrubTemplates(BaseTemplate.BaseTemplateCases): {% set power_on_command = '/usr/bin/pduclient --daemon services --hostname pdu09 --command on --port 05' %} {% set connection_command = 'telnet localhost 7012' %}""" self.assertTrue(self.validate_data('staging-mustang-01', data)) - test_template = prepare_jinja_template('staging-mustang-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-mustang-01', data, raw=False) self.assertIn('uefi-menu', template_dict['actions']['boot']['methods']) self.assertIn('pxe-grub', template_dict['actions']['boot']['methods']['uefi-menu']) self.assertNotIn('grub', template_dict['actions']['boot']['methods']['uefi-menu']) @@ -60,9 +57,7 @@ class TestGrubTemplates(BaseTemplate.BaseTemplateCases): {% set power_on_command = '/usr/bin/pduclient --daemon services --hostname pdu09 --command on --port 05' %} {% set connection_command = 'telnet localhost 7012' %}""" self.assertTrue(self.validate_data('staging-mustang-01', data)) - test_template = prepare_jinja_template('staging-mustang-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-mustang-01', data, raw=False) self.assertIn('uefi-menu', template_dict['actions']['boot']['methods']) self.assertNotIn('pxe-grub', template_dict['actions']['boot']['methods']['uefi-menu']) self.assertIn('grub', template_dict['actions']['boot']['methods']['uefi-menu']) @@ -85,9 +80,7 @@ class TestGrubTemplates(BaseTemplate.BaseTemplateCases): {% set power_on_command = '/usr/bin/pduclient --daemon services --hostname pdu09 --command on --port 05' %} {% set connection_command = 'telnet localhost 7012' %}""" self.assertTrue(self.validate_data('staging-mustang-01', data)) - test_template = prepare_jinja_template('staging-mustang-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-mustang-01', data, raw=False) parameters = { 'parameters': { 'media': { @@ -133,9 +126,7 @@ class TestGrubTemplates(BaseTemplate.BaseTemplateCases): 'iface3': '/sys/devices/pci0000:00/0000:00:02.1/0000:01:00.1/net/'} %} {% set boot_character_delay = 100 %}""" self.assertTrue(self.validate_data('staging-overdrive-01', data)) - test_template = prepare_jinja_template('staging-overdrive-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-overdrive-01', data, raw=False) self.assertIsNotNone(template_dict) self.assertIn('parameters', template_dict) self.assertIn('interfaces', template_dict['parameters']) @@ -157,9 +148,7 @@ class TestGrubTemplates(BaseTemplate.BaseTemplateCases): ) def test_synquacer_acpi_template(self): - rendered = self.render_device_dictionary_file('synquacer-acpi-01.jinja2') - template_dict = yaml.load(rendered) - template_dict = yaml.load(rendered) + template_dict = self.render_device_dictionary_file('synquacer-acpi-01.jinja2', raw=False) self.assertIsNotNone(template_dict['actions']['boot']['methods']['grub']) self.assertIn('ramdisk', template_dict['actions']['boot']['methods']['grub']) self.assertIn('commands', template_dict['actions']['boot']['methods']['grub']['ramdisk']) @@ -173,8 +162,7 @@ class TestGrubTemplates(BaseTemplate.BaseTemplateCases): ) def test_synquacer_dtb_template(self): - rendered = self.render_device_dictionary_file('synquacer-dtb-01.jinja2') - template_dict = yaml.load(rendered) + template_dict = self.render_device_dictionary_file('synquacer-dtb-01.jinja2', raw=False) self.assertIsNotNone(template_dict['actions']['boot']['methods']['grub']) self.assertIn('ramdisk', template_dict['actions']['boot']['methods']['grub']) self.assertIn('commands', template_dict['actions']['boot']['methods']['grub']['ramdisk']) @@ -188,8 +176,7 @@ class TestGrubTemplates(BaseTemplate.BaseTemplateCases): ) def test_moonshot_m400_template(self): - rendered = self.render_device_dictionary_file('moonshot-m400-17.jinja2') - template_dict = yaml.load(rendered) + template_dict = self.render_device_dictionary_file('moonshot-m400-17.jinja2', raw=False) self.assertIsNotNone(template_dict['actions']['boot']['methods']['grub']) self.assertIn('ramdisk', template_dict['actions']['boot']['methods']['grub']) self.assertIn('commands', template_dict['actions']['boot']['methods']['grub']['ramdisk']) @@ -199,8 +186,7 @@ class TestGrubTemplates(BaseTemplate.BaseTemplateCases): ) def test_minnowboard_turbot_template(self): - rendered = self.render_device_dictionary_file('minnowboard-turbot-E3826-01.jinja2') - template_dict = yaml.load(rendered) + template_dict = self.render_device_dictionary_file('minnowboard-turbot-E3826-01.jinja2', raw=False) grub = template_dict['actions']['boot']['methods']['grub'] self.assertIsNotNone(grub) diff --git a/lava_scheduler_app/tests/test_menus.py b/lava_scheduler_app/tests/test_menus.py index 52c8f3a99..5d20e1355 100644 --- a/lava_scheduler_app/tests/test_menus.py +++ b/lava_scheduler_app/tests/test_menus.py @@ -12,7 +12,7 @@ class YamlMenuFactory(YamlFactory): 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.load(test_support) + data = yaml.safe_load(test_support) data.update(kw) return data @@ -27,7 +27,7 @@ class TestPipelineMenu(TestCaseWithFactory): # pylint: disable=too-many-ancesto self.device_type = self.factory.make_device_type(name='mustang-uefi') def test_make_job_yaml(self): - data = yaml.load(self.factory.make_job_yaml()) + data = yaml.safe_load(self.factory.make_job_yaml()) self.assertIn('device_type', data) self.assertNotIn('timeout', data) self.assertIn('timeouts', data) diff --git a/lava_scheduler_app/tests/test_pipeline.py b/lava_scheduler_app/tests/test_pipeline.py index cfb7f0ce4..9675f4f78 100644 --- a/lava_scheduler_app/tests/test_pipeline.py +++ b/lava_scheduler_app/tests/test_pipeline.py @@ -13,10 +13,8 @@ from lava_scheduler_app.models import ( Tag, DevicesUnavailableException, _pipeline_protocols, - _check_submit_to_device, ) from lava_scheduler_app.dbutils import match_vlan_interface -from django.db.models import Q from django.contrib.auth.models import Group, Permission, User from lava_scheduler_app.utils import ( split_multinode_yaml, @@ -35,7 +33,6 @@ from lava_dispatcher.actions.boot.qemu import BootQEMU from lava_dispatcher.protocols.multinode import MultinodeProtocol from lava_common.constants import SYS_CLASS_KVM from lava_dispatcher.test.utils import DummyLogger -from django_restricted_resource.managers import RestrictedResourceQuerySet from unittest import TestCase @@ -90,7 +87,7 @@ class YamlFactory(ModelFactory): def make_job_data(self, actions=None, **kw): sample_job_file = os.path.join(os.path.dirname(__file__), 'devices', 'qemu.yaml') with open(sample_job_file, 'r') as test_support: - data = yaml.load(test_support) + data = yaml.safe_load(test_support) data.update(kw) return data @@ -115,7 +112,7 @@ class PipelineDeviceTags(TestCaseWithFactory): } def test_make_job_yaml(self): - data = yaml.load(self.factory.make_job_yaml()) + data = yaml.safe_load(self.factory.make_job_yaml()) self.assertIn('device_type', data) self.assertNotIn('timeout', data) self.assertIn('timeouts', data) @@ -150,7 +147,7 @@ class PipelineDeviceTags(TestCaseWithFactory): self.factory.ensure_tag('usb'), self.factory.ensure_tag('sata') ] - data = yaml.load(self.factory.make_job_yaml(tags=['usb', 'sata'])) + data = yaml.safe_load(self.factory.make_job_yaml(tags=['usb', 'sata'])) validate_submission(data) device = self.factory.make_device(self.device_type, 'fakeqemu4', tags=tag_list) job_ctx = data.get('context', {}) @@ -266,7 +263,7 @@ class TestPipelineSubmit(TestCaseWithFactory): user = self.factory.make_user() job = TestJob.from_yaml_and_user( self.factory.make_job_yaml(), user) - job_def = yaml.load(job.definition) + job_def = yaml.safe_load(job.definition) job_ctx = job_def.get('context', {}) device_config = device.load_configuration(job_ctx) # raw dict self.assertTrue(device.is_valid(system=False)) @@ -275,7 +272,7 @@ class TestPipelineSubmit(TestCaseWithFactory): 'qemu-system-x86_64' ) - job_data = yaml.load(self.factory.make_job_yaml()) + job_data = yaml.safe_load(self.factory.make_job_yaml()) job_data['context'].update({'netdevice': 'tap'}) job_ctx = job_data.get('context', {}) device_config = device.load_configuration(job_ctx) # raw dict @@ -325,7 +322,7 @@ class TestPipelineSubmit(TestCaseWithFactory): ) def test_metadata(self): - data = yaml.load(self.factory.make_job_yaml(metadata={15: 'test'})) + data = yaml.safe_load(self.factory.make_job_yaml(metadata={15: 'test'})) validate_submission(data) metadata = { 'build-url': None, @@ -335,13 +332,13 @@ class TestPipelineSubmit(TestCaseWithFactory): 'git-url': 'https://git.linaro.org/zephyrproject-org/zephyr.git', 'git-commit': 1234 } - data = yaml.load(self.factory.make_job_yaml(metadata=metadata)) + data = yaml.safe_load(self.factory.make_job_yaml(metadata=metadata)) self.assertRaises(SubmissionException, validate_submission, data) metadata['build-url'] = 'http://nowhere.com' - data = yaml.load(self.factory.make_job_yaml(metadata=metadata)) + data = yaml.safe_load(self.factory.make_job_yaml(metadata=metadata)) self.assertRaises(SubmissionException, validate_submission, data) metadata['zephyr-gcc-variant'] = '4.9' - data = yaml.load(self.factory.make_job_yaml(metadata=metadata)) + data = yaml.safe_load(self.factory.make_job_yaml(metadata=metadata)) validate_submission(data) def test_command_list(self): @@ -371,7 +368,7 @@ class TestPipelineSubmit(TestCaseWithFactory): self.assertIn('timeouts', device_dict) def test_auto_login(self): - data = yaml.load(self.factory.make_job_yaml()) + data = yaml.safe_load(self.factory.make_job_yaml()) validate_submission(data) boot_params = None @@ -406,7 +403,7 @@ class TestPipelineSubmit(TestCaseWithFactory): # public set in the YAML yaml_str = self.factory.make_job_yaml() - yaml_data = yaml.load(yaml_str) + yaml_data = yaml.safe_load(yaml_str) job = TestJob.from_yaml_and_user( yaml_str, user) self.assertTrue(job.is_public) @@ -436,7 +433,7 @@ class TestPipelineSubmit(TestCaseWithFactory): self.assertFalse(job2.can_view(user)) self.assertTrue(job2.can_view(user3)) - job_data = yaml.load(self.factory.make_job_yaml()) + job_data = yaml.safe_load(self.factory.make_job_yaml()) job_data['visibility'] = {'group': [group1.name]} param = job_data['visibility'] if isinstance(param, dict): @@ -458,7 +455,7 @@ class TestPipelineSubmit(TestCaseWithFactory): user = self.factory.make_user() # public set in the YAML yaml_str = self.factory.make_job_yaml() - yaml_data = yaml.load(yaml_str) + yaml_data = yaml.safe_load(yaml_str) job = TestJob.from_yaml_and_user( yaml_str, user) self.assertTrue(job.is_public) @@ -467,10 +464,10 @@ class TestPipelineSubmit(TestCaseWithFactory): self.assertEqual(job.pipeline_compatibility, 0) self.assertNotIn('compatibility', yaml_data) # FIXME: dispatcher master needs to make this kind of test more accessible. - definition = yaml.load(job.definition) + definition = yaml.safe_load(job.definition) self.assertNotIn('protocols', definition) job.actual_device = Device.objects.get(hostname='fakeqemu1') - job_def = yaml.load(job.definition) + job_def = yaml.safe_load(job.definition) job_ctx = job_def.get('context', {}) parser = JobParser() device = job.actual_device @@ -583,7 +580,7 @@ class TestYamlMultinode(TestCaseWithFactory): """ server_check = os.path.join(os.path.dirname(__file__), 'sample_jobs', 'kvm-multinode-server.yaml') client_check = os.path.join(os.path.dirname(__file__), 'sample_jobs', 'kvm-multinode-client.yaml') - submission = yaml.load(open( + submission = yaml.safe_load(open( os.path.join(os.path.dirname(__file__), 'sample_jobs', 'kvm-multinode.yaml'), 'r')) target_group = 'arbitrary-group-id' # for unit tests only @@ -594,16 +591,16 @@ class TestYamlMultinode(TestCaseWithFactory): del(job['protocols']['lava-multinode']['sub_id']) yaml.dump(job) # ensure the jobs can be serialised as YAML if role == 'client': - self.assertEqual(job, yaml.load(open(client_check, 'r'))) + self.assertEqual(job, yaml.safe_load(open(client_check, 'r'))) if role == 'server': - self.assertEqual(job, yaml.load(open(server_check, 'r'))) + self.assertEqual(job, yaml.safe_load(open(server_check, 'r'))) @unittest.skipIf(check_rpcinfo(), "rpcinfo returns non-zero for nfs") def test_secondary_connection(self): user = self.factory.make_user() device_type = self.factory.make_device_type(name='mustang') device = self.factory.make_device(device_type, 'mustang1') - submission = yaml.load(open( + submission = yaml.safe_load(open( os.path.join(os.path.dirname(__file__), 'sample_jobs', 'mustang-ssh-multinode.yaml'), 'r')) target_group = 'arbitrary-group-id' # for unit tests only jobs_dict = split_multinode_yaml(submission, target_group) @@ -664,7 +661,7 @@ class TestYamlMultinode(TestCaseWithFactory): Tag.objects.all().delete() self.factory.ensure_tag('tap'), self.factory.ensure_tag('virtio') - submission = yaml.load(open( + submission = yaml.safe_load(open( os.path.join(os.path.dirname(__file__), 'sample_jobs', 'kvm-multinode.yaml'), 'r')) roles_dict = submission['protocols'][MultinodeProtocol.name]['roles'] roles_dict['client']['tags'] = ['tap'] @@ -710,7 +707,7 @@ class TestYamlMultinode(TestCaseWithFactory): self.assertEqual(len(job_list), 3) def test_multinode_lxc(self): - submission = yaml.load(open( + submission = yaml.safe_load(open( os.path.join(os.path.dirname(__file__), 'sample_jobs', 'lxc-multinode.yaml'), 'r')) target_group = 'arbitrary-group-id' # for unit tests only @@ -731,7 +728,7 @@ class TestYamlMultinode(TestCaseWithFactory): self.fail('Unrecognised role: %s' % role) def test_multinode_hikey(self): - submission = yaml.load(open( + submission = yaml.safe_load(open( os.path.join(os.path.dirname(__file__), 'sample_jobs', 'hikey_multinode.yaml'), 'r')) target_group = 'arbitrary-group-id' # for unit tests only @@ -758,7 +755,7 @@ class TestYamlMultinode(TestCaseWithFactory): self.fail('Unrecognised role: %s' % role) def test_multinode_nexus4(self): - submission = yaml.load(open( + submission = yaml.safe_load(open( os.path.join(os.path.dirname(__file__), 'sample_jobs', 'nexus4_multinode.yaml'), 'r')) target_group = 'arbitrary-group-id' # for unit tests only @@ -783,7 +780,7 @@ class TestYamlMultinode(TestCaseWithFactory): device_type = self.factory.make_device_type() Device.objects.filter(device_type=device_type).delete() Tag.objects.all().delete() - submission = yaml.load(open( + submission = yaml.safe_load(open( os.path.join(os.path.dirname(__file__), 'sample_jobs', 'kvm-multinode.yaml'), 'r')) # no devices defined for the specified type self.assertRaises(DevicesUnavailableException, _pipeline_protocols, submission, user, yaml_data=None) @@ -810,7 +807,7 @@ class TestYamlMultinode(TestCaseWithFactory): self.assertEqual(len(job_object_list), 2) for job in job_object_list: self.assertEqual(list(job.sub_jobs_list), job_object_list) - check = yaml.load(job.definition) + check = yaml.safe_load(job.definition) if check['protocols']['lava-multinode']['role'] == 'client': self.assertEqual( check['protocols']['lava-multinode']['tags'], @@ -826,7 +823,7 @@ class TestYamlMultinode(TestCaseWithFactory): def test_multinode_group(self): user = self.factory.make_user() device_type = self.factory.make_device_type() - submission = yaml.load(open( + submission = yaml.safe_load(open( os.path.join(os.path.dirname(__file__), 'sample_jobs', 'kvm-multinode.yaml'), 'r')) self.factory.make_device(device_type, 'fakeqemu1') tag_list = [ @@ -838,8 +835,8 @@ class TestYamlMultinode(TestCaseWithFactory): job_object_list = _pipeline_protocols(submission, user, yaml.dump(submission)) for job in job_object_list: self.assertEqual(list(job.sub_jobs_list), job_object_list) - check_one = yaml.load(job_object_list[0].definition) - check_two = yaml.load(job_object_list[1].definition) + check_one = yaml.safe_load(job_object_list[0].definition) + check_two = yaml.safe_load(job_object_list[1].definition) self.assertEqual( job_object_list[0].target_group, job_object_list[1].target_group @@ -868,7 +865,7 @@ class TestYamlMultinode(TestCaseWithFactory): def test_multinode_definition(self): user = self.factory.make_user() device_type = self.factory.make_device_type() - submission = yaml.load(open( + submission = yaml.safe_load(open( os.path.join(os.path.dirname(__file__), 'sample_jobs', 'kvm-multinode.yaml'), 'r')) self.factory.make_device(device_type, 'fakeqemu1') self.factory.make_device(device_type, 'fakeqemu2') @@ -893,7 +890,7 @@ class TestYamlMultinode(TestCaseWithFactory): def test_invalid_multinode(self): # pylint: disable=too-many-locals user = self.factory.make_user() device_type = self.factory.make_device_type() - submission = yaml.load(open( + submission = yaml.safe_load(open( os.path.join(os.path.dirname(__file__), 'sample_jobs', 'kvm-multinode.yaml'), 'r')) tag_list = [ @@ -920,10 +917,10 @@ class TestYamlMultinode(TestCaseWithFactory): "%d.%d" % (int(job_object_list[0].id), 0)) # FIXME: dispatcher master needs to make this kind of test more accessible. for job in job_object_list: - definition = yaml.load(job.definition) + definition = yaml.safe_load(job.definition) self.assertNotEqual(definition['protocols']['lava-multinode']['sub_id'], '') job.actual_device = Device.objects.get(hostname='fakeqemu1') - job_def = yaml.load(job.definition) + job_def = yaml.safe_load(job.definition) job_ctx = job_def.get('context', {}) parser = JobParser() device = None @@ -974,7 +971,7 @@ class TestYamlMultinode(TestCaseWithFactory): self.factory.make_device(device_type, 'fakeqemu2') self.factory.make_device(device_type, 'fakeqemu3') self.factory.make_device(device_type, 'fakeqemu4') - submission = yaml.load(open( + submission = yaml.safe_load(open( os.path.join(os.path.dirname(__file__), 'sample_jobs', 'kvm-multinode.yaml'), 'r')) role_list = submission['protocols'][MultinodeProtocol.name]['roles'] for role in role_list: @@ -987,7 +984,7 @@ class TestYamlMultinode(TestCaseWithFactory): device_type = self.factory.make_device_type() self.factory.make_device(device_type, 'fakeqemu1') self.factory.make_device(device_type, 'fakeqemu2') - client_submission = yaml.load(open( + client_submission = yaml.safe_load(open( os.path.join(os.path.dirname(__file__), 'sample_jobs', 'kvm-multinode-client.yaml'), 'r')) job_ctx = client_submission.get('context', {}) device = Device.objects.get(hostname='fakeqemu1') @@ -1011,7 +1008,7 @@ class TestYamlMultinode(TestCaseWithFactory): }, meta_dict) # simulate dynamic connection - dynamic = yaml.load(open( + dynamic = yaml.load(open( # nosec - not suitable for safe_load os.path.join(os.path.dirname(__file__), 'pipeline_refs', 'connection-description.yaml'), 'r')) meta_dict = _get_action_metadata(dynamic['job']['actions']) self.assertEqual( @@ -1037,12 +1034,12 @@ class TestYamlMultinode(TestCaseWithFactory): bbb_type = self.factory.make_device_type('beaglebone-black') self.factory.make_device(hostname='bbb-01', device_type=bbb_type) self.factory.make_device(hostname='bbb-02', device_type=bbb_type) - submission = yaml.load(open( + submission = yaml.safe_load(open( os.path.join(os.path.dirname(__file__), 'sample_jobs', 'bbb-qemu-multinode.yaml'), 'r')) job_object_list = _pipeline_protocols(submission, user, yaml.dump(submission)) for job in job_object_list: - definition = yaml.load(job.definition) + definition = yaml.safe_load(job.definition) self.assertNotEqual(definition['protocols']['lava-multinode']['sub_id'], '') sub_ids = [job.sub_id for job in job_object_list] self.assertEqual(len(set(sub_ids)), len(sub_ids)) @@ -1052,7 +1049,7 @@ class TestYamlMultinode(TestCaseWithFactory): job.actual_device = Device.objects.get(hostname='bbb-01') else: self.fail('Unrecognised device type: %s' % job.requested_device_type) - job_def = yaml.load(job.definition) + job_def = yaml.safe_load(job.definition) job_ctx = job_def.get('context', {}) parser = JobParser() device_object = None @@ -1098,14 +1095,14 @@ class TestYamlMultinode(TestCaseWithFactory): bbb_type = self.factory.make_device_type('beaglebone-black') self.factory.make_device(hostname='bbb-01', device_type=bbb_type) self.factory.make_device(hostname='bbb-02', device_type=bbb_type) - submission = yaml.load(open( + submission = yaml.safe_load(open( os.path.join(os.path.dirname(__file__), 'sample_jobs', 'bbb-qemu-multinode.yaml'), 'r')) self.assertIn('protocols', submission) self.assertIn(MultinodeProtocol.name, submission['protocols']) submission['protocols'][MultinodeProtocol.name]['roles']['server']['essential'] = True job_object_list = _pipeline_protocols(submission, user, yaml.dump(submission)) for job in job_object_list: - definition = yaml.load(job.definition) + definition = yaml.safe_load(job.definition) role = definition['protocols'][MultinodeProtocol.name]['role'] self.assertNotEqual(definition['protocols']['lava-multinode']['sub_id'], '') if role == 'client': @@ -1132,7 +1129,7 @@ class VlanInterfaces(TestCaseWithFactory): self.filename = os.path.join(os.path.dirname(__file__), 'sample_jobs', 'bbb-cubie-vlan-group.yaml') def test_vlan_interface(self): # pylint: disable=too-many-locals - submission = yaml.load(open(self.filename, 'r')) + submission = yaml.safe_load(open(self.filename, 'r')) self.assertIn('protocols', submission) self.assertIn('lava-vland', submission['protocols']) roles = [role for role, _ in submission['protocols']['lava-vland'].items()] diff --git a/lava_scheduler_app/tests/test_qemu_templates.py b/lava_scheduler_app/tests/test_qemu_templates.py index eaa0d3536..a114b8825 100644 --- a/lava_scheduler_app/tests/test_qemu_templates.py +++ b/lava_scheduler_app/tests/test_qemu_templates.py @@ -32,14 +32,14 @@ class TestQemuTemplates(BaseTemplate.BaseTemplateCases): {% set memory = 512 %}""" job_ctx = {'arch': 'amd64', 'no_kvm': True} self.assertTrue(self.validate_data('staging-qemu-01', data, job_ctx)) - test_template = prepare_jinja_template('staging-qemu-01', data) + test_template = prepare_jinja_template('staging-qemu-01', data, job_ctx=None, raw=True) rendered = test_template.render(**job_ctx) - template_dict = yaml.load(rendered) + template_dict = yaml.safe_load(rendered) options = template_dict['actions']['boot']['methods']['qemu']['parameters']['options'] self.assertNotIn('-enable-kvm', options) job_ctx = {'arch': 'amd64', 'no_kvm': False} rendered = test_template.render(**job_ctx) - template_dict = yaml.load(rendered) + template_dict = yaml.safe_load(rendered) options = template_dict['actions']['boot']['methods']['qemu']['parameters']['options'] self.assertIn('-enable-kvm', options) @@ -48,9 +48,7 @@ class TestQemuTemplates(BaseTemplate.BaseTemplateCases): {% set mac_addr = 'DE:AD:BE:EF:28:01' %} {% set memory = 512 %}""" job_ctx = {'arch': 'amd64'} - test_template = prepare_jinja_template('staging-qemu-01', data) - rendered = test_template.render(**job_ctx) - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-qemu-01', data, job_ctx=job_ctx, raw=False) self.assertEqual( 'c', template_dict['actions']['boot']['methods']['qemu']['parameters']['boot_options']['boot_order'] @@ -69,10 +67,7 @@ class TestQemuTemplates(BaseTemplate.BaseTemplateCases): 'extra_options': ['-global', 'virtio-blk-device.scsi=off', '-smp', 1, '-device', 'virtio-scsi-device,id=scsi'] } self.assertTrue(self.validate_data('staging-qemu-01', data)) - test_template = prepare_jinja_template('staging-juno-01', data) - rendered = test_template.render(**job_ctx) - self.assertIsNotNone(rendered) - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-juno-01', data, job_ctx=job_ctx, raw=False) options = template_dict['actions']['boot']['methods']['qemu']['parameters']['options'] self.assertIn('-cpu cortex-a57', options) self.assertNotIn('-global', options) @@ -97,10 +92,7 @@ class TestQemuTemplates(BaseTemplate.BaseTemplateCases): 'extra_options': ['-smp', 1] } self.assertTrue(self.validate_data('staging-qemu-01', data)) - test_template = prepare_jinja_template('staging-juno-01', data) - rendered = test_template.render(**job_ctx) - self.assertIsNotNone(rendered) - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-juno-01', data, job_ctx=job_ctx, raw=False) self.assertIn('qemu-nfs', template_dict['actions']['boot']['methods']) params = template_dict['actions']['boot']['methods']['qemu-nfs']['parameters'] self.assertIn('command', params) @@ -117,9 +109,7 @@ class TestQemuTemplates(BaseTemplate.BaseTemplateCases): def test_docker_template(self): data = "{% extends 'docker.jinja2' %}" self.assertTrue(self.validate_data('docker-01', data)) - test_template = prepare_jinja_template('docker-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('docker-01', data, raw=False) self.assertEqual({'docker': None}, template_dict['actions']['deploy']['methods']) self.assertEqual({'docker': {'options': {'cpus': 0.0, 'memory': 0, 'devices': [], 'volumes': []}}}, template_dict['actions']['boot']['methods']) @@ -129,9 +119,7 @@ class TestQemuTemplates(BaseTemplate.BaseTemplateCases): {% set docker_memory="120M" %} {% set docker_volumes=["/home", "/tmp"] %}""" self.assertTrue(self.validate_data('docker-01', data)) - test_template = prepare_jinja_template('docker-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('docker-01', data, raw=False) self.assertEqual({'docker': None}, template_dict['actions']['deploy']['methods']) self.assertEqual({'docker': {'options': {'cpus': 2.1, 'memory': "120M", 'devices': [], @@ -143,9 +131,7 @@ class TestQemuTemplates(BaseTemplate.BaseTemplateCases): {% set docker_memory="120M" %} {% set docker_devices=["/dev/kvm"] %}""" self.assertTrue(self.validate_data('docker-01', data)) - test_template = prepare_jinja_template('docker-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('docker-01', data, raw=False) self.assertEqual({'docker': None}, template_dict['actions']['deploy']['methods']) self.assertEqual({'docker': {'options': {'cpus': 2.1, 'memory': "120M", 'devices': ["/dev/kvm"], diff --git a/lava_scheduler_app/tests/test_templates.py b/lava_scheduler_app/tests/test_templates.py index dc53d2f02..2bdc7cc27 100644 --- a/lava_scheduler_app/tests/test_templates.py +++ b/lava_scheduler_app/tests/test_templates.py @@ -1,4 +1,3 @@ -import yaml from lava_scheduler_app.tests.test_base_templates import ( BaseTemplate, prepare_jinja_template, @@ -32,9 +31,7 @@ class TestTemplates(BaseTemplate.BaseTemplateCases): {% set power_on_command = '/usr/bin/pduclient --daemon localhost --port 02 --hostname lngpdu01 --command on' %} {% set connection_command = 'telnet localhost 7302' %}""" self.assertTrue(self.validate_data('staging-x86-01', data)) - test_template = prepare_jinja_template('staging-x86-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-x86-01', data, raw=False) for _, value in template_dict['actions']['boot']['methods']['ipxe'].items(): if 'commands' in value: for item in value['commands']: @@ -62,10 +59,8 @@ class TestTemplates(BaseTemplate.BaseTemplateCases): if item.endswith(','): self.fail("%s ends with a comma" % item) self.assertEqual(depth, 8) - test_template = prepare_jinja_template('staging-x86-01', data) job_ctx = {} - rendered = test_template.render(**job_ctx) - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-x86-01', data, job_ctx=job_ctx, raw=False) self.assertIsNotNone(template_dict['actions']['boot']['methods']['ipxe']['nfs']['commands']) @@ -77,23 +72,19 @@ class TestTemplates(BaseTemplate.BaseTemplateCases): # override template in job context job_ctx = {'boot_character_delay': 150} - rendered = test_template.render(**job_ctx) - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-x86-01', data, job_ctx=job_ctx, raw=False) self.assertEqual(150, template_dict['character_delays']['boot']) # add device dictionary override # overrides the template default data += """{% set boot_character_delay = 400 %}""" - test_template = prepare_jinja_template('staging-x86-01', data) job_ctx = {} - rendered = test_template.render(**job_ctx) - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-x86-01', data, job_ctx=job_ctx, raw=False) self.assertEqual(400, template_dict['character_delays']['boot']) # job context does not override device dictionary job_ctx = {'boot_character_delay': 150} - rendered = test_template.render(**job_ctx) - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-x86-01', data, job_ctx=job_ctx, raw=False) self.assertNotEqual(150, template_dict['character_delays']['boot']) self.assertEqual(400, template_dict['character_delays']['boot']) @@ -105,9 +96,7 @@ class TestTemplates(BaseTemplate.BaseTemplateCases): {% set power_on_command = '/usr/bin/pduclient --daemon localhost --port 02 --hostname lngpdu01 --command on' %} {% set connection_command = 'telnet localhost 7302' %}""" self.assertTrue(self.validate_data('staging-x86-01', data)) - test_template = prepare_jinja_template('staging-x86-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-x86-01', data, raw=False) for _, value in template_dict['actions']['boot']['methods']['ipxe'].items(): if 'commands' in value: self.assertIn('dhcp net0', value['commands']) @@ -120,9 +109,7 @@ class TestTemplates(BaseTemplate.BaseTemplateCases): {% set power_on_command = '/usr/bin/pduclient --daemon localhost --port 02 --hostname lngpdu01 --command on' %} {% set connection_command = 'telnet localhost 7302' %}""" self.assertTrue(self.validate_data('staging-x86-01', data)) - test_template = prepare_jinja_template('staging-x86-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-x86-01', data, raw=False) for _, value in template_dict['actions']['boot']['methods']['ipxe'].items(): if 'commands' in value: self.assertIn('dhcp net1', value['commands']) @@ -143,9 +130,7 @@ class TestTemplates(BaseTemplate.BaseTemplateCases): {% set power_on_command = '/usr/bin/pduclient --daemon localhost --hostname lngpdu01 --command on --port 19' %} {% set connection_command = 'telnet localhost 7333' %}""" self.assertTrue(self.validate_data('staging-thunderx-01', data)) - test_template = prepare_jinja_template('staging-thunderx-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-thunderx-01', data, raw=False) self.assertIn('character_delays', template_dict) self.assertIn('boot', template_dict['character_delays']) self.assertEqual(150, template_dict['character_delays']['boot']) @@ -162,9 +147,7 @@ class TestTemplates(BaseTemplate.BaseTemplateCases): {% set power_on_command = 'ipmitool -H calxeda02-07-02 -U admin -P admin chassis power on' %} {% set hard_reset_command = 'ipmitool -H calxeda02-07-02 -U admin -P admin chassis power off; sleep 20; ipmitool -H calxeda02-07-02 -U admin -P admin chassis power on' %}""" self.assertTrue(self.validate_data('highbank-07', data)) - test_template = prepare_jinja_template('highbank-07', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('highbank-07', data, raw=False) self.assertIsNotNone(template_dict) self.assertEqual(template_dict['character_delays']['boot'], 100) self.assertEqual(template_dict['actions']['boot']['methods']['u-boot']['parameters']['bootloader_prompt'], 'Highbank') @@ -218,15 +201,12 @@ class TestTemplates(BaseTemplate.BaseTemplateCases): {% set connection_command = 'telnet localhost 7301' %} {% set lava_mac = 'd8:9d:67:26:ae:e8' %}""" self.assertTrue(self.validate_data('staging-x86-01', data)) - test_template = prepare_jinja_template('staging-qemu-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-qemu-01', data, raw=False) self.assertIn( 'set console console=ttyS0,115200n8 lava_mac={LAVA_MAC}', template_dict['actions']['boot']['methods']['ipxe']['nfs']['commands']) context = {'extra_kernel_args': 'intel_mmio=on mmio=on'} - rendered = test_template.render(**context) - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-qemu-01', data, job_ctx=context, raw=False) self.assertIn( 'set extraargs root=/dev/nfs rw nfsroot={NFS_SERVER_IP}:{NFSROOTFS},tcp,hard,intr intel_mmio=on mmio=on ip=dhcp', template_dict['actions']['boot']['methods']['ipxe']['nfs']['commands']) @@ -235,10 +215,8 @@ class TestTemplates(BaseTemplate.BaseTemplateCases): data = """{% extends 'arduino101.jinja2' %} {% set board_id = 'AE6642EK61804EZ' %}""" self.assertTrue(self.validate_data('staging-arduino101-01', data)) - test_template = prepare_jinja_template('staging-arduino101-01', - data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template( + 'staging-arduino101-01', data, raw=False) self.assertIsNotNone(template_dict) self.assertEqual(template_dict['board_id'], 'AE6642EK61804EZ') @@ -256,9 +234,7 @@ class TestTemplates(BaseTemplate.BaseTemplateCases): {% set console_device = console_device | default('ttyS2') %} {% set extra_kernel_args = 'earlyprintk=ttyS2,115200n8 console=tty1' %}""" self.assertTrue(self.validate_data('jaq-01', data)) - test_template = prepare_jinja_template('jaq-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('jaq-01', data, raw=False) load_addr = template_dict['parameters']['load_address'] self.assertEqual(load_addr, '0x5678') depthcharge = template_dict['actions']['boot']['methods']['depthcharge'] @@ -268,8 +244,7 @@ class TestTemplates(BaseTemplate.BaseTemplateCases): depthcharge['ramdisk']['cmdline']) def test_x86_atom330_template(self): - rendered = self.render_device_dictionary_file('x86-atom330-01.jinja2') - template_dict = yaml.load(rendered) + template_dict = self.render_device_dictionary_file('x86-atom330-01.jinja2', raw=False) self.assertIsNotNone(template_dict['actions']['boot']['methods']['ipxe']) self.assertIn('ramdisk', template_dict['actions']['boot']['methods']['ipxe']) self.assertIn('commands', template_dict['actions']['boot']['methods']['ipxe']['nfs']) diff --git a/lava_scheduler_app/tests/test_uboot_templates.py b/lava_scheduler_app/tests/test_uboot_templates.py index 1e2b43303..79b69e605 100644 --- a/lava_scheduler_app/tests/test_uboot_templates.py +++ b/lava_scheduler_app/tests/test_uboot_templates.py @@ -54,9 +54,7 @@ class TestUbootTemplates(BaseTemplate.BaseTemplateCases): {% set uboot_bootx_cmd = "bootm {KERNEL_ADDR} {RAMDISK_ADDR}" %} """ self.assertTrue(self.validate_data('armada-375-01', data)) - test_template = prepare_jinja_template('armada-375-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('armada-375-01', data, raw=False) params = template_dict['actions']['deploy']['parameters'] self.assertIsNotNone(params) self.assertIn('use_xip', params) @@ -81,25 +79,19 @@ class TestUbootTemplates(BaseTemplate.BaseTemplateCases): {% set power_on_command = '/usr/bin/pduclient --daemon localhost --hostname lngpdu01 --command on --port 19' %} {% set connection_command = 'telnet localhost 7333' %}""" self.assertTrue(self.validate_data('staging-bbb-01', data)) - test_template = prepare_jinja_template('staging-bbb-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-bbb-01', data, raw=False) self.assertIsNotNone(template_dict['actions']['deploy']['methods']['ssh']['host']) self.assertEqual('', template_dict['actions']['deploy']['methods']['ssh']['host']) self.assertNotEqual('None', template_dict['actions']['deploy']['methods']['ssh']['host']) data += "{% set ssh_host = '192.168.0.10' %}" - test_template = prepare_jinja_template('staging-bbb-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-bbb-01', data, raw=False) self.assertIsNotNone(template_dict['actions']['deploy']['methods']['ssh']['host']) self.assertEqual('192.168.0.10', template_dict['actions']['deploy']['methods']['ssh']['host']) def test_b2260_template(self): data = """{% extends 'b2260.jinja2' %}""" self.assertTrue(self.validate_data('staging-b2260-01', data)) - test_template = prepare_jinja_template('staging-b2260-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-b2260-01', data, raw=False) self.assertEqual({'seconds': 15}, template_dict['timeouts']['actions']['power-off']) def test_mustang_template(self): @@ -109,9 +101,7 @@ class TestUbootTemplates(BaseTemplate.BaseTemplateCases): {% set power_off_command = '/usr/bin/pduclient --daemon staging-master --hostname pdu15 --command off --port 05' %} {% set power_on_command = '/usr/bin/pduclient --daemon staging-master --hostname pdu15 --command on --port 05' %}""" self.assertTrue(self.validate_data('staging-mustang-01', data)) - test_template = prepare_jinja_template('staging-mustang-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-mustang-01', data, raw=False) self.assertIsInstance(template_dict['parameters']['text_offset'], str) commands = template_dict['actions']['boot']['methods']['u-boot']['ramdisk']['commands'] for line in commands: @@ -127,9 +117,7 @@ class TestUbootTemplates(BaseTemplate.BaseTemplateCases): # test appending to kernel args context = {'extra_kernel_args': 'extra_arg=extra_val'} - test_template = prepare_jinja_template('staging-rpi3-01', data) - rendered = test_template.render(**context) - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-rpi3-01', data, job_ctx=context, raw=False) commands = template_dict['actions']['boot']['methods']['u-boot']['ramdisk']['commands'] self.assertIsNotNone(commands) self.assertIsInstance(commands, list) @@ -143,9 +131,7 @@ class TestUbootTemplates(BaseTemplate.BaseTemplateCases): # test overwriting kernel args checked = False context = {'custom_kernel_args': 'custom_arg=custom_val'} - test_template = prepare_jinja_template('staging-rpi3-01', data) - rendered = test_template.render(**context) - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-rpi3-01', data, job_ctx=context, raw=False) commands = template_dict['actions']['boot']['methods']['u-boot']['ramdisk']['commands'] self.assertIsNotNone(commands) self.assertIsInstance(commands, list) @@ -168,9 +154,7 @@ class TestUbootTemplates(BaseTemplate.BaseTemplateCases): {% set power_on_command = '/usr/bin/pduclient --daemon staging-master --hostname pdu15 --command on --port 05' %}""" self.assertTrue(self.validate_data('staging-panda-01', data)) context = {'extra_kernel_args': 'intel_mmio=on mmio=on'} - test_template = prepare_jinja_template('staging-panda-01', data) - rendered = test_template.render(**context) - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-panda-01', data, job_ctx=context, raw=False) self.assertIn('bootloader-commands', template_dict['timeouts']['actions']) self.assertEqual(180.0, Timeout.parse(template_dict['timeouts']['actions']['bootloader-commands'])) commands = template_dict['actions']['boot']['methods']['u-boot']['ramdisk']['commands'] @@ -203,9 +187,7 @@ class TestUbootTemplates(BaseTemplate.BaseTemplateCases): " - setenv bootcmd 'dhcp; setenv serverip {SERVER_IP}; run loadkernel; run loadinitrd; run loadfdt; {BOOTX}' - boot") %}""" self.assertTrue(self.validate_data('staging-juno-01', data)) - test_template = prepare_jinja_template('staging-juno-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-juno-01', data, raw=False) self.assertIsNotNone(template_dict) def test_cubietruck_template(self): @@ -218,9 +200,7 @@ class TestUbootTemplates(BaseTemplate.BaseTemplateCases): {% set connection_command = 'telnet localhost 6002' %} {% set console_device = 'ttyfake1' %}""" self.assertTrue(self.validate_data('staging-cubietruck-01', data)) - test_template = prepare_jinja_template('staging-cubietruck-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-cubietruck-01', data, raw=False) self.assertIsNotNone(template_dict) self.assertIn('u-boot', template_dict['actions']['boot']['methods']) self.assertIn('SanDisk_Ultra', template_dict['parameters']['media']['usb']) @@ -242,17 +222,13 @@ class TestUbootTemplates(BaseTemplate.BaseTemplateCases): {% set power_off_command = '/usr/bin/pduclient --daemon staging-master --hostname pdu15 --command off --port 05' %} {% set power_on_command = '/usr/bin/pduclient --daemon staging-master --hostname pdu15 --command on --port 05' %}""" job_ctx = {} - test_template = prepare_jinja_template('staging-panda-01', data) - rendered = test_template.render(**job_ctx) - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-panda-01', data, job_ctx=job_ctx, raw=False) for line in template_dict['actions']['boot']['methods']['u-boot']['nfs']['commands']: if line.startswith("setenv nfsargs"): self.assertIn(',tcp,hard,intr ', line) self.assertNotIn('nfsvers', line) job_ctx = {'extra_nfsroot_args': ',nolock,nfsvers=3'} - test_template = prepare_jinja_template('staging-panda-01', data) - rendered = test_template.render(**job_ctx) - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-panda-01', data, job_ctx=job_ctx, raw=False) for line in template_dict['actions']['boot']['methods']['u-boot']['nfs']['commands']: if line.startswith("setenv nfsargs"): self.assertIn(',tcp,hard,intr,nolock,nfsvers=3 ', line) @@ -277,9 +253,7 @@ class TestUbootTemplates(BaseTemplate.BaseTemplateCases): {% set power_on_command = '/usr/local/lab-scripts/snmp_pdu_control --hostname lngpdu01 --command on --port 19' %} {% set connection_command = 'telnet localhost 7333' %}""" self.assertTrue(self.validate_data('staging-x86-01', data)) - test_template = prepare_jinja_template('staging-qemu-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-qemu-01', data, raw=False) self.assertIn('interfaces', template_dict['parameters']) self.assertIn('iface0', template_dict['parameters']['interfaces']) self.assertIn('port', template_dict['parameters']['interfaces']['iface0']) @@ -304,9 +278,7 @@ class TestUbootTemplates(BaseTemplate.BaseTemplateCases): {% set connection_command = 'telnet serial4 7010' %} {% set power_on_command = '/usr/local/lab-scripts/snmp_pdu_control --hostname pdu15 --command on --port 07' %}""" self.assertTrue(self.validate_data('staging-panda-01', data)) - test_template = prepare_jinja_template('staging-panda-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-panda-01', data, raw=False) fdesc, device_yaml = tempfile.mkstemp() os.write(fdesc, yaml.dump(template_dict).encode()) panda = NewDevice(device_yaml) @@ -328,9 +300,7 @@ class TestUbootTemplates(BaseTemplate.BaseTemplateCases): {% set power_on_command = '/usr/local/lab-scripts/snmp_pdu_control --port 14 --hostname pdu18 --command on' %} {% set uboot_mac_addr = '00:80:e1:12:81:30' %}""" self.assertTrue(self.validate_data('staging-b2260-01', data)) - test_template = prepare_jinja_template('staging-b2260-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-b2260-01', data, raw=False) ethaddr = False for command in template_dict['actions']['boot']['methods']['u-boot']['ramdisk']['commands']: if command.startswith('setenv ethaddr'): @@ -351,9 +321,7 @@ class TestUbootTemplates(BaseTemplate.BaseTemplateCases): {% set connection_command = 'telnet serial4 7010' %} {% set power_on_command = '/usr/local/lab-scripts/snmp_pdu_control --hostname pdu15 --command on --port 07' %}""" self.assertTrue(self.validate_data('staging-arndale-01', data)) - test_template = prepare_jinja_template('staging-panda-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-panda-01', data, raw=False) for line in template_dict['actions']['boot']['methods']['u-boot']['ramdisk']['commands']: if line.startswith("setenv nfsargs"): self.assertIn('ip=:::::eth0:dhcp', line) @@ -370,10 +338,8 @@ class TestUbootTemplates(BaseTemplate.BaseTemplateCases): {% set power_on_command = '/usr/bin/pduclient --daemon services --hostname pdu09 --command on --port 07' %} {% set boot_character_delay = 30 %}""" self.assertTrue(self.validate_data('staging-d03-01', data)) - test_template = prepare_jinja_template('staging-d03-01', - data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-d03-01', + data, raw=False) self.assertIn('character_delays', template_dict) self.assertIn('boot', template_dict['character_delays']) self.assertNotIn('test', template_dict['character_delays']) @@ -392,9 +358,9 @@ class TestUbootTemplates(BaseTemplate.BaseTemplateCases): " - setenv bootcmd 'dhcp; setenv serverip {SERVER_IP}; run loadkernel; run loadinitrd; run loadfdt; {BOOTX}' - boot") %}""" self.assertTrue(self.validate_data('staging-juno-01', data)) - test_template = prepare_jinja_template('staging-juno-01', data) + test_template = prepare_jinja_template('staging-juno-01', data, raw=True) rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = yaml.safe_load(rendered) self.assertIsNotNone(template_dict) self.assertEqual({'boot': 30}, template_dict['character_delays']) self.assertIn('error-messages', template_dict['constants']['u-boot']) @@ -403,7 +369,7 @@ class TestUbootTemplates(BaseTemplate.BaseTemplateCases): self.assertEqual('Start:', template_dict['actions']['boot']['methods']['uefi-menu']['parameters']['bootloader_prompt']) rendered = test_template.render(bootloader_prompt="vexpress>") - template_dict = yaml.load(rendered) + template_dict = yaml.safe_load(rendered) self.assertIsNotNone(template_dict) self.assertEqual({'boot': 30}, template_dict['character_delays']) self.assertIn('error-messages', template_dict['constants']['u-boot']) @@ -431,14 +397,14 @@ class TestUbootTemplates(BaseTemplate.BaseTemplateCases): 'userdata'] rendered = self.render_device_dictionary_file('imx8m-01.jinja2') - template_dict = yaml.load(rendered) + template_dict = yaml.safe_load(rendered) self.assertIsNotNone(template_dict) self.assertIn('error-messages', template_dict['constants']['u-boot']) self.assertEqual('u-boot=>', template_dict['actions']['boot']['methods']['u-boot']['parameters']['bootloader_prompt']) context = {'bootloader_prompt': 'imx8m=>'} rendered = self.render_device_dictionary_file('imx8m-01.jinja2', context) - template_dict = yaml.load(rendered) + template_dict = yaml.safe_load(rendered) self.assertIsNotNone(template_dict) self.assertIn('error-messages', template_dict['constants']['u-boot']) self.assertEqual('imx8m=>', template_dict['actions']['boot']['methods']['u-boot']['parameters']['bootloader_prompt']) @@ -450,7 +416,7 @@ class TestUbootTemplates(BaseTemplate.BaseTemplateCases): checked = False context = {'console_device': 'ttyUSB1'} rendered = self.render_device_dictionary_file('imx8m-01.jinja2', context) - template_dict = yaml.load(rendered) + template_dict = yaml.safe_load(rendered) commands = template_dict['actions']['boot']['methods']['u-boot']['ramdisk']['commands'] self.assertIsNotNone(commands) self.assertIsInstance(commands, list) @@ -464,9 +430,7 @@ class TestUbootTemplates(BaseTemplate.BaseTemplateCases): with open(os.path.join(os.path.dirname(__file__), 'devices', 'zcu102.jinja2')) as zcu: data = zcu.read() self.assertTrue(self.validate_data('zcu-01', data)) - test_template = prepare_jinja_template('zcu-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('zcu-01', data, raw=False) self.assertIn('u-boot', template_dict['actions']['boot']['methods']) self.assertIn('ramdisk', template_dict['actions']['boot']['methods']['u-boot']) commands = template_dict['actions']['boot']['methods']['u-boot']['ramdisk']['commands'] @@ -490,9 +454,7 @@ class TestUbootTemplates(BaseTemplate.BaseTemplateCases): {% set flasher_deploy_commands = ['flashing', 'something --else'] %} """ self.assertTrue(self.validate_data('staging-b2260-01', data)) - test_template = prepare_jinja_template('staging-b2260-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-b2260-01', data, raw=False) self.assertEqual(['flashing', 'something --else'], template_dict['actions']['deploy']['methods']['flasher']['commands']) @@ -502,16 +464,13 @@ class TestUbootTemplates(BaseTemplate.BaseTemplateCases): 'set_boot_to_sd': {'do': '/bin/true', 'undo': '/bin/true'}} %} """ self.assertTrue(self.validate_data('staging-b2260-01', data)) - test_template = prepare_jinja_template('staging-b2260-01', data) - rendered = test_template.render() - template_dict = yaml.load(rendered) + template_dict = prepare_jinja_template('staging-b2260-01', data, raw=False) self.assertEqual({'set_boot_to_usb': {'do': '/bin/true', 'undo': '/bin/true'}, 'set_boot_to_sd': {'do': '/bin/true', 'undo': '/bin/true'}}, template_dict['commands']['users']) def test_meson8b_template(self): - rendered = self.render_device_dictionary_file('meson8b-odroidc1-1.jinja2') - template_dict = yaml.load(rendered) + template_dict = self.render_device_dictionary_file('meson8b-odroidc1-1.jinja2', raw=False) self.assertIsNotNone(template_dict) template_dict['constants']['u-boot'].get('interrupt_ctrl_list', self.fail) self.assertEqual(template_dict['constants']['u-boot']['interrupt_ctrl_list'], ['c']) diff --git a/lava_scheduler_app/tests/test_vlan.py b/lava_scheduler_app/tests/test_vlan.py index 32739dd0a..40d69672d 100644 --- a/lava_scheduler_app/tests/test_vlan.py +++ b/lava_scheduler_app/tests/test_vlan.py @@ -5,13 +5,11 @@ import tempfile from lava_scheduler_app.utils import split_multinode_yaml from lava_scheduler_app.dbutils import match_vlan_interface from lava_scheduler_app.models import ( - Device, TestJob, Tag, ) from lava_scheduler_app.tests.test_submission import TestCaseWithFactory from lava_scheduler_app.tests.test_pipeline import YamlFactory -from lava_scheduler_app.dbutils import match_vlan_interface from lava_dispatcher.device import NewDevice from lava_dispatcher.parser import JobParser from lava_dispatcher.protocols.vland import VlandProtocol @@ -39,7 +37,7 @@ class VlandFactory(YamlFactory): def make_vland_job(self, **kw): sample_job_file = os.path.join(os.path.dirname(__file__), 'sample_jobs', 'bbb-cubie-vlan-group.yaml') with open(sample_job_file, 'r') as test_support: - data = yaml.load(test_support) + data = yaml.safe_load(test_support) data.update(kw) return data @@ -105,12 +103,12 @@ class TestVlandDevices(TestCaseWithFactory): user = self.factory.make_user() sample_job_file = os.path.join(os.path.dirname(__file__), 'sample_jobs', 'bbb-cubie-vlan-group.yaml') with open(sample_job_file, 'r') as test_support: - data = yaml.load(test_support) + data = yaml.safe_load(test_support) vlan_job = TestJob.from_yaml_and_user(yaml.dump(data), user) assignments = {} for job in vlan_job: - self.assertFalse(match_vlan_interface(self.bbb3, yaml.load(job.definition))) - self.assertFalse(match_vlan_interface(self.cubie2, yaml.load(job.definition))) + self.assertFalse(match_vlan_interface(self.bbb3, yaml.safe_load(job.definition))) + self.assertFalse(match_vlan_interface(self.cubie2, yaml.safe_load(job.definition))) def test_jinja_template(self): yaml_data = self.factory.bbb1.load_configuration() |