Matteo Scandolo | 564009f | 2019-01-16 14:11:02 -0800 | [diff] [blame] | 1 | # Copyright 2017-present Open Networking Foundation |
| 2 | # |
| 3 | # Licensed under the Apache License, Version 2.0 (the "License"); |
| 4 | # you may not use this file except in compliance with the License. |
| 5 | # You may obtain a copy of the License at |
| 6 | # |
| 7 | # http://www.apache.org/licenses/LICENSE-2.0 |
| 8 | # |
| 9 | # Unless required by applicable law or agreed to in writing, software |
| 10 | # distributed under the License is distributed on an "AS IS" BASIS, |
| 11 | # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 12 | # See the License for the specific language governing permissions and |
| 13 | # limitations under the License. |
| 14 | |
| 15 | from tinydb import TinyDB, Query |
| 16 | from robot.api import logger |
| 17 | |
| 18 | class devices(object): |
| 19 | |
| 20 | def __init__(self): |
| 21 | self.olts = TinyDB('olts.json') |
| 22 | self.pon_ports = TinyDB('pon_ports.json') |
| 23 | self.onus = TinyDB('onus.json') |
| 24 | self.uni_ports = TinyDB('uni_ports.json') |
| 25 | |
| 26 | def get_mock_data(self): |
| 27 | """ |
| 28 | Get all the mock data, |
| 29 | this method is mostly intended for debugging |
| 30 | :return: a dictionary containing all the mocked data |
| 31 | """ |
| 32 | olts = self.olts.all() |
| 33 | pon_ports = self.pon_ports.all() |
| 34 | onus = self.onus.all() |
| 35 | uni_ports = self.uni_ports.all() |
| 36 | return { |
| 37 | 'olts': olts, |
| 38 | 'pon_ports': pon_ports, |
| 39 | 'onus': onus, |
| 40 | 'uni_ports': uni_ports, |
| 41 | } |
| 42 | |
| 43 | def clean_storage(self): |
| 44 | self.olts.purge() |
| 45 | self.pon_ports.purge() |
| 46 | self.onus.purge() |
| 47 | self.uni_ports.purge() |
| 48 | |
| 49 | ############################################################### |
| 50 | # OLT # |
| 51 | ############################################################### |
| 52 | |
| 53 | def create_mock_olts(self, num_olts, voltservice_id): |
| 54 | """ |
| 55 | :param num_olts: Number of OLTs to be created |
| 56 | :param voltservice_id: ID if the vOLT service |
| 57 | :return: |
| 58 | """ |
| 59 | olts = [] |
| 60 | for index in range(1, int(num_olts) + 1): |
| 61 | olt = { |
| 62 | 'name': 'Test OLT%s' % index, |
| 63 | 'volt_service_id': voltservice_id, |
| 64 | 'device_type': 'fake_olt', |
| 65 | 'host': '127.0.0.1', |
| 66 | 'port': index, |
| 67 | 'uplink': '65536', |
| 68 | 'switch_datapath_id': 'of:0000000000000001', |
| 69 | 'switch_port': str(index), |
| 70 | 'of_id': 'of:000000%s' % index, |
| 71 | } |
| 72 | # logger.info('Created OLT %s' % olt, also_console=True) |
| 73 | olts.append(olt) |
| 74 | self.olts.insert(olt) |
| 75 | |
| 76 | return olts |
| 77 | |
| 78 | def get_rest_olts(self): |
| 79 | """ |
| 80 | Get all the OLTs that have been created for the test |
| 81 | formatted for the XOS Rest API |
| 82 | :return: a list of OLTs |
| 83 | """ |
| 84 | return self.olts.all() |
| 85 | |
| 86 | def update_olt_id(self, olt, id): |
| 87 | """ |
| 88 | Update in the memory storage the XOS ID |
| 89 | of a particular OLT as it's needed to create PON Ports |
| 90 | :param olt: The OLT object to update |
| 91 | :param id: The ID returned from XOS |
| 92 | :return: None |
| 93 | """ |
| 94 | Olt = Query() |
| 95 | self.olts.update({'id': id}, Olt.name == olt['name']) |
| 96 | |
| 97 | ############################################################### |
| 98 | # PON PORT # |
| 99 | ############################################################### |
| 100 | |
| 101 | def create_mock_pon_ports(self, num_pon): |
| 102 | |
| 103 | ports = [] |
| 104 | for olt in self.olts.all(): |
| 105 | for index in range(1, int(num_pon) +1): |
| 106 | port = { |
| 107 | 'name': 'Test PonPort %s' % index, |
| 108 | 'port_no': index, |
| 109 | 'olt_device_id': olt['id'] |
| 110 | } |
| 111 | ports.append(port) |
| 112 | self.pon_ports.insert(port) |
| 113 | return ports |
| 114 | |
| 115 | def get_rest_pon_ports(self): |
| 116 | """ |
| 117 | Get all the PON Ports that have been created for the test |
| 118 | formatted for the XOS Rest API |
| 119 | :return: a list of PON Ports |
| 120 | """ |
| 121 | return self.pon_ports.all(); |
| 122 | |
| 123 | def update_pon_port_id(self, pon_port, id): |
| 124 | """ |
| 125 | Update in the memory storage the XOS ID |
| 126 | of a particular PON Port as it's needed to create ONUs |
| 127 | :param pon_port: The PON Port object to update |
| 128 | :param id: The ID returned from XOS |
| 129 | :return: None |
| 130 | """ |
| 131 | PonPort = Query() |
| 132 | self.pon_ports.update({'id': id}, PonPort.name == pon_port['name']) |
| 133 | |
| 134 | ############################################################### |
| 135 | # ONU # |
| 136 | ############################################################### |
| 137 | |
| 138 | def create_mock_onus(self, num_onus): |
| 139 | onus = [] |
| 140 | j = 0 |
| 141 | for port in self.pon_ports.all(): |
| 142 | j = j + 1 |
| 143 | for index in range(1, int(num_onus) + 1): |
| 144 | onu = { |
| 145 | 'serial_number': "ROBOT%s%s" % (j, index), |
| 146 | 'vendor': 'Robot', |
| 147 | 'pon_port_id': port['id'] |
| 148 | } |
| 149 | onus.append(onu) |
| 150 | self.onus.insert(onu) |
| 151 | return onus |
| 152 | |
| 153 | def get_rest_onus(self): |
| 154 | return self.onus.all(); |
| 155 | |
| 156 | def update_onu_id(self, onu, id): |
| 157 | Onu = Query() |
| 158 | self.onus.update({'id': id}, Onu.serial_number == onu['serial_number']) |
| 159 | |
| 160 | ############################################################### |
| 161 | # UNI # |
| 162 | ############################################################### |
| 163 | |
| 164 | def create_mock_unis(self): |
| 165 | # NOTE I believe UNI port number must be unique across OLT |
| 166 | unis = [] |
| 167 | i = 0 |
| 168 | for onu in self.onus.all(): |
| 169 | uni = { |
| 170 | 'name': 'Test UniPort %s' % i, |
| 171 | 'port_no': i, |
| 172 | 'onu_device_id': onu['id'] |
| 173 | } |
| 174 | unis.append(uni) |
| 175 | self.uni_ports.insert(uni) |
| 176 | i = i+1 |
| 177 | return unis |
| 178 | |
| 179 | def get_rest_unis(self): |
| 180 | return self.uni_ports.all(); |
| 181 | |
| 182 | def update_uni_id(self, uni, id): |
| 183 | UniPort = Query() |
| 184 | self.uni_ports.update({'id': id}, UniPort.name == uni['name']) |
| 185 | |
| 186 | ############################################################### |
| 187 | # WHITELIST # |
| 188 | ############################################################### |
| 189 | |
| 190 | def create_mock_whitelist(self, attworkflowservice_id): |
| 191 | entries = [] |
| 192 | for onu in self.onus.all(): |
| 193 | e = { |
| 194 | 'owner_id': attworkflowservice_id, |
| 195 | 'serial_number': onu['serial_number'], |
| 196 | 'pon_port_id': self._find_pon_port_by_onu(onu)['port_no'], |
| 197 | 'device_id': self._find_olt_by_onu(onu)['of_id'] |
| 198 | } |
| 199 | entries.append(e) |
| 200 | |
| 201 | return entries |
| 202 | |
| 203 | ############################################################### |
| 204 | # EVENTS # |
| 205 | ############################################################### |
| 206 | |
| 207 | def generate_onu_events(self): |
| 208 | events = [] |
| 209 | for onu in self.onus.all(): |
| 210 | ev = { |
| 211 | 'status': 'activated', |
| 212 | 'serial_number': onu['serial_number'], |
| 213 | 'uni_port_id': self._find_uni_by_onu(onu)['port_no'], |
| 214 | 'of_dpid': self._find_olt_by_onu(onu)['of_id'], |
| 215 | } |
| 216 | events.append(ev) |
| 217 | return events |
| 218 | |
| 219 | ############################################################### |
| 220 | # HELPERS # |
| 221 | ############################################################### |
| 222 | |
| 223 | def _find_uni_by_onu(self, onu): |
| 224 | Uni = Query() |
| 225 | # NOTE there's an assumption that 1 ONU has 1 UNI Port |
| 226 | return self.uni_ports.search(Uni.onu_device_id == onu['id'])[0] |
| 227 | |
| 228 | def _find_pon_port_by_onu(self, onu): |
| 229 | PonPort = Query() |
| 230 | return self.pon_ports.search(PonPort.id == onu['pon_port_id'])[0] |
| 231 | |
| 232 | def _find_olt_by_onu(self, onu): |
| 233 | pon_port = self._find_pon_port_by_onu(onu) |
| 234 | Olt = Query() |
| 235 | return self.olts.search(Olt.id == pon_port['olt_device_id'])[0] |