Gitiles
Code Review Sign In
gerrit.opencord.org / pyvoltha / 67b674aba8766c077c94c3e7ad703a8ae3601a5d / . / test / unit / extensions / omci / test_me_frame.py
blob: 74e2b78c8c9e88b39ac5523342d75b78ae68d8b9 [file] [log] [blame]
Chip Boling67b674a2019-02-08 11:42:18 -0600[diff] [blame^]1#
2# Copyright 2017 the original author or authors.
3#
4# Licensed under the Apache License, Version 2.0 (the "License");
5# you may not use this file except in compliance with the License.
6# You may obtain a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15#
16from unittest import TestCase, main
17from nose.tools import assert_raises
18from pyvoltha.adapters.extensions.omci.me_frame import *
19from pyvoltha.adapters.extensions.omci.omci_me import *
20from pyvoltha.adapters.extensions.omci.omci import *
21
22
23def hexify(buffer):
24 """Return a hexadecimal string encoding of input buffer"""
25 return ''.join('%02x' % ord(c) for c in buffer)
26
27
28class TestSelectMeFrameGeneration(TestCase):
29
30 def assertGeneratedFrameEquals(self, frame, ref):
31 assert isinstance(frame, Packet)
32 serialized_hexified_frame = hexify(str(frame)).upper()
33 ref = ref.upper()
34 if serialized_hexified_frame != ref:
35 self.fail('Mismatch:\nReference:\n{}\nGenerated (bad):\n{}'.format(
36 ref, serialized_hexified_frame
37 ))
38
39 def test_mib_reset_message_serialization(self):
40 ref = '00004F0A000200000000000000000000' \
41 '00000000000000000000000000000000' \
42 '000000000000000000000028'
43 frame = OntDataFrame().mib_reset()
44 self.assertGeneratedFrameEquals(frame, ref)
45
46 def test_create_gal_ethernet_profile(self):
47 ref = '0000440A011000010030000000000000' \
48 '00000000000000000000000000000000' \
49 '000000000000000000000028'
50 frame = GalEthernetProfileFrame(1, max_gem_payload_size=48).create()
51 self.assertGeneratedFrameEquals(frame, ref)
52
53 def test_set_tcont_1(self):
54 ref = '0000480A010680008000040000000000' \
55 '00000000000000000000000000000000' \
56 '000000000000000000000028'
57
58 frame = TcontFrame(0x8000, alloc_id=0x400).set()
59 self.assertGeneratedFrameEquals(frame, ref)
60
61 def test_set_tcont_2(self):
62 ref = '0000480A010680018000040100000000' \
63 '00000000000000000000000000000000' \
64 '000000000000000000000028'
65
66 frame = TcontFrame(0x8001, alloc_id=0x401).set()
67 self.assertGeneratedFrameEquals(frame, ref)
68
69 def test_create_8021p_mapper_service_profile(self):
70 ref = '0000440A00828000ffffffffffffffff' \
71 'ffffffffffffffffffff000000000000' \
72 '000000000000000000000028'
73 frame = Ieee8021pMapperServiceProfileFrame(0x8000).create()
74 self.assertGeneratedFrameEquals(frame, ref)
75
76 def test_create_mac_bridge_service_profile(self):
77 ref = '0000440A002D02010001008000140002' \
78 '000f0001000000000000000000000000' \
79 '000000000000000000000028'
80 data = dict(
81 spanning_tree_ind=False,
82 learning_ind=True,
83 priority=0x8000,
84 max_age=20 * 256,
85 hello_time=2 * 256,
86 forward_delay=15 * 256,
87 unknown_mac_address_discard=True
88 )
89 frame = MacBridgeServiceProfileFrame(0x201, attributes=data).create()
90 self.assertGeneratedFrameEquals(frame, ref)
91
92 def test_create_gem_port_network_ctp(self):
93 ref = '0000440A010C01000400800003010000' \
94 '00000000000000000000000000000000' \
95 '000000000000000000000028'
96
97 data = dict(
98 port_id=0x400,
99 tcont_pointer=0x8000,
100 direction=3,
101 traffic_management_pointer_upstream=0x100
102 )
103 frame = GemPortNetworkCtpFrame(0x100, attributes=data).create()
104 self.assertGeneratedFrameEquals(frame, ref)
105
106 # Also test direction as a string parameter
107 frame = GemPortNetworkCtpFrame(0x100, port_id=0x400,
108 tcont_id=0x8000,
109 direction='bi-directional',
110 upstream_tm=0x100).create()
111 self.assertGeneratedFrameEquals(frame, ref)
112
113 def test_create_gem_inteworking_tp(self):
114 ref = '0000440A010A80010100058000000000' \
115 '01000000000000000000000000000000' \
116 '000000000000000000000028'
117 frame = GemInterworkingTpFrame(0x8001,
118 gem_port_network_ctp_pointer=0x100,
119 interworking_option=5,
120 service_profile_pointer=0x8000,
121 interworking_tp_pointer=0x0,
122 gal_profile_pointer=0x1).create()
123
124 self.assertGeneratedFrameEquals(frame, ref)
125
126 def test_set_8021p_mapper_service_profile(self):
127 ref = '0000480A008280007F80800100000000' \
128 '00000000000000000000000000000000' \
129 '000000000000000000000028'
130 ptrs = [0x8001, 0, 0, 0, 0, 0, 0, 0]
131 frame = Ieee8021pMapperServiceProfileFrame(0x8000,
132 interwork_tp_pointers=ptrs).set()
133
134 self.assertGeneratedFrameEquals(frame, ref)
135
136 ptrs = [0x8001, 0]
137 frame = Ieee8021pMapperServiceProfileFrame(0x8000,
138 interwork_tp_pointers=ptrs).set()
139
140 self.assertGeneratedFrameEquals(frame, ref)
141
142 def test_create_mac_bridge_port_configuration_data(self):
143 ref = '0000440A002F21010201020380000000' \
144 '00000000000000000000000000000000' \
145 '000000000000000000000028'
146
147 frame = MacBridgePortConfigurationDataFrame(0x2101,
148 bridge_id_pointer=0x201,
149 port_num=2,
150 tp_type=3,
151 tp_pointer=0x8000).create()
152 self.assertGeneratedFrameEquals(frame, ref)
153
154 def test_create_vlan_tagging_filter_data(self):
155 ref = '0000440A005421010400000000000000' \
156 '00000000000000000000000000000000' \
157 '100100000000000000000028'
158 frame = VlanTaggingFilterDataFrame(0x2101,
159 vlan_tcis=[0x400],
160 forward_operation=0x10).create()
161 self.assertGeneratedFrameEquals(frame, ref)
162
163 def test_create_extended_vlan_tagging_operation_configuration_data(self):
164 ref = '0000440A00AB02020A04010000000000' \
165 '00000000000000000000000000000000' \
166 '000000000000000000000028'
167 data = dict(
168 association_type=10,
169 associated_me_pointer=0x401
170 )
171 frame = \
172 ExtendedVlanTaggingOperationConfigurationDataFrame(0x202,
173 attributes=data)\
174 .create()
175
176 self.assertGeneratedFrameEquals(frame, ref)
177
178 def test_set_extended_vlan_tagging_operation_configuration_data(self):
179 ref = '0000480A00AB02023800810081000000' \
180 '00000000000000000000000000000000' \
181 '000000000000000000000028'
182 data = dict(
183 input_tpid=0x8100,
184 output_tpid=0x8100,
185 downstream_mode=0, # inverse of upstream
186 )
187 frame = \
188 ExtendedVlanTaggingOperationConfigurationDataFrame(0x202,
189 attributes=data)\
190 .set()
191
192 self.assertGeneratedFrameEquals(frame, ref)
193
194 def test_set_extended_vlan_tagging_1(self):
195 ref = '0000480A00AB02020400f00000008200' \
196 '5000402f000000082004000000000000' \
197 '000000000000000000000028'
198 data = dict(
199 received_frame_vlan_tagging_operation_table=\
200 VlanTaggingOperation(
201 filter_outer_priority=15,
202 filter_inner_priority=8,
203 filter_inner_vid=1024,
204 filter_inner_tpid_de=5,
205 filter_ether_type=0,
206 treatment_tags_to_remove=1,
207 pad3=2,
208 treatment_outer_priority=15,
209 treatment_inner_priority=8,
210 treatment_inner_vid=1024,
211 treatment_inner_tpid_de=4
212 )
213 )
214 frame = \
215 ExtendedVlanTaggingOperationConfigurationDataFrame(0x202,
216 attributes=data)\
217 .set()
218
219 self.assertGeneratedFrameEquals(frame, ref)
220
221 def test_set_extended_vlan_tagging_2(self):
222 ref = '0000480A00AB02020400F00000008200' \
223 'd000402f00000008200c000000000000' \
224 '000000000000000000000028'
225 data = dict(
226 received_frame_vlan_tagging_operation_table=
227 VlanTaggingOperation(
228 filter_outer_priority=15,
229 filter_inner_priority=8,
230 filter_inner_vid=1025,
231 filter_inner_tpid_de=5,
232 filter_ether_type=0,
233 treatment_tags_to_remove=1,
234 pad3=2,
235 treatment_outer_priority=15,
236 treatment_inner_priority=8,
237 treatment_inner_vid=1025,
238 treatment_inner_tpid_de=4
239 )
240 )
241
242 frame = \
243 ExtendedVlanTaggingOperationConfigurationDataFrame(0x202,
244 attributes=data)\
245 .set()
246
247 self.assertGeneratedFrameEquals(frame, ref)
248
249 def test_create_mac_bridge_port_configuration_data2(self):
250 ref = '0000440A002F02010201010b04010000' \
251 '00000000000000000000000000000000' \
252 '000000000000000000000028'
253 data = dict(
254 bridge_id_pointer=0x201,
255 encapsulation_methods=0,
256 port_num=1,
257 port_priority=0,
258 port_path_cost=0,
259 port_spanning_tree_in=0,
260 lan_fcs_ind=0,
261 tp_type=11,
262 tp_pointer=0x401,
263 mac_learning_depth=0
264 )
265 frame = MacBridgePortConfigurationDataFrame(0x201,
266 attributes=data).create()
267
268 self.assertGeneratedFrameEquals(frame, ref)
269
270 def test_set_pptp_ethernet_uni_frame(self):
271 ref = '0000480A000B020109000005EE000000' \
272 '00000000000000000000000000000000' \
273 '000000000000000000000028'
274 data = dict(
275 administrative_state=0, # 0 - Unlock
276 max_frame_size=1518 # two-octet field
277 )
278 frame = PptpEthernetUniFrame(0x201,
279 attributes=data).set()
280
281 self.assertGeneratedFrameEquals(frame, ref)
282
283 def test_constraint_errors(self):
284 self.assertTrue(True) # TODO Also test some attribute constraint failures
285
286 def test_mib_upload_next(self):
287 # Test for VOL-649 error. SCAPY was only originally coded for a 'get'
288 # action (8-bit MIB Data Sync value) but MIB Upload Next commands have
289 # a 16-bit field.
290 #
291 # 255 and less always worked
292 OntDataFrame(sequence_number=0).mib_upload_next()
293 OntDataFrame(sequence_number=255).mib_upload_next()
294 # But not 256+
295 OntDataFrame(sequence_number=256).mib_upload_next()
296 OntDataFrame(sequence_number=1000).mib_upload_next()
297 OntDataFrame(sequence_number=0xFFFE).mib_upload_next()
298
299 # Also test the optional arguments for the other actions
300 OntDataFrame().get()
301 OntDataFrame(mib_data_sync=4).set()
302 OntDataFrame().mib_reset()
303 OntDataFrame().mib_upload()
304 # OntDataFrame(ignore_arc=True).get_all_alarms() Not yet coded
305 # OntDataFrame(ignore_arc=False).get_all_alarms() Not yet coded
306
307 # Range/type checks
308 assert_raises(ValueError, OntDataFrame, mib_data_sync=-1)
309 assert_raises(ValueError, OntDataFrame, mib_data_sync=256)
310 assert_raises(ValueError, OntDataFrame, sequence_number=-1)
311 assert_raises(ValueError, OntDataFrame, sequence_number=0x10000)
312 assert_raises(TypeError, OntDataFrame, ignore_arc=123)
313
314
315if __name__ == '__main__':
316 main()
317