Gitiles
Code Review Sign In
gerrit.opencord.org / pyvoltha / 6eb68176020707ed932b3de239f285541c1e17d1 / . / test / unit / extensions / omci / mock / mock_onu.py
blob: e63c5cdd1a567888c8957ccdc7661f7f85418215 [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 pyvoltha.adapters.extensions.omci.omci_frame import OmciFrame
17from pyvoltha.adapters.extensions.omci.omci_defs import *
18from pyvoltha.adapters.extensions.omci.omci_entities import *
19from pyvoltha.adapters.extensions.omci.omci_messages import *
20
21# abbreviations
22OP = EntityOperations
23RC = ReasonCodes
24
25
26class MockOnu(object):
27 """
28 Minimal class that acts line an ONU. The Mock OLT handler will call into this
29 object with OMCI frames that it will respond to appropriately
30 """
31 def __init__(self, serial_number, adapter_agent, handler_id):
32 self.serial_number = serial_number
33 self._adapter_agent = adapter_agent # TODO: Remove any unused attributes
34 self._handler_id = handler_id
35 self.mib_data_sync = 0 # Assume at reboot!
36
37 # NOTE: when creating response frames, use the basic method of constructing
38 # these frames as the encoding created is unit-tested elsewhere
39 self._omci_response = {
40 OP.Get.value: {
41 CircuitPack.class_id: {
42 257: OmciFrame(transaction_id=0, # Will get replaced
43 message_type=OmciGetResponse.message_id,
44 omci_message=OmciGetResponse(
45 entity_class=CircuitPack.class_id,
46 entity_id=0,
47 success_code=RC.Success.value,
48 attributes_mask=CircuitPack.mask_for('number_of_ports'),
49 data=OmciMaskedData('value',
50 entity_class=CircuitPack.class_id,
51 attributes_mask=CircuitPack.mask_for('number_of_ports'))
52 ))
53 },
54 # Additional OMCI GET request responses here if needed
55 },
56 OP.GetNext.value: {},
57 OP.Create.value: {
58 # TODO: Create some OMCI CREATE request responses here.
59
60 # def send_create_gal_ethernet_profile(self,
61 # entity_id,
62 # max_gem_payload_size):
63 # frame = OmciFrame(
64 # transaction_id=self.get_tx_id(),
65 # message_type=OmciCreate.message_id,
66 # omci_message=OmciCreate(
67 # entity_class=GalEthernetProfile.class_id,
68 # entity_id=entity_id,
69 # data=dict(
70 # max_gem_payload_size=max_gem_payload_size
71 # )
72 # )
73 # )
74 # self.send_omci_message(frame)
75 },
76 OP.Set.value: {
77 # TODO: Create some OMCI SET request responses here.
78
79 # def send_set_admin_state(self,
80 # entity_id,
81 # admin_state):
82 # data = dict(
83 # administrative_state=admin_state
84 # )
85 # frame = OmciFrame(
86 # transaction_id=self.get_tx_id(),
87 # message_type=OmciSet.message_id,
88 # omci_message=OmciSet(
89 # entity_class=OntG.class_id,
90 # entity_id=entity_id,
91 # attributes_mask=OntG.mask_for(*data.keys()),
92 # data=data
93 # )
94 # )
95 # self.send_omci_message(frame)
96
97 },
98 OP.Delete.value: {
99 # TODO: Create some OMCI DELETE responses here.
100 },
101 OP.MibReset.value: {
102 OntData.class_id: {
103 0: OmciFrame(transaction_id=0, # Will get replaced
104 message_type=OmciMibResetResponse.message_id,
105 omci_message=OmciMibResetResponse(
106 entity_class=OntData.class_id,
107 entity_id=0,
108 success_code=RC.Success.value
109 ))
110 }
111 },
112 OP.MibUpload.value: {
113 OntData.class_id: {
114 0: OmciFrame(transaction_id=0, # Will get replaced
115 message_type=OmciMibUploadResponse.message_id,
116 omci_message=OmciMibUploadResponse(
117 entity_class=OntData.class_id,
118 entity_id=0,
119 number_of_commands=3 # Should match list size for MibUploadNext below
120 ))
121 }
122 },
123 # OP.MibUploadNext.value: {
124 # OntData.class_id: {
125 # 0: [
126 # OmciFrame(transaction_id=0,
127 # message_type=OmciMibUploadNextResponse.message_id,
128 # omci_message=OmciMibUploadNextResponse(
129 # entity_class=OntData.class_id,
130 # entity_id=0,
131 # object_entity_id=0, # TODO: Pick one
132 # object_attributes_mask=0, # TODO: Pick one
133 # object_data=None # TODO: Pick one
134 # )),
135 # OmciFrame(transaction_id=0,
136 # message_type=OmciMibUploadNextResponse.message_id,
137 # omci_message=OmciMibUploadNextResponse(
138 # entity_class=OntData.class_id,
139 # entity_id=0,
140 # object_entity_id=0, # TODO: Pick one
141 # object_attributes_mask=0, # TODO: Pick one
142 # object_data=None # TODO: Pick one
143 # )),
144 # OmciFrame(transaction_id=0,
145 # message_type=OmciMibUploadNextResponse.message_id,
146 # omci_message=OmciMibUploadNextResponse(
147 # entity_class=OntData.class_id,
148 # entity_id=0,
149 # object_entity_id=0, # TODO: Pick one
150 # object_attributes_mask=0, # TODO: Pick one
151 # object_data=None # TODO: Pick one
152 # )),
153 # ]
154 # }
155 # },
156 OP.Reboot.value: {
157 OntData.class_id: {
158 0: OmciFrame(transaction_id=0, # Will get replaced
159 message_type=OmciRebootResponse.message_id,
160 omci_message=OmciRebootResponse(
161 entity_class=OntG.class_id,
162 entity_id=0,
163 success_code=RC.Success.value
164 ))
165 }
166 },
167 }
168 # TODO: Support Autonomous ONU messages as well
169
170 def tearDown(self):
171 """Test case cleanup"""
172 pass
173
174 def _request_to_response_type(self, message_type):
175 return {
176 OP.Create.value: OmciCreateResponse,
177 OP.Delete.value: OmciDeleteResponse,
178 OP.Set.value: OmciSetResponse,
179 OP.Get.value: OmciGetResponse,
180 OP.GetNext.value: OmciGetNextResponse,
181 OP.MibUpload.value: OmciMibUploadResponse,
182 OP.MibUploadNext.value: OmciMibUploadNextResponse,
183 OP.MibReset.value: OmciMibResetResponse,
184 OP.Reboot.value: OmciRebootResponse,
185 }.get(message_type & 0x1F, None)
186
187 def rx_omci_frame(self, msg):
188 try:
189 frame = OmciFrame(msg.decode('hex'))
190 response = None
191 response_type = self._request_to_response_type(frame.fields['message_type'])
192 transaction_id = frame.fields['transaction_id']
193
194 omci_message = frame.fields.get('omci_message')
195
196 class_id = omci_message.fields.get('entity_class') \
197 if omci_message is not None else None
198 instance_id = omci_message.fields.get('entity_id') \
199 if omci_message is not None else None
200
201 # Look up hardcode responses based on class and instance ID. If found
202 # return the response, otherwise send back an error
203
204 if response_type is None:
205 status = RC.ProcessingError.value
206 elif class_id is None:
207 status = RC.UnknownEntity.value
208 elif instance_id is None:
209 status = RC.UnknownInstance.value
210 else:
211 status = RC.Success.value
212 try:
213 response_id = response_type.message_id & 0x1f
214 response = self._omci_response[response_id][class_id][instance_id]
215
216 if response_id == OP.MibUploadNext.value:
217 # Special case. Need to get requested entry
218 assert isinstance(response, list)
219 pass
220 pass
221 pass
222 pass
223
224 if isinstance(omci_message, OmciGetNext):
225 response = response[omci_message.fields['command_sequence_number']]
226
227 if isinstance(response, dict):
228 if response['failures'] > 0:
229 response['failures'] -= 1
230 return None
231 else: response = response['frame']
232
233 response.fields['transaction_id'] = transaction_id
234 if 'success_code' in response.fields['omci_message'].fields:
235 response.fields['omci_message'].fields['success_code'] = status
236
237 if status == RC.Success.value:
238 if response_type.message_id in [OmciCreateResponse.message_id,
239 OmciDeleteResponse.message_id,
240 OmciSetResponse.message_id]:
241 self.mib_data_sync += 1
242 if self.mib_data_sync > 255:
243 self.mib_data_sync = 1
244 elif response_type.message_id == OmciMibResetResponse.message_id:
245 self.mib_data_sync = 0
246
247 except KeyError as e:
248 bad_key = e.args[0]
249 if bad_key == class_id:
250 status = RC.UnknownEntity.value
251 elif bad_key == instance_id:
252 status = RC.UnknownInstance.value
253 else:
254 status = RC.ProcessingError.value
255
256 if status != RC.Success.value and \
257 response_type not in [OmciMibUploadResponse,
258 OmciMibUploadNextResponse]:
259 response = OmciFrame(transaction_id=transaction_id,
260 message_type=response_type.message_id,
261 omci_message=response_type(
262 entity_class=class_id,
263 entity_id=instance_id,
264 success_code=status
265 ))
266 return response
267
268 except Exception as e:
269 pass
270
271 @property
272 def proxy_address(self, device_id='1'):
273 if self._proxy_address is None:
274 self._proxy_address = Device.ProxyAddress(
275 device_id=device_id,
276 channel_group_id=1,
277 channel_id=1,
278 channel_termination="XGSPON",
279 onu_id=20,
280 onu_session_id=1)
281
282 return self._proxy_address
283