Loading src/device/service/drivers/mikrotik/handlers/InterfaceHandler.py 0 → 100644 +56 −0 Original line number Diff line number Diff line # Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/) # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from typing import Any, Dict, List, Tuple from ._Handler import _Handler class InterfaceHandler(_Handler): def get_resource_key(self) -> str: return '/interfaces' def compose(self, resource_key: str, resource_value: Dict[str, Any]) -> Tuple[str, Dict[str, Any]]: if resource_key.startswith('/interfaces/ip'): endpoint = '/ip/address' payload = { 'address': str(resource_value['address']), 'interface': str(resource_value['interface']), 'comment': str(resource_value.get('comment', '')), } return endpoint, payload if resource_key.startswith('/interfaces/vxlan'): endpoint = '/interface/vxlan' payload: Dict[str, Any] = {} if 'name' in resource_value: payload['name'] = str(resource_value['name']) if 'vni' in resource_value: payload['vni'] = int(resource_value['vni']) if 'local-address' in resource_value: payload['local-address'] = str(resource_value['local-address']) if 'port' in resource_value: payload['port'] = int(resource_value.get('port', 4789)) return endpoint, payload if resource_key.startswith('/interfaces/bridge/port'): endpoint = '/interface/bridge/port' payload = { 'bridge': str(resource_value['bridge']), 'interface': str(resource_value['interface']), } return endpoint, payload raise NotImplementedError(f'Resource not implemented: {resource_key}') def parse(self, raw_payload: List[Dict[str, Any]]) -> List[Tuple[str, Dict[str, Any]]]: return [] src/device/service/drivers/mikrotik/handlers/NetworkInstanceHandler.py 0 → 100644 +103 −0 Original line number Diff line number Diff line # Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/) # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from typing import Any, Dict, List, Tuple from ._Handler import _Handler class NetworkInstanceHandler(_Handler): def get_resource_key(self) -> str: return '/network_instances' def compose(self, resource_key: str, resource_value: Dict[str, Any]) -> Tuple[str, Dict[str, Any]]: if resource_key.startswith('/network_instances/bgp_instance'): return '/routing/bgp/instance', { 'name': str(resource_value['name']), 'as': int(resource_value['as']), 'router-id': str(resource_value['router_id']), } if resource_key.startswith('/network_instances/bgp_session'): endpoint = '/routing/bgp/connection' payload: Dict[str, Any] = {} if 'name' in resource_value: payload['name'] = str(resource_value['name']) if 'instance' in resource_value: payload['instance'] = resource_value['instance'] if 'local.address' in resource_value: payload['local.address'] = resource_value['local.address'] if 'remote.address' in resource_value: payload['remote.address'] = resource_value['remote.address'] if 'remote.as' in resource_value: payload['remote.as'] = resource_value['remote.as'] if 'routing-table' in resource_value: payload['routing-table'] = resource_value['routing-table'] if 'local.role' in resource_value: payload['local.role'] = resource_value['local.role'] if 'multihop' in resource_value: payload['multihop'] = resource_value['multihop'] if 'afi' in resource_value: payload['afi'] = resource_value['afi'] return endpoint, payload if resource_key.startswith('/network_instances/bgp_evpn'): endpoint = '/routing/bgp/evpn' payload: Dict[str, Any] = {} if 'name' in resource_value: payload['name'] = str(resource_value['name']) if 'instance' in resource_value: payload['instance'] = str(resource_value['instance']) if 'vni' in resource_value: payload['vni'] = int(resource_value['vni']) return endpoint, payload if resource_key.startswith('/network_instances/ospf_instance'): return '/routing/ospf/instance', { 'name': str(resource_value['name']), 'router-id': str(resource_value['router-id']), 'version': int(resource_value.get('version', 2)), } if resource_key.startswith('/network_instances/ospf_area'): return '/routing/ospf/area', { 'name': str(resource_value['name']), 'instance': str(resource_value['instance']), 'area-id': str(resource_value.get('area-id', '0.0.0.0')), } if resource_key.startswith('/network_instances/ospf_interface'): return '/routing/ospf/interface-template', { 'interfaces': [str(resource_value['interface'])], 'area': str(resource_value.get('area', 'backbone')), } if resource_key.startswith('/network_instances/isis_instance'): return '/routing/isis/instance', { 'name': str(resource_value['name']), 'areas': str(resource_value['areas']), 'system-id': str(resource_value['system-id']), } if resource_key.startswith('/network_instances/isis_interface'): level_val = resource_value.get('levels', 'l2') return '/routing/isis/interface-template', { 'instance': str(resource_value['instance']), 'interfaces': [str(resource_value['interface'])], 'levels': [str(level_val)], } raise NotImplementedError(f'Resource not implemented: {resource_key}') def parse(self, raw_payload: List[Dict[str, Any]]) -> List[Tuple[str, Dict[str, Any]]]: # Placeholder: no behavior wiring yet. return [] src/device/service/drivers/mikrotik/handlers/Tools.py 0 → 100644 +20 −0 Original line number Diff line number Diff line # Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/) # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import re RE_REMOVE_FILTERS = re.compile(r'\[[^\]]+\]') def get_schema(resource_key: str) -> str: return RE_REMOVE_FILTERS.sub('', resource_key) src/device/service/drivers/mikrotik/handlers/_Handler.py 0 → 100644 +25 −0 Original line number Diff line number Diff line # Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/) # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from typing import Any, Dict, List, Tuple class _Handler: def get_resource_key(self) -> str: raise NotImplementedError() def compose(self, resource_key: str, resource_value: Dict[str, Any]) -> Tuple[str, Dict[str, Any]]: raise NotImplementedError() def parse(self, raw_payload: List[Dict[str, Any]]) -> List[Tuple[str, Dict[str, Any]]]: raise NotImplementedError() src/device/service/drivers/mikrotik/handlers/__init__.py 0 → 100644 +60 −0 Original line number Diff line number Diff line # Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/) # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from typing import Any, Dict, List, Tuple from device.service.driver_api._Driver import ( RESOURCE_ENDPOINTS, RESOURCE_INTERFACES, RESOURCE_NETWORK_INSTANCES, ) from ._Handler import _Handler from .InterfaceHandler import InterfaceHandler from .NetworkInstanceHandler import NetworkInstanceHandler from .Tools import get_schema _interface_handler = InterfaceHandler() _network_instance_handler = NetworkInstanceHandler() ALL_RESOURCE_KEYS = [ RESOURCE_ENDPOINTS, RESOURCE_INTERFACES, RESOURCE_NETWORK_INSTANCES, ] RESOURCE_KEY_MAPPER = { RESOURCE_ENDPOINTS: _interface_handler.get_resource_key(), RESOURCE_INTERFACES: _interface_handler.get_resource_key(), RESOURCE_NETWORK_INSTANCES: _network_instance_handler.get_resource_key(), } RESOURCE_KEY_TO_HANDLER: Dict[str, _Handler] = { _interface_handler.get_resource_key(): _interface_handler, _network_instance_handler.get_resource_key(): _network_instance_handler, } def get_handler(resource_key: str) -> _Handler: resource_key_schema = get_schema(resource_key) resource_key_schema = RESOURCE_KEY_MAPPER.get(resource_key_schema, resource_key_schema) for handler_schema, handler in RESOURCE_KEY_TO_HANDLER.items(): if resource_key_schema.startswith(handler_schema): return handler raise Exception(f'Handler not found: resource_key={resource_key} resource_key_schema={resource_key_schema}') def compose(resource_key: str, resource_value: Dict[str, Any]) -> Tuple[str, Dict[str, Any]]: handler = get_handler(resource_key=resource_key) return handler.compose(resource_key, resource_value) def parse(resource_key: str, raw_payload: List[Dict[str, Any]]) -> List[Tuple[str, Dict[str, Any]]]: handler = get_handler(resource_key=resource_key) return handler.parse(raw_payload) Loading
src/device/service/drivers/mikrotik/handlers/InterfaceHandler.py 0 → 100644 +56 −0 Original line number Diff line number Diff line # Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/) # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from typing import Any, Dict, List, Tuple from ._Handler import _Handler class InterfaceHandler(_Handler): def get_resource_key(self) -> str: return '/interfaces' def compose(self, resource_key: str, resource_value: Dict[str, Any]) -> Tuple[str, Dict[str, Any]]: if resource_key.startswith('/interfaces/ip'): endpoint = '/ip/address' payload = { 'address': str(resource_value['address']), 'interface': str(resource_value['interface']), 'comment': str(resource_value.get('comment', '')), } return endpoint, payload if resource_key.startswith('/interfaces/vxlan'): endpoint = '/interface/vxlan' payload: Dict[str, Any] = {} if 'name' in resource_value: payload['name'] = str(resource_value['name']) if 'vni' in resource_value: payload['vni'] = int(resource_value['vni']) if 'local-address' in resource_value: payload['local-address'] = str(resource_value['local-address']) if 'port' in resource_value: payload['port'] = int(resource_value.get('port', 4789)) return endpoint, payload if resource_key.startswith('/interfaces/bridge/port'): endpoint = '/interface/bridge/port' payload = { 'bridge': str(resource_value['bridge']), 'interface': str(resource_value['interface']), } return endpoint, payload raise NotImplementedError(f'Resource not implemented: {resource_key}') def parse(self, raw_payload: List[Dict[str, Any]]) -> List[Tuple[str, Dict[str, Any]]]: return []
src/device/service/drivers/mikrotik/handlers/NetworkInstanceHandler.py 0 → 100644 +103 −0 Original line number Diff line number Diff line # Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/) # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from typing import Any, Dict, List, Tuple from ._Handler import _Handler class NetworkInstanceHandler(_Handler): def get_resource_key(self) -> str: return '/network_instances' def compose(self, resource_key: str, resource_value: Dict[str, Any]) -> Tuple[str, Dict[str, Any]]: if resource_key.startswith('/network_instances/bgp_instance'): return '/routing/bgp/instance', { 'name': str(resource_value['name']), 'as': int(resource_value['as']), 'router-id': str(resource_value['router_id']), } if resource_key.startswith('/network_instances/bgp_session'): endpoint = '/routing/bgp/connection' payload: Dict[str, Any] = {} if 'name' in resource_value: payload['name'] = str(resource_value['name']) if 'instance' in resource_value: payload['instance'] = resource_value['instance'] if 'local.address' in resource_value: payload['local.address'] = resource_value['local.address'] if 'remote.address' in resource_value: payload['remote.address'] = resource_value['remote.address'] if 'remote.as' in resource_value: payload['remote.as'] = resource_value['remote.as'] if 'routing-table' in resource_value: payload['routing-table'] = resource_value['routing-table'] if 'local.role' in resource_value: payload['local.role'] = resource_value['local.role'] if 'multihop' in resource_value: payload['multihop'] = resource_value['multihop'] if 'afi' in resource_value: payload['afi'] = resource_value['afi'] return endpoint, payload if resource_key.startswith('/network_instances/bgp_evpn'): endpoint = '/routing/bgp/evpn' payload: Dict[str, Any] = {} if 'name' in resource_value: payload['name'] = str(resource_value['name']) if 'instance' in resource_value: payload['instance'] = str(resource_value['instance']) if 'vni' in resource_value: payload['vni'] = int(resource_value['vni']) return endpoint, payload if resource_key.startswith('/network_instances/ospf_instance'): return '/routing/ospf/instance', { 'name': str(resource_value['name']), 'router-id': str(resource_value['router-id']), 'version': int(resource_value.get('version', 2)), } if resource_key.startswith('/network_instances/ospf_area'): return '/routing/ospf/area', { 'name': str(resource_value['name']), 'instance': str(resource_value['instance']), 'area-id': str(resource_value.get('area-id', '0.0.0.0')), } if resource_key.startswith('/network_instances/ospf_interface'): return '/routing/ospf/interface-template', { 'interfaces': [str(resource_value['interface'])], 'area': str(resource_value.get('area', 'backbone')), } if resource_key.startswith('/network_instances/isis_instance'): return '/routing/isis/instance', { 'name': str(resource_value['name']), 'areas': str(resource_value['areas']), 'system-id': str(resource_value['system-id']), } if resource_key.startswith('/network_instances/isis_interface'): level_val = resource_value.get('levels', 'l2') return '/routing/isis/interface-template', { 'instance': str(resource_value['instance']), 'interfaces': [str(resource_value['interface'])], 'levels': [str(level_val)], } raise NotImplementedError(f'Resource not implemented: {resource_key}') def parse(self, raw_payload: List[Dict[str, Any]]) -> List[Tuple[str, Dict[str, Any]]]: # Placeholder: no behavior wiring yet. return []
src/device/service/drivers/mikrotik/handlers/Tools.py 0 → 100644 +20 −0 Original line number Diff line number Diff line # Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/) # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import re RE_REMOVE_FILTERS = re.compile(r'\[[^\]]+\]') def get_schema(resource_key: str) -> str: return RE_REMOVE_FILTERS.sub('', resource_key)
src/device/service/drivers/mikrotik/handlers/_Handler.py 0 → 100644 +25 −0 Original line number Diff line number Diff line # Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/) # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from typing import Any, Dict, List, Tuple class _Handler: def get_resource_key(self) -> str: raise NotImplementedError() def compose(self, resource_key: str, resource_value: Dict[str, Any]) -> Tuple[str, Dict[str, Any]]: raise NotImplementedError() def parse(self, raw_payload: List[Dict[str, Any]]) -> List[Tuple[str, Dict[str, Any]]]: raise NotImplementedError()
src/device/service/drivers/mikrotik/handlers/__init__.py 0 → 100644 +60 −0 Original line number Diff line number Diff line # Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/) # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from typing import Any, Dict, List, Tuple from device.service.driver_api._Driver import ( RESOURCE_ENDPOINTS, RESOURCE_INTERFACES, RESOURCE_NETWORK_INSTANCES, ) from ._Handler import _Handler from .InterfaceHandler import InterfaceHandler from .NetworkInstanceHandler import NetworkInstanceHandler from .Tools import get_schema _interface_handler = InterfaceHandler() _network_instance_handler = NetworkInstanceHandler() ALL_RESOURCE_KEYS = [ RESOURCE_ENDPOINTS, RESOURCE_INTERFACES, RESOURCE_NETWORK_INSTANCES, ] RESOURCE_KEY_MAPPER = { RESOURCE_ENDPOINTS: _interface_handler.get_resource_key(), RESOURCE_INTERFACES: _interface_handler.get_resource_key(), RESOURCE_NETWORK_INSTANCES: _network_instance_handler.get_resource_key(), } RESOURCE_KEY_TO_HANDLER: Dict[str, _Handler] = { _interface_handler.get_resource_key(): _interface_handler, _network_instance_handler.get_resource_key(): _network_instance_handler, } def get_handler(resource_key: str) -> _Handler: resource_key_schema = get_schema(resource_key) resource_key_schema = RESOURCE_KEY_MAPPER.get(resource_key_schema, resource_key_schema) for handler_schema, handler in RESOURCE_KEY_TO_HANDLER.items(): if resource_key_schema.startswith(handler_schema): return handler raise Exception(f'Handler not found: resource_key={resource_key} resource_key_schema={resource_key_schema}') def compose(resource_key: str, resource_value: Dict[str, Any]) -> Tuple[str, Dict[str, Any]]: handler = get_handler(resource_key=resource_key) return handler.compose(resource_key, resource_value) def parse(resource_key: str, raw_payload: List[Dict[str, Any]]) -> List[Tuple[str, Dict[str, Any]]]: handler = get_handler(resource_key=resource_key) return handler.parse(raw_payload)
