1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
|
import os
import yaml
import logging
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
from lava_dispatcher.protocols.multinode import MultinodeProtocol
# pylint does not like TestCaseWithFactory
# pylint: disable=too-many-ancestors,no-self-use,too-many-statements
# pylint: disable=superfluous-parens,too-many-locals
class VlandFactory(YamlFactory):
def __init__(self):
super().__init__()
self.bbb1 = None
self.cubie1 = None
self.bbb_type = None
def setUp(self): # pylint: disable=invalid-name
self.bbb_type = self.make_device_type(name='bbb')
self.cubie_type = self.make_device_type(name='cubietruck')
self.bbb1 = self.make_device(self.bbb_type, hostname='bbb-01')
self.cubie1 = self.make_device(self.cubie_type, hostname='cubie1')
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.update(kw)
return data
class TestVlandSplit(TestCaseWithFactory):
"""
Test the splitting of lava-vland data from submission YAML
Same tests as test_submission but converted to use and look for YAML.
"""
def setUp(self):
super().setUp()
self.factory = VlandFactory()
def test_split_vland(self):
target_group = "unit-test-only"
job_dict = split_multinode_yaml(self.factory.make_vland_job(), target_group)
self.assertEqual(len(job_dict), 2)
roles = job_dict.keys()
self.assertEqual({'server', 'client'}, set(roles))
for role in roles:
self.assertEqual(len(job_dict[role]), 1) # count = 1
client_job = job_dict['client'][0]
server_job = job_dict['server'][0]
self.assertIn('lava-multinode', client_job['protocols'])
self.assertIn('lava-multinode', server_job['protocols'])
self.assertIn('lava-vland', client_job['protocols'])
self.assertIn('lava-vland', server_job['protocols'])
client_vlan = client_job['protocols']['lava-vland']
server_vlan = server_job['protocols']['lava-vland']
self.assertIn('vlan_one', client_vlan)
self.assertIn('vlan_two', server_vlan)
self.assertEqual(['RJ45', '10M'], list(client_vlan.values())[0]['tags'])
self.assertEqual(['RJ45', '100M'],
list(server_vlan.values())[0]['tags'])
class TestVlandDevices(TestCaseWithFactory):
"""
Test the matching of vland device requirements with submission YAML
"""
def setUp(self):
super().setUp()
self.factory = VlandFactory()
self.factory.setUp()
logger = logging.getLogger('lava-master')
logger.disabled = True
def test_match_devices_without_map(self):
"""
Without a map, there is no support for knowing which interfaces to
put onto a VLAN, so these devices cannot be assigned to a VLAN testjob
See http://localhost/static/docs/v2/vland.html#vland-and-interface-tags-in-lava
"""
self.bbb3 = self.factory.make_device(self.factory.bbb_type, hostname='bbb-03')
self.cubie2 = self.factory.make_device(self.factory.cubie_type, hostname='cubie2')
devices = [self.bbb3, self.cubie2]
self.factory.ensure_tag('usb-eth')
self.factory.ensure_tag('sata')
self.factory.bbb1.tags = Tag.objects.filter(name='usb-eth')
self.factory.bbb1.save()
self.factory.cubie1.tags = Tag.objects.filter(name='sata')
self.factory.cubie1.save()
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)
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)))
def test_jinja_template(self):
yaml_data = self.factory.bbb1.load_configuration()
self.assertIn('parameters', yaml_data)
self.assertIn('interfaces', yaml_data['parameters'])
self.assertIn('bootm', yaml_data['parameters'])
self.assertIn('bootz', yaml_data['parameters'])
self.assertIn('actions', yaml_data)
self.assertIn('eth0', yaml_data['parameters']['interfaces'])
self.assertIn('eth1', yaml_data['parameters']['interfaces'])
self.assertIn('sysfs', yaml_data['parameters']['interfaces']['eth0'])
self.assertEqual(
'/sys/devices/pci0000:00/0000:00:1c.1/0000:03:00.0/net/eth1',
yaml_data['parameters']['interfaces']['eth1']['sysfs']
)
class TestVlandProtocolSplit(TestCaseWithFactory):
"""
Test the handling of protocols in dispatcher after splitting the YAML
"""
def setUp(self):
super().setUp()
self.factory = VlandFactory()
self.factory.setUp()
def test_job_protocols(self):
self.factory.ensure_tag('usb-eth')
self.factory.ensure_tag('sata')
self.factory.bbb1.tags = Tag.objects.filter(name='usb-eth')
self.factory.bbb1.save()
self.factory.cubie1.tags = Tag.objects.filter(name='sata')
self.factory.cubie1.save()
target_group = "unit-test-only"
job_dict = split_multinode_yaml(self.factory.make_vland_job(), target_group)
client_job = job_dict['client'][0]
client_handle, client_file_name = tempfile.mkstemp()
yaml.dump(client_job, open(client_file_name, 'w'))
# YAML device file, as required by lava-dispatch --target
device_yaml_file = os.path.realpath(os.path.join(os.path.dirname(__file__), 'devices', 'bbb-01.yaml'))
self.assertTrue(os.path.exists(device_yaml_file))
parser = JobParser()
bbb_device = NewDevice(device_yaml_file)
with open(client_file_name) as sample_job_data:
bbb_job = parser.parse(sample_job_data, bbb_device, 4212, None, "")
os.close(client_handle)
os.unlink(client_file_name)
self.assertIn('protocols', bbb_job.parameters)
self.assertIn(VlandProtocol.name, bbb_job.parameters['protocols'])
self.assertIn(MultinodeProtocol.name, bbb_job.parameters['protocols'])
|
