Gitiles
Code Review Sign In
gerrit.opencord.org / pyvoltha / 6eb68176020707ed932b3de239f285541c1e17d1 / . / test / unit / extensions / omci / test_omci_cc.py
blob: ced24b0c879454513745e60f9908b697954ba454 [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#
16import binascii
Chip Bolingce2daf62019-02-12 13:53:39 -0600[diff] [blame]17from pyvoltha.adapters.common.frameio.frameio import hexify
Chip Boling67b674a2019-02-08 11:42:18 -0600[diff] [blame]18from twisted.python.failure import Failure
19from unittest import TestCase, main, skip
20from mock.mock_adapter_agent import MockAdapterAgent
21from mock.mock_onu_handler import MockOnuHandler
22from mock.mock_olt_handler import MockOltHandler
23from mock.mock_onu import MockOnu
24from pyvoltha.adapters.extensions.omci.omci_defs import *
25from pyvoltha.adapters.extensions.omci.omci_frame import *
26from pyvoltha.adapters.extensions.omci.omci_entities import *
27from pyvoltha.adapters.extensions.omci.omci_me import ExtendedVlanTaggingOperationConfigurationDataFrame
28from pyvoltha.adapters.extensions.omci.omci_cc import OMCI_CC, UNKNOWN_CLASS_ATTRIBUTE_KEY,\
29 MAX_OMCI_REQUEST_AGE
30
31DEFAULT_OLT_DEVICE_ID = 'default_olt_mock'
32DEFAULT_ONU_DEVICE_ID = 'default_onu_mock'
33DEFAULT_PON_ID = 0
34DEFAULT_ONU_ID = 0
35DEFAULT_ONU_SN = 'TEST00000001'
36
37OP = EntityOperations
38RC = ReasonCodes
39
40successful = False
41error_reason = None
42
43
44def chunk(indexable, chunk_size):
45 for i in range(0, len(indexable), chunk_size):
46 yield indexable[i:i + chunk_size]
47
48
49def hex2raw(hex_string):
50 return ''.join(chr(int(byte, 16)) for byte in chunk(hex_string, 2))
51
52
53class TestOmciCc(TestCase):
54 """
55 Test the Open OMCI Communication channels
56
57 Note also added some testing of MockOnu behaviour since its behaviour during more
58 complicated unit/integration tests may be performed in the future.
59 """
60 def setUp(self, let_msg_timeout=False):
61 self.adapter_agent = MockAdapterAgent()
62
63 def tearDown(self):
64 if self.adapter_agent is not None:
65 self.adapter_agent.tearDown()
66
67 def setup_mock_olt(self, device_id=DEFAULT_OLT_DEVICE_ID):
68 handler = MockOltHandler(self.adapter_agent, device_id)
69 self.adapter_agent.add_device(handler.device)
70 return handler
71
72 def setup_mock_onu(self, parent_id=DEFAULT_OLT_DEVICE_ID,
73 device_id=DEFAULT_ONU_DEVICE_ID,
74 pon_id=DEFAULT_PON_ID,
75 onu_id=DEFAULT_ONU_ID,
76 serial_no=DEFAULT_ONU_SN):
77 handler = MockOnuHandler(self.adapter_agent, parent_id, device_id, pon_id, onu_id)
78 handler.serial_number = serial_no
79 onu = MockOnu(serial_no, self.adapter_agent, handler.device_id) \
80 if serial_no is not None else None
81 handler.onu_mock = onu
82 return handler
83
84 def setup_one_of_each(self, timeout_messages=False):
85 # Most tests will use at lease one or more OLT and ONU
86 self.olt_handler = self.setup_mock_olt()
87 self.onu_handler = self.setup_mock_onu(parent_id=self.olt_handler.device_id)
88 self.onu_device = self.onu_handler.onu_mock
89 self.adapter_agent.timeout_the_message = timeout_messages
90
91 self.adapter_agent.add_child_device(self.olt_handler.device,
92 self.onu_handler.device)
93
94 def _is_omci_frame(self, results, omci_msg_type):
95 assert isinstance(results, OmciFrame), 'Not OMCI Frame'
96 assert 'omci_message' in results.fields, 'Not OMCI Frame'
97 if omci_msg_type is not None:
98 assert isinstance(results.fields['omci_message'], omci_msg_type)
99 return results
100
101 def _check_status(self, results, value):
102 if value is not None: assert results is not None, 'unexpected emtpy message'
103 status = results.fields['omci_message'].fields['success_code']
104 assert status == value,\
105 'Unexpected Status Code. Got {}, Expected: {}'.format(status, value)
106 return results
107
108 def _check_mib_sync(self, results, value):
109 assert self.onu_device.mib_data_sync == value, \
110 'Unexpected MIB DATA Sync value. Got {}, Expected: {}'.format(
111 self.onu_device.mib_data_sync, value)
112 return results
113
114 def _check_stats(self, results, _, stat, expected):
115 snapshot = self._snapshot_stats()
116 assert snapshot[stat] == expected, \
117 'Invalid statistic "{}". Got {}, Expected: {}'.format(stat,
118 snapshot[stat],
119 expected)
120 return results
121
122 def _check_value_equal(self, results, name, value, expected):
123 assert value == expected, \
124 'Value "{}" not equal. Got {}, Expected: {}'.format(name, value,
125 expected)
126 return results
127
128 def _default_errback(self, failure):
129 from twisted.internet.defer import TimeoutError
130 assert isinstance(failure.type, type(TimeoutError))
131
132 def _snapshot_stats(self):
133 omci_cc = self.onu_handler.omci_cc
134 return {
135 'tx_frames': omci_cc.tx_frames,
136 'rx_frames': omci_cc.rx_frames,
137 'rx_unknown_tid': omci_cc.rx_unknown_tid,
138 'rx_onu_frames': omci_cc.rx_onu_frames,
139 'rx_onu_discards': omci_cc.rx_onu_discards,
140 'rx_timeouts': omci_cc.rx_timeouts,
141 'rx_unknown_me': omci_cc.rx_unknown_me,
142 'rx_late': omci_cc.rx_late,
143 'tx_errors': omci_cc.tx_errors,
144 'consecutive_errors': omci_cc.consecutive_errors,
145 'reply_min': omci_cc.reply_min,
146 'reply_max': omci_cc.reply_max,
147 'reply_average': omci_cc.reply_average,
148 'hp_tx_queue_len': omci_cc.hp_tx_queue_len,
149 'lp_tx_queue_len': omci_cc.lp_tx_queue_len,
150 'max_hp_tx_queue': omci_cc.max_hp_tx_queue,
151 'max_lp_tx_queue': omci_cc._max_lp_tx_queue,
152 }
153
154 def test_default_init(self):
155 self.setup_one_of_each()
156 # Test default construction of OMCI_CC as well as
157 # various other parameter settings
158 omci_cc = self.onu_handler.omci_cc
159
160 # No device directly associated
161 self.assertIsNotNone(omci_cc._adapter_agent)
162 self.assertIsNone(omci_cc._proxy_address)
163
164 # No outstanding requests
165 self.assertEqual(len(omci_cc._pending[OMCI_CC.LOW_PRIORITY]), 0)
166 self.assertEqual(len(omci_cc._pending[OMCI_CC.HIGH_PRIORITY]), 0)
167
168 # No active requests
169 self.assertIsNone(omci_cc._tx_request[OMCI_CC.LOW_PRIORITY])
170 self.assertIsNone(omci_cc._tx_request[OMCI_CC.HIGH_PRIORITY])
171
172 # Flags/properties
173 self.assertFalse(omci_cc.enabled)
174
175 # Statistics
176 self.assertEqual(omci_cc.tx_frames, 0)
177 self.assertEqual(omci_cc.rx_frames, 0)
178 self.assertEqual(omci_cc.rx_unknown_tid, 0)
179 self.assertEqual(omci_cc.rx_onu_frames, 0)
180 self.assertEqual(omci_cc.rx_onu_discards, 0)
181 self.assertEqual(omci_cc.rx_unknown_me, 0)
182 self.assertEqual(omci_cc.rx_timeouts, 0)
183 self.assertEqual(omci_cc.rx_late, 0)
184 self.assertEqual(omci_cc.tx_errors, 0)
185 self.assertEqual(omci_cc.consecutive_errors, 0)
186 self.assertNotEquals(omci_cc.reply_min, 0.0)
187 self.assertEqual(omci_cc.reply_max, 0.0)
188 self.assertEqual(omci_cc.reply_average, 0.0)
189 self.assertEqual(omci_cc.lp_tx_queue_len, 0.0)
190 self.assertEqual(omci_cc.max_hp_tx_queue, 0.0)
191 self.assertEqual(omci_cc._max_hp_tx_queue, 0.0)
192 self.assertEqual(omci_cc._max_lp_tx_queue, 0.0)
193
194 def test_enable_disable(self):
195 self.setup_one_of_each()
196
197 # Test enable property
198 omci_cc = self.onu_handler.omci_cc
199
200 # Initially disabled
201 self.assertFalse(omci_cc.enabled)
202 omci_cc.enabled = False
203 self.assertFalse(omci_cc.enabled)
204
205 omci_cc.enabled = True
206 self.assertTrue(omci_cc.enabled)
207 self.assertIsNotNone(omci_cc._proxy_address)
208 self.assertEqual(len(omci_cc._pending[OMCI_CC.LOW_PRIORITY]), 0)
209 self.assertEqual(len(omci_cc._pending[OMCI_CC.HIGH_PRIORITY]), 0)
210
211 omci_cc.enabled = True # Should be a NOP
212 self.assertTrue(omci_cc.enabled)
213 self.assertIsNotNone(omci_cc._proxy_address)
214 self.assertEqual(len(omci_cc._pending[OMCI_CC.LOW_PRIORITY]), 0)
215 self.assertEqual(len(omci_cc._pending[OMCI_CC.HIGH_PRIORITY]), 0)
216
217 omci_cc.enabled = False
218 self.assertFalse(omci_cc.enabled)
219 self.assertIsNone(omci_cc._proxy_address)
220
221 def test_rx_discard_if_disabled(self):
222 # ME without a known decoder
223 self.setup_one_of_each()
224
225 omci_cc = self.onu_handler.omci_cc
226 omci_cc.enabled = False
227 snapshot = self._snapshot_stats()
228
229 msg = '00fc2e0a00020000ff780000e00000010000000c' \
230 '0000000000000000000000000000000000000000' \
231 '00000028105a86ef'
232
233 omci_cc.receive_message(hex2raw(msg))
234
235 # Note: No counter increments
236 self.assertEqual(omci_cc.rx_frames, snapshot['rx_frames'])
237 self.assertEqual(omci_cc.rx_unknown_me, snapshot['rx_unknown_me'])
238 self.assertEqual(omci_cc.rx_unknown_tid, snapshot['rx_unknown_tid'])
239 self.assertEqual(omci_cc.rx_onu_frames, snapshot['rx_onu_frames'])
240
241 def test_message_send_get(self):
242 # Various tests of sending an OMCI message and it either
243 # getting a response or send catching some errors of
244 # importance
245 self.setup_one_of_each()
246
247 omci_cc = self.onu_handler.omci_cc
248 omci_cc.enabled = True
249 snapshot = self._snapshot_stats()
250 mib_data_sync = self.onu_device.mib_data_sync
251
252 # GET
253 # d = omci_cc.send() # TODO: Implement
254 #
255 # d.addCallbacks(self._is_omci_frame, self._default_errback)
256 # d.addCallback(self._check_status, RC.Success.value)
257 # d.addCallback(self._check_mib_sync, mib_data_sync)
258 #
259 # d.addCallback(self._check_stats, snapshot, 'tx_frames', snapshot['tx_frames'] + 1)
260 # d.addCallback(self._check_stats, snapshot, 'rx_frames', snapshot['rx_frames'] + 1)
261 # d.addCallback(self._check_stats, snapshot, 'rx_unknown_tid', snapshot['rx_unknown_tid'])
262 # d.addCallback(self._check_stats, snapshot, 'rx_onu_frames', snapshot['rx_onu_frames'])
263 # d.addCallback(self._check_stats, snapshot, 'rx_onu_discards', snapshot['rx_onu_discards'])
264 # d.addCallback(self._check_stats, snapshot, 'rx_unknown_me', snapshot['rx_unknown_me'])
265 # d.addCallback(self._check_stats, snapshot, 'rx_timeouts', snapshot['rx_timeouts'])
266 # d.addCallback(self._check_stats, snapshot, 'tx_errors', snapshot['tx_errors'])
267 # d.addCallback(self._check_value_equal, 'consecutive_errors', 0, omci_cc.consecutive_errors)
268
269 # return d
270
271 def test_message_send_set(self):
272 # Various tests of sending an OMCI message and it either
273 # getting a response or send catching some errors of
274 # importance
275 self.setup_one_of_each()
276
277 omci_cc = self.onu_handler.omci_cc
278 omci_cc.enabled = True
279 snapshot = self._snapshot_stats()
280 mib_data_sync = self.onu_device.mib_data_sync
281
282 # SET
283 # d = omci_cc.send() # TODO: Implement
284 #
285 # d.addCallbacks(self._is_omci_frame, self._default_errback)
286 # d.addCallback(self._check_status, RC.Success.value)
287 # d.addCallback(self._check_mib_sync, mib_data_sync + 1 if mib_data_sync < 255 else 1)
288 #
289 # d.addCallback(self._check_stats, snapshot, 'tx_frames', snapshot['tx_frames'] + 1)
290 # d.addCallback(self._check_stats, snapshot, 'rx_frames', snapshot['rx_frames'] + 1)
291 # d.addCallback(self._check_stats, snapshot, 'rx_unknown_tid', snapshot['rx_unknown_tid'])
292 # d.addCallback(self._check_stats, snapshot, 'rx_onu_frames', snapshot['rx_onu_frames'])
293 # d.addCallback(self._check_stats, snapshot, 'rx_onu_discards', snapshot['rx_onu_discards'])
294 # d.addCallback(self._check_stats, snapshot, 'rx_unknown_me', snapshot['rx_unknown_me'])
295 # d.addCallback(self._check_stats, snapshot, 'rx_timeouts', snapshot['rx_timeouts'])
296 # d.addCallback(self._check_stats, snapshot, 'tx_errors', snapshot['tx_errors'])
297 # d.addCallback(self._check_value_equal, 'consecutive_errors', 0, omci_cc.consecutive_errors)
298
299 # return d
300 #
301 # # Also test mib_data_sync rollover. 255 -> 1 (zero reserved)
302 #
303 # self.onu_device.mib_data_sync = 255
304 # # SET
305 # self.assertTrue(True) # TODO: Implement (copy previous one here)
306 # self.assertEqual(1, self.onu_device.mib_data_sync)
307
308 def test_message_send_create(self):
309 # Various tests of sending an OMCI message and it either
310 # getting a response or send catching some errors of
311 # importance
312 self.setup_one_of_each()
313
314 omci_cc = self.onu_handler.omci_cc
315 omci_cc.enabled = True
316 snapshot = self._snapshot_stats()
317 mib_data_sync = self.onu_device.mib_data_sync
318
319 # Create
320 # d = omci_cc.send() # TODO: Implement
321 #
322 # d.addCallbacks(self._is_omci_frame, self._default_errback)
323 # d.addCallback(self._check_status, RC.Success.value)
324 # d.addCallback(self._check_mib_sync, mib_data_sync + 1 if mib_data_sync < 255 else 1)
325 #
326 # d.addCallback(self._check_stats, snapshot, 'tx_frames', snapshot['tx_frames'] + 1)
327 # d.addCallback(self._check_stats, snapshot, 'rx_frames', snapshot['rx_frames'] + 1)
328 # d.addCallback(self._check_stats, snapshot, 'rx_unknown_tid', snapshot['rx_unknown_tid'])
329 # d.addCallback(self._check_stats, snapshot, 'rx_onu_frames', snapshot['rx_onu_frames'])
330 # d.addCallback(self._check_stats, snapshot, 'rx_onu_discards', snapshot['rx_onu_discards'])
331 # d.addCallback(self._check_stats, snapshot, 'rx_unknown_me', snapshot['rx_unknown_me'])
332 # d.addCallback(self._check_stats, snapshot, 'rx_timeouts', snapshot['rx_timeouts'])
333 # d.addCallback(self._check_stats, snapshot, 'tx_errors', snapshot['tx_errors'])
334 # d.addCallback(self._check_value_equal, 'consecutive_errors', 0, omci_cc.consecutive_errors)
335
336 # return d
337
338 def test_message_send_delete(self):
339 # Various tests of sending an OMCI message and it either
340 # getting a response or send catching some errors of
341 # importance
342 self.setup_one_of_each()
343
344 omci_cc = self.onu_handler.omci_cc
345 omci_cc.enabled = True
346 snapshot = self._snapshot_stats()
347 mib_data_sync = self.onu_device.mib_data_sync
348
349 # Delete
350 # d = omci_cc.send() # TODO: Implement
351 #
352 # d.addCallbacks(self._is_omci_frame, self._default_errback)
353 # d.addCallback(self._check_status, RC.Success.value)
354 # d.addCallback(self._check_mib_sync, mib_data_sync + 1 if mib_data_sync < 255 else 1)
355 #
356 # d.addCallback(self._check_stats, snapshot, 'tx_frames', snapshot['tx_frames'] + 1)
357 # d.addCallback(self._check_stats, snapshot, 'rx_frames', snapshot['rx_frames'] + 1)
358 # d.addCallback(self._check_stats, snapshot, 'rx_unknown_tid', snapshot['rx_unknown_tid'])
359 # d.addCallback(self._check_stats, snapshot, 'rx_onu_frames', snapshot['rx_onu_frames'])
360 # d.addCallback(self._check_stats, snapshot, 'rx_onu_discards', snapshot['rx_onu_discards'])
361 # d.addCallback(self._check_stats, snapshot, 'rx_unknown_me', snapshot['rx_unknown_me'])
362 # d.addCallback(self._check_stats, snapshot, 'rx_timeouts', snapshot['rx_timeouts'])
363 # d.addCallback(self._check_stats, snapshot, 'tx_errors', snapshot['tx_errors'])
364 # d.addCallback(self._check_value_equal, 'consecutive_errors', 0, omci_cc.consecutive_errors)
365
366 # return d
367
368 def test_message_send_mib_reset(self):
369 self.setup_one_of_each()
370
371 omci_cc = self.onu_handler.omci_cc
372 omci_cc.enabled = True
373 self.onu_device.mib_data_sync = 10
374 snapshot = self._snapshot_stats()
375
376 # Successful MIB Reset
377 d = omci_cc.send_mib_reset(timeout=1.0)
378
379 d.addCallbacks(self._is_omci_frame, self._default_errback)
380 d.addCallback(self._check_status, RC.Success)
381 d.addCallback(self._check_mib_sync, 0)
382
383 d.addCallback(self._check_stats, snapshot, 'tx_frames', snapshot['tx_frames'] + 1)
384 d.addCallback(self._check_stats, snapshot, 'rx_frames', snapshot['rx_frames'] + 1)
385 d.addCallback(self._check_stats, snapshot, 'rx_unknown_tid', snapshot['rx_unknown_tid'])
386 d.addCallback(self._check_stats, snapshot, 'rx_onu_frames', snapshot['rx_onu_frames'])
387 d.addCallback(self._check_stats, snapshot, 'rx_onu_discards', snapshot['rx_onu_discards'])
388 d.addCallback(self._check_stats, snapshot, 'rx_unknown_me', snapshot['rx_unknown_me'])
389 d.addCallback(self._check_stats, snapshot, 'rx_timeouts', snapshot['rx_timeouts'])
390 d.addCallback(self._check_stats, snapshot, 'rx_late', snapshot['rx_late'])
391 d.addCallback(self._check_stats, snapshot, 'tx_errors', snapshot['tx_errors'])
392 d.addCallback(self._check_value_equal, 'consecutive_errors', 0, omci_cc.consecutive_errors)
393 return d
394
395 def test_message_send_mib_upload(self):
396 self.setup_one_of_each()
397
398 omci_cc = self.onu_handler.omci_cc
399 omci_cc.enabled = True
400 snapshot = self._snapshot_stats()
401 mib_data_sync = self.onu_device.mib_data_sync
402
403 # MIB Upload
404 d = omci_cc.send_mib_upload(timeout=1.0)
405
406 d.addCallbacks(self._is_omci_frame, self._default_errback)
407 d.addCallback(self._check_status, RC.Success)
408 d.addCallback(self._check_mib_sync, mib_data_sync)
409
410 # TODO: MIB Upload Results specific tests here
411
412 d.addCallback(self._check_stats, snapshot, 'tx_frames', snapshot['tx_frames'] + 1)
413 d.addCallback(self._check_stats, snapshot, 'rx_frames', snapshot['rx_frames'] + 1)
414 d.addCallback(self._check_stats, snapshot, 'rx_unknown_tid', snapshot['rx_unknown_tid'])
415 d.addCallback(self._check_stats, snapshot, 'rx_onu_frames', snapshot['rx_onu_frames'])
416 d.addCallback(self._check_stats, snapshot, 'rx_onu_discards', snapshot['rx_onu_discards'])
417 d.addCallback(self._check_stats, snapshot, 'rx_unknown_me', snapshot['rx_unknown_me'])
418 d.addCallback(self._check_stats, snapshot, 'rx_timeouts', snapshot['rx_timeouts'])
419 d.addCallback(self._check_stats, snapshot, 'rx_late', snapshot['rx_late'])
420 d.addCallback(self._check_stats, snapshot, 'tx_errors', snapshot['tx_errors'])
421 d.addCallback(self._check_value_equal, 'consecutive_errors', 0, omci_cc.consecutive_errors)
422 return d
423
424 def test_message_send_mib_upload_next(self):
425 self.setup_one_of_each()
426
427 omci_cc = self.onu_handler.omci_cc
428 omci_cc.enabled = True
429 snapshot = self._snapshot_stats()
430 mib_data_sync = self.onu_device.mib_data_sync
431
432 # # MIB Upload Next
433 # d = omci_cc.send_mib_upload_next(0, timeout=1.0)
434 #
435 # d.addCallbacks(self._is_omci_frame, self._default_errback)
436 # d.addCallback(self._check_status, RC.Success)
437 # d.addCallback(self._check_mib_sync, mib_data_sync)
438 #
439 # # TODO: MIB Upload Next Results specific tests here
440 #
441 # d.addCallback(self._check_stats, snapshot, 'tx_frames', snapshot['tx_frames'] + 1)
442 # d.addCallback(self._check_stats, snapshot, 'rx_frames', snapshot['rx_frames'] + 1)
443 # d.addCallback(self._check_stats, snapshot, 'rx_unknown_tid', snapshot['rx_unknown_tid'])
444 # d.addCallback(self._check_stats, snapshot, 'rx_onu_frames', snapshot['rx_onu_frames'])
445 # d.addCallback(self._check_stats, snapshot, 'rx_onu_discards', snapshot['rx_onu_discards'])
446 # d.addCallback(self._check_stats, snapshot, 'rx_unknown_me', snapshot['rx_unknown_me'])
447 # d.addCallback(self._check_stats, snapshot, 'rx_timeouts', snapshot['rx_timeouts'])
448 # d.addCallback(self._check_stats, snapshot, 'tx_errors', snapshot['tx_errors'])
449 # d.addCallback(self._check_value_equal, 'consecutive_errors', 0, omci_cc.consecutive_errors)
450 # return d
451
452 def test_message_send_no_timeout(self):
453 self.setup_one_of_each()
454
455 omci_cc = self.onu_handler.omci_cc
456 omci_cc.enabled = True
457 self.onu_device.mib_data_sync = 10
458 snapshot = self._snapshot_stats()
459
460 d = omci_cc.send_mib_reset(timeout=0)
461 d.addCallback(self._check_stats, snapshot, 'tx_frames', snapshot['tx_frames'] + 1)
462 d.addCallback(self._check_stats, snapshot, 'rx_frames', snapshot['rx_frames'])
463 d.addCallback(self._check_stats, snapshot, 'tx_errors', snapshot['tx_errors'])
464 return d
465
466 def test_message_send_bad_timeout(self):
467 self.setup_one_of_each()
468
469 omci_cc = self.onu_handler.omci_cc
470 omci_cc.enabled = True
471 self.onu_device.mib_data_sync = 10
472 snapshot = self._snapshot_stats()
473
474 d = omci_cc.send_mib_reset(timeout=MAX_OMCI_REQUEST_AGE + 1)
475 d.addCallback(self._check_stats, snapshot, 'tx_frames', snapshot['tx_frames'])
476 d.addCallback(self._check_stats, snapshot, 'rx_frames', snapshot['rx_frames'])
477 d.addCallback(self._check_stats, snapshot, 'tx_errors', snapshot['tx_errors'] + 1)
478 return d
479
480 def test_message_send_not_a_frame(self):
481 self.setup_one_of_each()
482
483 omci_cc = self.onu_handler.omci_cc
484 omci_cc.enabled = True
485 self.onu_device.mib_data_sync = 10
486 snapshot = self._snapshot_stats()
487
488 d = omci_cc.send('hello world', timeout=1)
489 d.addCallback(self._check_stats, snapshot, 'tx_frames', snapshot['tx_frames'])
490 d.addCallback(self._check_stats, snapshot, 'rx_frames', snapshot['rx_frames'])
491 d.addCallback(self._check_stats, snapshot, 'tx_errors', snapshot['tx_errors'] + 1)
492 return d
493
494 def test_message_send_reboot(self):
495 self.setup_one_of_each()
496
497 omci_cc = self.onu_handler.omci_cc
498 omci_cc.enabled = True
499 snapshot = self._snapshot_stats()
500
501 # ONU Reboot
502 d = omci_cc.send_reboot(timeout=1.0)
503
504 d.addCallbacks(self._is_omci_frame, self._default_errback)
505 d.addCallback(self._check_status, RC.Success)
506
507 d.addCallback(self._check_stats, snapshot, 'tx_frames', snapshot['tx_frames'] + 1)
508 d.addCallback(self._check_stats, snapshot, 'rx_frames', snapshot['rx_frames'] + 1)
509 d.addCallback(self._check_stats, snapshot, 'rx_unknown_tid', snapshot['rx_unknown_tid'])
510 d.addCallback(self._check_stats, snapshot, 'rx_onu_frames', snapshot['rx_onu_frames'])
511 d.addCallback(self._check_stats, snapshot, 'rx_onu_discards', snapshot['rx_onu_discards'])
512 d.addCallback(self._check_stats, snapshot, 'rx_unknown_me', snapshot['rx_unknown_me'])
513 d.addCallback(self._check_stats, snapshot, 'rx_timeouts', snapshot['rx_timeouts'])
514 d.addCallback(self._check_stats, snapshot, 'rx_late', snapshot['rx_late'])
515 d.addCallback(self._check_stats, snapshot, 'tx_errors', snapshot['tx_errors'])
516 d.addCallback(self._check_value_equal, 'consecutive_errors', 0, omci_cc.consecutive_errors)
517 return d
518
519 def test_message_send_with_omci_disabled(self):
520 self.setup_one_of_each()
521
522 omci_cc = self.onu_handler.omci_cc
523 self.assertFalse(omci_cc.enabled)
524
525 # Successful MIB Reset
526 d = omci_cc.send_mib_reset(timeout=1.0)
527
528 def success_is_bad(_results):
529 assert False, 'This test should throw a failure/error'
530
531 def fail_fast(_failure):
532 pass
533 return None
534
535 d.addCallbacks(success_is_bad, fail_fast)
536 return d
537
538 def test_message_send_get_with_latency(self):
539 # Various tests of sending an OMCI message and it either
540 # getting a response or send catching some errors of
541 # importance
542 self.setup_one_of_each()
543 self.olt_handler.latency = 0.500 # 1/2 second
544
545 omci_cc = self.onu_handler.omci_cc
546 omci_cc.enabled = True
547
548 # Successful MIB Reset
549 d = omci_cc.send_mib_reset(timeout=1.0)
550
551 d.addCallbacks(self._is_omci_frame, self._default_errback)
552 d.addCallback(self._check_status, RC.Success)
553
554 def check_latency_values(_):
555 self.assertGreaterEqual(omci_cc.reply_min, self.olt_handler.latency)
556 self.assertGreaterEqual(omci_cc.reply_max, self.olt_handler.latency)
557 self.assertGreaterEqual(omci_cc.reply_average, self.olt_handler.latency)
558
559 d.addCallback(check_latency_values)
560 return d
561
562 def test_message_failures(self):
563 # Various tests of sending an OMCI message and it fails
564 self.setup_one_of_each()
565
566 omci_cc = self.onu_handler.omci_cc
567 omci_cc.enabled = True
568 snapshot = self._snapshot_stats()
569
570 self.assertEqual(omci_cc.tx_frames, 0)
571 self.assertEqual(omci_cc.rx_frames, 0)
572 self.assertEqual(omci_cc.rx_unknown_tid, 0)
573 self.assertEqual(omci_cc.rx_timeouts, 0)
574 self.assertEqual(omci_cc.rx_late, 0)
575 self.assertEqual(omci_cc.tx_errors, 0)
576
577 # # Class ID not found
578 # d = omci_cc.send_mib_reset(timeout=1.0)
579 # self.assertTrue(True) # TODO: Implement
580 # todo: Test non-zero consecutive errors
581 #
582 # # Instance ID not found
583 # d = omci_cc.send_mib_reset(timeout=1.0)
584 # self.assertTrue(True) # TODO: Implement
585 # todo: Test non-zero consecutive errors
586 #
587 # # PON is disabled
588 # d = omci_cc.send_mib_reset(timeout=1.0)
589 # self.assertTrue(True) # TODO: Implement
590 # todo: Test non-zero consecutive errors
591 #
592 # # ONU is disabled
593 # d = omci_cc.send_mib_reset(timeout=1.0)
594 # self.assertTrue(True) # TODO: Implement
595 # todo: Test non-zero consecutive errors
596 #
597 # # ONU is not activated
598 # d = omci_cc.send_mib_reset(timeout=1.0)
599 # self.assertTrue(True) # TODO: Implement
600 # todo: Test non-zero consecutive errors
601
602 # TODO: make OLT send back an unknown TID (
603
604 # todo: Test non-zero consecutive errors
605 # todo: Send a good frame
606 # todo: Test zero consecutive errors
607 # d.addCallback(self._check_value_equal, 'consecutive_errors', 0, omci_cc.consecutive_errors)
608
609 def test_rx_unknown_me(self):
610 # ME without a known decoder
611 self.setup_one_of_each()
612
613 omci_cc = self.onu_handler.omci_cc
614 omci_cc.enabled = True
615 snapshot = self._snapshot_stats()
616
617 # This is the ID ------+
618 # v
619 msg = '00fc2e0a00020000ff780000e00000010000000c' \
620 '0000000000000000000000000000000000000000' \
621 '00000028'
622
623 omci_cc.receive_message(hex2raw(msg))
624
625 # Note: After successful frame decode, a lookup of the corresponding request by
626 # TID is performed. None should be found, so we should see the Rx Unknown TID
627 # increment.
628 self.assertEqual(omci_cc.rx_frames, snapshot['rx_frames'])
629 self.assertEqual(omci_cc.rx_unknown_me, snapshot['rx_unknown_me'] + 1)
630 self.assertEqual(omci_cc.rx_unknown_tid, snapshot['rx_unknown_tid'] + 1)
631 self.assertEqual(omci_cc.rx_onu_frames, snapshot['rx_onu_frames'])
632 self.assertEqual(omci_cc.consecutive_errors, 0)
633
634 def test_rx_decode_unknown_me(self):
635 # ME without a known decoder
636 self.setup_one_of_each()
637
638 omci_cc = self.onu_handler.omci_cc
639 omci_cc.enabled = True
640 snapshot = self._snapshot_stats()
641
642 # This is a MIB Upload Next Response. Where we would probably first see an
643 # unknown Class ID
644 #
645 # This is the ID ------+
646 # v
647 msg = '00fc2e0a00020000ff780001e000'
648 blob = '00010000000c0000000000000000000000000000000000000000'
649 msg += blob + '00000028'
650
651 # Dig into the internal method so we can get the returned frame
652 frame = omci_cc._decode_unknown_me(hex2raw(msg))
653
654 self.assertEqual(frame.fields['transaction_id'], 0x00fc)
655 self.assertEqual(frame.fields['message_type'], 0x2e)
656
657 omci_fields = frame.fields['omci_message'].fields
658
659 self.assertEqual(omci_fields['entity_class'], 0x0002)
660 self.assertEqual(omci_fields['entity_id'], 0x00)
661 self.assertEqual(omci_fields['object_entity_class'], 0x0ff78)
662 self.assertEqual(omci_fields['object_entity_id'], 0x01)
663 self.assertEqual(omci_fields['object_attributes_mask'], 0xe000)
664
665 data_fields = omci_fields['object_data']
666
667 decoded_blob = data_fields.get(UNKNOWN_CLASS_ATTRIBUTE_KEY)
668 self.assertIsNotNone(decoded_blob)
669 self.assertEqual(decoded_blob, blob)
670
671 def test_flush(self):
672 # Test flush of autonomous ONU queues
673 self.setup_one_of_each()
674
675 omci_cc = self.onu_handler.omci_cc
676 omci_cc.enabled = True
677 snapshot = self._snapshot_stats()
678
679 # TODO: add more
680 self.assertTrue(True) # TODO: Implement
681
682 def test_avc_rx(self):
683 # Test flush of autonomous ONU queues
684 self.setup_one_of_each()
685
686 omci_cc = self.onu_handler.omci_cc
687 omci_cc.enabled = True
688 snapshot = self._snapshot_stats()
689
690 # TODO: add more
691 self.assertTrue(True) # TODO: Implement
692
693 def test_rx_discard_if_disabled(self):
694 # ME without a known decoder
695 self.setup_one_of_each()
696
697 omci_cc = self.onu_handler.omci_cc
698 omci_cc.enabled = False
699 snapshot = self._snapshot_stats()
700
701 msg = '00fc2e0a00020000ff780000e00000010000000c' \
702 '0000000000000000000000000000000000000000' \
703 '00000028105a86ef'
704
705 omci_cc.receive_message(hex2raw(msg))
706
707 # Note: No counter increments
708 self.assertEqual(omci_cc.rx_frames, snapshot['rx_frames'])
709 self.assertEqual(omci_cc.rx_unknown_me, snapshot['rx_unknown_me'])
710 self.assertEqual(omci_cc.rx_unknown_tid, snapshot['rx_unknown_tid'])
711 self.assertEqual(omci_cc.rx_onu_frames, snapshot['rx_onu_frames'])
712
713 def test_omci_alarm_decode(self):
714 """
715 This test covers an issue discovered in Sept 2018 (JIRA-1213). It was
716 an exception during frame decode.
717 """
718 self.setup_one_of_each()
719
720 omci_cc = self.onu_handler.omci_cc
721 omci_cc.enabled = True
722 snapshot = self._snapshot_stats()
723
724 # Frame from the JIRA issue
725 msg = '0000100a000b0102800000000000000000000000' \
726 '0000000000000000000000000000000000000015' \
727 '000000282d3ae0a6'
728
729 _results = omci_cc.receive_message(hex2raw(msg))
730
731 self.assertEqual(omci_cc.rx_frames, snapshot['rx_frames'])
732 self.assertEqual(omci_cc.rx_unknown_me, snapshot['rx_unknown_me'])
733 self.assertEqual(omci_cc.rx_unknown_tid, snapshot['rx_unknown_tid'])
734 self.assertEqual(omci_cc.rx_onu_frames, snapshot['rx_onu_frames'] + 1)
735 self.assertEqual(omci_cc.rx_onu_discards, snapshot['rx_onu_discards'])
736
737 def test_omci_avc_decode(self):
738 self.setup_one_of_each()
739
740 omci_cc = self.onu_handler.omci_cc
741 omci_cc.enabled = True
742 snapshot = self._snapshot_stats()
743
744 # Frame from the JIRA issue
745 msg = '0000110a0007000080004d4c2d33363236000000' \
746 '0000000020202020202020202020202020202020' \
747 '00000028'
748
749 _results = omci_cc.receive_message(hex2raw(msg))
750
751 self.assertEqual(omci_cc.rx_frames, snapshot['rx_frames'])
752 self.assertEqual(omci_cc.rx_unknown_me, snapshot['rx_unknown_me'])
753 self.assertEqual(omci_cc.rx_unknown_tid, snapshot['rx_unknown_tid'])
754 self.assertEqual(omci_cc.rx_onu_frames, snapshot['rx_onu_frames'] + 1)
755 self.assertEqual(omci_cc.rx_onu_discards, snapshot['rx_onu_discards'])
756
757 def test_omci_unknown_onu_decode(self):
758 self.setup_one_of_each()
759
760 omci_cc = self.onu_handler.omci_cc
761 omci_cc.enabled = True
762 snapshot = self._snapshot_stats()
763
764 # Frame from the JIRA issue
765 msg = '0000190a0007000080004d4c2d33363236000000' \
766 '0000000020202020202020202020202020202020' \
767 '00000028'
768
769 _results = omci_cc.receive_message(hex2raw(msg))
770
771 self.assertEqual(omci_cc.rx_frames, snapshot['rx_frames'])
772 self.assertEqual(omci_cc.rx_unknown_me, snapshot['rx_unknown_me'])
773 self.assertEqual(omci_cc.rx_unknown_tid, snapshot['rx_unknown_tid'])
774 self.assertEqual(omci_cc.rx_onu_frames, snapshot['rx_onu_frames'] + 1)
775 self.assertEqual(omci_cc.rx_onu_discards, snapshot['rx_onu_discards'] + 1)
776
777 def test_omci_bad_frame_decode(self):
778 self.setup_one_of_each()
779
780 omci_cc = self.onu_handler.omci_cc
781 omci_cc.enabled = True
782 snapshot = self._snapshot_stats()
783
784 # Frame from the JIRA issue
785 msg = '0020190a0007000080004d4c2d33363236000000' \
786 '0000000000000028'
787
788 _results = omci_cc.receive_message(hex2raw(msg))
789 # NOTE: Currently do not increment any Rx Discard counters, just throw it away
790 self.assertEqual(omci_cc.rx_frames, snapshot['rx_frames'] + 1)
791 self.assertEqual(omci_cc.rx_unknown_me, snapshot['rx_unknown_me'])
792 self.assertEqual(omci_cc.rx_unknown_tid, snapshot['rx_unknown_tid'] + 1)
793 self.assertEqual(omci_cc.rx_onu_frames, snapshot['rx_onu_frames'])
794 self.assertEqual(omci_cc.rx_onu_discards, snapshot['rx_onu_discards'])
795
796 def test_rx_decode_onu_g(self):
797 self.setup_one_of_each()
798
799 omci_cc = self.onu_handler.omci_cc
800 omci_cc.enabled = True
801 snapshot = self._snapshot_stats()
802
803 msg = '001e2e0a0002000001000000e000424657530000' \
804 '0000000000000000000000324246575300107496' \
805 '00000028e7fb4a91'
806
807 omci_cc.receive_message(hex2raw(msg))
808
809 # Note: No counter increments
810 self.assertEqual(omci_cc.rx_frames, snapshot['rx_frames'] + 1)
811 self.assertEqual(omci_cc.rx_unknown_me, snapshot['rx_unknown_me'])
812 self.assertEqual(omci_cc.rx_unknown_tid, snapshot['rx_unknown_tid'] + 1)
813 self.assertEqual(omci_cc.rx_onu_frames, snapshot['rx_onu_frames'])
814
815 def test_rx_decode_extvlantagging(self):
816 self.setup_one_of_each()
817
818 omci_cc = self.onu_handler.omci_cc
819 omci_cc.enabled = True
820 snapshot = self._snapshot_stats()
821
822 msg = '030a290a00ab0201000d00000000001031323334' \
823 '3536373839303132333435363738393031323334' \
824 '000000281166d283'
825
826 omci_cc.receive_message(hex2raw(msg))
827
828 self.assertEqual(omci_cc.rx_frames, snapshot['rx_frames'] + 1)
829 self.assertEqual(omci_cc.rx_unknown_me, snapshot['rx_unknown_me'])
830 self.assertEqual(omci_cc.rx_unknown_tid, snapshot['rx_unknown_tid'] + 1)
831 self.assertEqual(omci_cc.rx_onu_frames, snapshot['rx_onu_frames'])
832
833 def _check_vlan_tag_op(self, results, attr, expected):
834 omci_msg = results.fields['omci_message']
835 data = omci_msg.fields['data']
836 val = data[attr]
837 self.assertEqual(expected, val)
838 return results
839
840 @skip('for unknown omci failure')
841 #@deferred()
842 def test_rx_table_get_extvlantagging(self):
843 self.setup_one_of_each()
844
845 onu = self.onu_handler.onu_mock
846 entity_id = 1
847 vlan_tag_op1 = VlanTaggingOperation(
848 filter_outer_priority=15,
849 filter_outer_vid=4096,
850 filter_outer_tpid_de=2,
851 filter_inner_priority=15,
852 filter_inner_vid=4096,
853 filter_inner_tpid_de=0,
854 filter_ether_type=0,
855 treatment_tags_to_remove=0,
856 treatment_outer_priority=15,
857 treatment_outer_vid=1234,
858 treatment_outer_tpid_de=0,
859 treatment_inner_priority=0,
860 treatment_inner_vid=4091,
861 treatment_inner_tpid_de=4,
862 )
863 vlan_tag_op2 = VlanTaggingOperation(
864 filter_outer_priority=14,
865 filter_outer_vid=1234,
866 filter_outer_tpid_de=5,
867 filter_inner_priority=1,
868 filter_inner_vid=2345,
869 filter_inner_tpid_de=1,
870 filter_ether_type=0,
871 treatment_tags_to_remove=1,
872 treatment_outer_priority=15,
873 treatment_outer_vid=2222,
874 treatment_outer_tpid_de=1,
875 treatment_inner_priority=1,
876 treatment_inner_vid=3333,
877 treatment_inner_tpid_de=5,
878 )
879 vlan_tag_op3 = VlanTaggingOperation(
880 filter_outer_priority=13,
881 filter_outer_vid=55,
882 filter_outer_tpid_de=1,
883 filter_inner_priority=7,
884 filter_inner_vid=4567,
885 filter_inner_tpid_de=1,
886 filter_ether_type=0,
887 treatment_tags_to_remove=1,
888 treatment_outer_priority=2,
889 treatment_outer_vid=1111,
890 treatment_outer_tpid_de=1,
891 treatment_inner_priority=1,
892 treatment_inner_vid=3131,
893 treatment_inner_tpid_de=5,
894 )
895 tbl = [vlan_tag_op1, vlan_tag_op2, vlan_tag_op3]
896 tblstr = str(vlan_tag_op1) + str(vlan_tag_op2) + str(vlan_tag_op3)
897
898 onu._omci_response[OP.Get.value][ExtendedVlanTaggingOperationConfigurationData.class_id] = {
899 entity_id: OmciFrame(transaction_id=0,
900 message_type=OmciGetResponse.message_id,
901 omci_message=OmciGetResponse(
902 entity_class=ExtendedVlanTaggingOperationConfigurationData.class_id,
903 entity_id=1,
904 success_code=RC.Success.value,
905 attributes_mask=ExtendedVlanTaggingOperationConfigurationData.mask_for(
906 'received_frame_vlan_tagging_operation_table'),
907 data={'received_frame_vlan_tagging_operation_table': 16 * len(tbl)}
908 ))
909 }
910
911 rsp1 = binascii.a2b_hex(hexify(tblstr[0:OmciTableField.PDU_SIZE]))
912 rsp2 = binascii.a2b_hex(hexify(tblstr[OmciTableField.PDU_SIZE:]))
913 onu._omci_response[OP.GetNext.value][ExtendedVlanTaggingOperationConfigurationData.class_id] = {
914 entity_id: {0: {'failures':2,
915 'frame':OmciFrame(transaction_id=0,
916 message_type=OmciGetNextResponse.message_id,
917 omci_message=OmciGetNextResponse(
918 entity_class=ExtendedVlanTaggingOperationConfigurationData.class_id,
919 entity_id=1,
920 success_code=RC.Success.value,
921 attributes_mask=ExtendedVlanTaggingOperationConfigurationData.mask_for(
922 'received_frame_vlan_tagging_operation_table'),
923 data={'received_frame_vlan_tagging_operation_table': rsp1
924 }
925 ))},
926 1: OmciFrame(transaction_id=0,
927 message_type=OmciGetNextResponse.message_id,
928 omci_message=OmciGetNextResponse(
929 entity_class=ExtendedVlanTaggingOperationConfigurationData.class_id,
930 entity_id=1,
931 success_code=RC.Success.value,
932 attributes_mask=ExtendedVlanTaggingOperationConfigurationData.mask_for(
933 'received_frame_vlan_tagging_operation_table'),
934 data={'received_frame_vlan_tagging_operation_table': rsp2
935 }
936 ))
937 }
938 }
939
940 omci_cc = self.onu_handler.omci_cc
941 omci_cc.enabled = True
942
943 msg = ExtendedVlanTaggingOperationConfigurationDataFrame(
944 entity_id,
945 attributes={'received_frame_vlan_tagging_operation_table':True}
946 )
947
948 snapshot = self._snapshot_stats()
949
950 frame = msg.get()
951 d = omci_cc.send(frame, timeout=5.0)
952
953 d.addCallbacks(self._is_omci_frame, self._default_errback, [OmciGetResponse])
954 d.addCallback(self._check_status, RC.Success)
955
956 d.addCallback(self._check_stats, snapshot, 'tx_frames', snapshot['tx_frames'] + 5)
957 d.addCallback(self._check_stats, snapshot, 'rx_frames', snapshot['rx_frames'] + 3)
958 d.addCallback(self._check_stats, snapshot, 'rx_unknown_tid', snapshot['rx_unknown_tid'])
959 d.addCallback(self._check_stats, snapshot, 'rx_onu_frames', snapshot['rx_onu_frames'])
960 d.addCallback(self._check_stats, snapshot, 'rx_onu_discards', snapshot['rx_onu_discards'])
961 d.addCallback(self._check_stats, snapshot, 'rx_unknown_me', snapshot['rx_unknown_me'])
962 d.addCallback(self._check_stats, snapshot, 'rx_timeouts', snapshot['rx_timeouts'] + 2)
963 d.addCallback(self._check_stats, snapshot, 'rx_late', snapshot['rx_late'])
964 d.addCallback(self._check_stats, snapshot, 'tx_errors', snapshot['tx_errors'])
965 d.addCallback(self._check_stats, snapshot, 'consecutive_errors', 0)
966 d.addCallback(self._check_vlan_tag_op, 'received_frame_vlan_tagging_operation_table', tbl)
967
968 return d
969
970 ##################################################################
971 # Start of tests specific to new stop_and_wait changes
972 #
973 def test_message_send_low_priority(self):
974 # self.setup_one_of_each(timeout_messages=True)
975 self.setup_one_of_each()
976
977 omci_cc = self.onu_handler.omci_cc
978 omci_cc.enabled = True
979 snapshot = self._snapshot_stats()
980
981 # MIB Upload
982 d = omci_cc.send_mib_upload(timeout=1.0, high_priority=False)
983 d.addCallback(self._check_stats, snapshot, 'lp_tx_queue_len', snapshot['lp_tx_queue_len'])
984 d.addCallback(self._check_stats, snapshot, 'hp_tx_queue_len', snapshot['hp_tx_queue_len'])
985 d.addCallback(self._check_stats, snapshot, 'max_lp_tx_queue', snapshot['max_lp_tx_queue'] + 1)
986 d.addCallback(self._check_stats, snapshot, 'max_hp_tx_queue', snapshot['max_hp_tx_queue'])
987
988 # Flush to get ready for next test (one frame queued)
989 omci_cc.flush()
990 self.assertEqual(omci_cc.lp_tx_queue_len, 0)
991 self.assertEqual(omci_cc.hp_tx_queue_len, 0)
992
993 self.adapter_agent.timeout_the_message = True
994 omci_cc.send_mib_upload(timeout=1.0, high_priority=False)
995 omci_cc.send_mib_upload(timeout=1.0, high_priority=False)
996
997 self.assertEqual(omci_cc.lp_tx_queue_len, 1)
998 self.assertEqual(omci_cc.hp_tx_queue_len, 0)
999 self.assertEqual(omci_cc.max_lp_tx_queue, 1)
1000 self.assertEqual(omci_cc.max_hp_tx_queue, 0)
1001
1002 # Flush to get ready for next test (two queued and new max)
1003 omci_cc.flush()
1004 omci_cc.send_mib_upload(timeout=1.0, high_priority=False)
1005 omci_cc.send_mib_upload(timeout=1.0, high_priority=False)
1006 omci_cc.send_mib_upload(timeout=1.0, high_priority=False)
1007
1008 self.assertEqual(omci_cc.lp_tx_queue_len, 2)
1009 self.assertEqual(omci_cc.hp_tx_queue_len, 0)
1010 self.assertEqual(omci_cc.max_lp_tx_queue, 2)
1011 self.assertEqual(omci_cc.max_hp_tx_queue, 0)
1012
1013 def test_message_send_high_priority(self):
1014 # self.setup_one_of_each(timeout_messages=True)
1015 self.setup_one_of_each()
1016
1017 omci_cc = self.onu_handler.omci_cc
1018 omci_cc.enabled = True
1019 snapshot = self._snapshot_stats()
1020
1021 # MIB Upload
1022 d = omci_cc.send_mib_upload(high_priority=True)
1023 d.addCallback(self._check_stats, snapshot, 'lp_tx_queue_len', snapshot['lp_tx_queue_len'])
1024 d.addCallback(self._check_stats, snapshot, 'hp_tx_queue_len', snapshot['hp_tx_queue_len'])
1025 d.addCallback(self._check_stats, snapshot, 'max_lp_tx_queue', snapshot['max_lp_tx_queue'])
1026 d.addCallback(self._check_stats, snapshot, 'max_hp_tx_queue', snapshot['max_hp_tx_queue'] + 1)
1027
1028 # Flush to get ready for next test (one frame queued)
1029 omci_cc.flush()
1030 self.assertEqual(omci_cc.lp_tx_queue_len, 0)
1031 self.assertEqual(omci_cc.hp_tx_queue_len, 0)
1032
1033 self.adapter_agent.timeout_the_message = True
1034 omci_cc.send_mib_upload(high_priority=True)
1035 omci_cc.send_mib_upload(high_priority=True)
1036
1037 self.assertEqual(omci_cc.lp_tx_queue_len, 0)
1038 self.assertEqual(omci_cc.hp_tx_queue_len, 1)
1039 self.assertEqual(omci_cc.max_lp_tx_queue, 0)
1040 self.assertEqual(omci_cc.max_hp_tx_queue, 1)
1041
1042 # Flush to get ready for next test (two queued and new max)
1043 omci_cc.flush()
1044 omci_cc.send_mib_upload(high_priority=True)
1045 omci_cc.send_mib_upload(high_priority=True)
1046 omci_cc.send_mib_upload(high_priority=True)
1047
1048 self.assertEqual(omci_cc.lp_tx_queue_len, 0)
1049 self.assertEqual(omci_cc.hp_tx_queue_len, 2)
1050 self.assertEqual(omci_cc.max_lp_tx_queue, 0)
1051 self.assertEqual(omci_cc.max_hp_tx_queue, 2)
1052
1053 def test_message_send_and_cancel(self):
1054 global error_reason
1055 global successful
1056 # Do not send messages to adapter_agent
1057 self.setup_one_of_each(timeout_messages=True)
1058
1059 omci_cc = self.onu_handler.omci_cc
1060 omci_cc.enabled = True
1061
1062 def success(_results):
1063 global successful
1064 successful = True
1065
1066 def failure(reason):
1067 global error_reason
1068 error_reason = reason
1069
1070 def notCalled(reason):
1071 assert isinstance(reason, Failure), 'Should not be called with success'
1072
1073 # Cancel one that is actively being sent
1074 d = omci_cc.send_mib_upload(high_priority=False)
1075 d.addCallbacks(success, failure)
1076 self.assertEqual(omci_cc.lp_tx_queue_len, 0)
1077 self.assertEqual(omci_cc.hp_tx_queue_len, 0)
1078
1079 d.cancel()
1080 self.assertIsInstance(error_reason, Failure)
1081 self.assertFalse(successful)
1082 self.assertTrue(d.called)
1083
1084 self.assertEqual(omci_cc.max_lp_tx_queue, 1)
1085 self.assertEqual(omci_cc.max_hp_tx_queue, 0)
1086
1087 # Flush to get ready for next test (one running, one queued, cancel the
1088 # running one, so queued runs)
1089 omci_cc.flush()
1090 self.assertEqual(omci_cc.lp_tx_queue_len, 0)
1091 self.assertEqual(omci_cc.hp_tx_queue_len, 0)
1092
1093 error_reason = None
1094 d1 = omci_cc.send_mib_upload(high_priority=False)
1095 d2 = omci_cc.send_mib_upload(high_priority=False)
1096 d1.addCallbacks(success, failure)
1097 d2.addCallbacks(notCalled, notCalled)
1098 self.assertEqual(omci_cc.lp_tx_queue_len, 1)
1099 self.assertEqual(omci_cc.hp_tx_queue_len, 0)
1100
1101 d1.cancel()
1102 self.assertIsInstance(error_reason, Failure)
1103 self.assertFalse(successful)
1104 self.assertTrue(d1.called)
1105 self.assertFalse(d2.called)
1106
1107 self.assertEqual(omci_cc.max_lp_tx_queue, 1)
1108 self.assertEqual(omci_cc.max_hp_tx_queue, 0)
1109 self.assertEqual(omci_cc.lp_tx_queue_len, 0)
1110 self.assertEqual(omci_cc.hp_tx_queue_len, 0)
1111
1112 # Flush to get ready for next test (one running, one queued, cancel the queued one)
1113
1114 omci_cc.flush()
1115 self.assertEqual(omci_cc.lp_tx_queue_len, 0)
1116 self.assertEqual(omci_cc.hp_tx_queue_len, 0)
1117
1118 error_reason = None
1119 d3 = omci_cc.send_mib_upload(timeout=55, high_priority=False)
1120 d4 = omci_cc.send_mib_upload(timeout=55, high_priority=False)
1121 d5 = omci_cc.send_mib_upload(timeout=55, high_priority=False)
1122 d3.addCallbacks(notCalled, notCalled)
1123 d4.addCallbacks(success, failure)
1124 d5.addCallbacks(notCalled, notCalled)
1125 self.assertEqual(omci_cc.lp_tx_queue_len, 2)
1126 self.assertEqual(omci_cc.hp_tx_queue_len, 0)
1127
1128 d4.cancel()
1129 self.assertIsInstance(error_reason, Failure)
1130 self.assertFalse(successful)
1131 self.assertFalse(d3.called)
1132 self.assertTrue(d4.called)
1133 self.assertFalse(d5.called)
1134
1135 def test_message_send_low_and_high_priority(self):
1136 self.setup_one_of_each(timeout_messages=True)
1137
1138 omci_cc = self.onu_handler.omci_cc
1139 omci_cc.enabled = True
1140
1141 omci_cc.send_mib_reset(high_priority=False)
1142 omci_cc.send_mib_reset(high_priority=True)
1143 self.assertEqual(omci_cc.lp_tx_queue_len, 0)
1144 self.assertEqual(omci_cc.hp_tx_queue_len, 0)
1145
1146 omci_cc.flush()
1147 self.assertEqual(omci_cc.lp_tx_queue_len, 0)
1148 self.assertEqual(omci_cc.hp_tx_queue_len, 0)
1149
1150 omci_cc.send_mib_reset(high_priority=False)
1151 omci_cc.send_mib_reset(high_priority=True)
1152 omci_cc.send_mib_reset(high_priority=False)
1153 omci_cc.send_mib_reset(high_priority=True)
1154 self.assertEqual(omci_cc.lp_tx_queue_len, 1)
1155 self.assertEqual(omci_cc.hp_tx_queue_len, 1)
1156
1157 def test_no_sw_download_and_mib_upload_at_same_time(self):
1158 # Section B.2.3 of ITU G.988-2017 specifies that a MIB
1159 # upload or software download at a given priority level
1160 # is not allowed while a similar action in the other
1161 # priority level is in progress. Relates to possible memory
1162 # consumption/needs on the ONU.
1163 #
1164 # OMCI_CC only checks if the commands are currently in
1165 # progress. ONU should reject messages if the upload/download
1166 # is in progress (but not an active request is in progress).
1167
1168 self.setup_one_of_each(timeout_messages=True)
1169 omci_cc = self.onu_handler.omci_cc
1170 omci_cc.enabled = True
1171
1172 mib_upload_msgs = [omci_cc.send_mib_upload,
1173 # omci_cc.send_mib_upload_next
1174 ]
1175 sw_download_msgs = [omci_cc.send_start_software_download,
1176 # omci_cc.send_download_section,
1177 # omci_cc.send_end_software_download
1178 ]
1179
1180 for upload in mib_upload_msgs:
1181 for download in sw_download_msgs:
1182 self.assertEqual(omci_cc.lp_tx_queue_len, 0)
1183 self.assertEqual(omci_cc.hp_tx_queue_len, 0)
1184
1185 upload(high_priority=False)
1186 download(1, 1, 1, high_priority=True) # Should stall send-next 50mS
1187 self.assertEqual(omci_cc.lp_tx_queue_len, 0)
1188 self.assertEqual(omci_cc.hp_tx_queue_len, 1)
1189
1190 omci_cc.flush()
1191 self.assertEqual(omci_cc.lp_tx_queue_len, 0)
1192 self.assertEqual(omci_cc.hp_tx_queue_len, 0)
1193
1194 upload(high_priority=True)
1195 download(1, 1, 1, high_priority=False) # Should stall send-next 50mS
1196 self.assertEqual(omci_cc.lp_tx_queue_len, 1)
1197 self.assertEqual(omci_cc.hp_tx_queue_len, 0)
1198
1199 omci_cc.flush()
1200 self.assertEqual(omci_cc.lp_tx_queue_len, 0)
1201 self.assertEqual(omci_cc.hp_tx_queue_len, 0)
1202
1203 download(1, 1, 1, high_priority=False)
1204 upload(high_priority=True) # Should stall send-next 50mS
1205 self.assertEqual(omci_cc.lp_tx_queue_len, 0)
1206 self.assertEqual(omci_cc.hp_tx_queue_len, 1)
1207
1208 omci_cc.flush()
1209 self.assertEqual(omci_cc.lp_tx_queue_len, 0)
1210 self.assertEqual(omci_cc.hp_tx_queue_len, 0)
1211
1212 download(1, 1, 1, high_priority=True)
1213 upload(high_priority=False) # Should stall send-next 50mS)
1214 self.assertEqual(omci_cc.lp_tx_queue_len, 1)
1215 self.assertEqual(omci_cc.hp_tx_queue_len, 0)
1216
1217 omci_cc.flush()
1218 self.assertEqual(omci_cc.lp_tx_queue_len, 0)
1219 self.assertEqual(omci_cc.hp_tx_queue_len, 0)
1220
1221 # Some more ideas for tests that we could add
1222 # Send explicit tid that is not valid
1223 # - Look at top of 'Send' method and test all the error conditions could may hit
1224
1225 # Send multiple and have the OLT proxy throw an exception. Should call errback and
1226 # schedule remainder in queue to still tx.
1227
1228 # Send a frame and then inject a response and test the RX logic out, including late
1229 # rx and retries by the OMCI_CC transmitter.
1230
1231
1232if __name__ == '__main__':
1233 main()