aboutsummaryrefslogtreecommitdiff
path: root/lava_scheduler_app
diff options
context:
space:
mode:
Diffstat (limited to 'lava_scheduler_app')
-rw-r--r--lava_scheduler_app/tests/test_api.py58
-rw-r--r--lava_scheduler_app/tests/test_base_templates.py62
-rw-r--r--lava_scheduler_app/tests/test_connections.py14
-rw-r--r--lava_scheduler_app/tests/test_device.py4
-rw-r--r--lava_scheduler_app/tests/test_fastboot_templates.py72
-rw-r--r--lava_scheduler_app/tests/test_grub_templates.py30
-rw-r--r--lava_scheduler_app/tests/test_menus.py4
-rw-r--r--lava_scheduler_app/tests/test_pipeline.py85
-rw-r--r--lava_scheduler_app/tests/test_qemu_templates.py32
-rw-r--r--lava_scheduler_app/tests/test_templates.py55
-rw-r--r--lava_scheduler_app/tests/test_uboot_templates.py97
-rw-r--r--lava_scheduler_app/tests/test_vlan.py10
12 files changed, 195 insertions, 328 deletions
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()
generated by cgit v1.2.3 (git 2.25.1) at 2024-06-29 06:38:35 +0000