Loading src/service/service/service_handlers/l2nm_gnmi_openconfig/ConfigRuleComposer.py +42 −5 Original line number Diff line number Diff line Loading @@ -32,6 +32,21 @@ def _safe_int(value: Optional[object]) -> Optional[int]: except (TypeError, ValueError): return None def _safe_bool(value: Optional[object]) -> Optional[bool]: if value is None: return None if isinstance(value, bool): return value if isinstance(value, (int, float)): return bool(value) if isinstance(value, str): lowered = value.strip().lower() if lowered in {'true', '1', 'yes', 'y', 'on', 'tagged'}: return True if lowered in {'false', '0', 'no', 'n', 'off', 'untagged'}: return False return None def _interface_switched_vlan( interface : str, interface_mode : str, access_vlan_id : Optional[int] = None, trunk_vlan_id : Optional[int] = None, native_vlan : int = 1 Loading Loading @@ -102,14 +117,16 @@ class EndpointComposer: def has_vlan(self, vlan_id : int) -> bool: return vlan_id in self.get_vlan_ids() def get_config_rules(self, service_vlan_id : int, delete : bool = False) -> List[Dict]: def get_config_rules( self, service_vlan_id : int, access_vlan_tagged : bool = False, delete : bool = False ) -> List[Dict]: if self.objekt is None: MSG = 'Endpoint object not defined for uuid={:s}' LOGGER.warning(MSG.format(self.uuid)) return [] config_rules : List[Dict] = list() json_config_rule = json_config_rule_delete if delete else json_config_rule_set if self.force_trunk or len(self.explicit_vlan_ids) > 0: if self.force_trunk or access_vlan_tagged or len(self.explicit_vlan_ids) > 0: trunk_vlan_id = self._select_trunk_vlan_id(service_vlan_id) config_rules.append(json_config_rule(*_interface_switched_vlan( self.objekt.name, 'TRUNK', trunk_vlan_id=trunk_vlan_id Loading Loading @@ -161,7 +178,8 @@ class DeviceComposer: self.get_endpoint(endpoint_obj.name).configure(endpoint_obj, None) def get_config_rules( self, network_instance_name : str, service_vlan_id : int, delete : bool = False self, network_instance_name : str, service_vlan_id : int, access_vlan_tagged : bool = False, delete : bool = False ) -> List[Dict]: SELECTED_DEVICES = { DeviceTypeEnum.PACKET_POP.value, Loading @@ -176,7 +194,9 @@ class DeviceComposer: if network_instance_name != DEFAULT_NETWORK_INSTANCE: json_config_rule(*_network_instance(network_instance_name, 'L3VRF')) for endpoint in self.endpoints.values(): config_rules.extend(endpoint.get_config_rules(service_vlan_id, delete=delete)) config_rules.extend(endpoint.get_config_rules( service_vlan_id, access_vlan_tagged=access_vlan_tagged, delete=delete )) for vlan_id in sorted(self.vlan_ids): vlan_name = 'tfs-vlan-{:s}'.format(str(vlan_id)) config_rules.append(json_config_rule(*_network_instance_vlan( Loading Loading @@ -206,6 +226,7 @@ class ConfigRuleComposer: self.aliases : Dict[str, str] = dict() # device_name => device_uuid self.devices : Dict[str, DeviceComposer] = dict() # device_uuid => DeviceComposer self.vlan_id = None self.access_vlan_tagged = False def set_device_alias(self, device_name : str, device_uuid : str) -> None: self.aliases[device_name] = device_uuid Loading Loading @@ -234,6 +255,18 @@ class ConfigRuleComposer: MSG = 'Invalid VLAN ID value in service settings: {:s}' raise Exception(MSG.format(str(json_settings))) access_vlan_tagged = json_settings.get('access_vlan_tagged', json_settings.get('access-vlan-tagged')) if access_vlan_tagged is None: self.access_vlan_tagged = False else: parsed = _safe_bool(access_vlan_tagged) if parsed is None: MSG = 'Invalid access_vlan_tagged value in service settings: {:s}' LOGGER.warning(MSG.format(str(access_vlan_tagged))) self.access_vlan_tagged = False else: self.access_vlan_tagged = parsed def get_config_rules( self, network_instance_name : str = NETWORK_INSTANCE, delete : bool = False ) -> Dict[str, List[Dict]]: Loading @@ -241,7 +274,10 @@ class ConfigRuleComposer: raise Exception('VLAN ID must be configured at service level before composing rules') return { device_uuid : device.get_config_rules(network_instance_name, self.vlan_id, delete=delete) device_uuid : device.get_config_rules( network_instance_name, self.vlan_id, access_vlan_tagged=self.access_vlan_tagged, delete=delete ) for device_uuid, device in self.devices.items() } Loading @@ -252,4 +288,5 @@ class ConfigRuleComposer: for device_uuid, device in self.devices.items() }, 'vlan_id': self.vlan_id, 'access_vlan_tagged': self.access_vlan_tagged, } Loading
src/service/service/service_handlers/l2nm_gnmi_openconfig/ConfigRuleComposer.py +42 −5 Original line number Diff line number Diff line Loading @@ -32,6 +32,21 @@ def _safe_int(value: Optional[object]) -> Optional[int]: except (TypeError, ValueError): return None def _safe_bool(value: Optional[object]) -> Optional[bool]: if value is None: return None if isinstance(value, bool): return value if isinstance(value, (int, float)): return bool(value) if isinstance(value, str): lowered = value.strip().lower() if lowered in {'true', '1', 'yes', 'y', 'on', 'tagged'}: return True if lowered in {'false', '0', 'no', 'n', 'off', 'untagged'}: return False return None def _interface_switched_vlan( interface : str, interface_mode : str, access_vlan_id : Optional[int] = None, trunk_vlan_id : Optional[int] = None, native_vlan : int = 1 Loading Loading @@ -102,14 +117,16 @@ class EndpointComposer: def has_vlan(self, vlan_id : int) -> bool: return vlan_id in self.get_vlan_ids() def get_config_rules(self, service_vlan_id : int, delete : bool = False) -> List[Dict]: def get_config_rules( self, service_vlan_id : int, access_vlan_tagged : bool = False, delete : bool = False ) -> List[Dict]: if self.objekt is None: MSG = 'Endpoint object not defined for uuid={:s}' LOGGER.warning(MSG.format(self.uuid)) return [] config_rules : List[Dict] = list() json_config_rule = json_config_rule_delete if delete else json_config_rule_set if self.force_trunk or len(self.explicit_vlan_ids) > 0: if self.force_trunk or access_vlan_tagged or len(self.explicit_vlan_ids) > 0: trunk_vlan_id = self._select_trunk_vlan_id(service_vlan_id) config_rules.append(json_config_rule(*_interface_switched_vlan( self.objekt.name, 'TRUNK', trunk_vlan_id=trunk_vlan_id Loading Loading @@ -161,7 +178,8 @@ class DeviceComposer: self.get_endpoint(endpoint_obj.name).configure(endpoint_obj, None) def get_config_rules( self, network_instance_name : str, service_vlan_id : int, delete : bool = False self, network_instance_name : str, service_vlan_id : int, access_vlan_tagged : bool = False, delete : bool = False ) -> List[Dict]: SELECTED_DEVICES = { DeviceTypeEnum.PACKET_POP.value, Loading @@ -176,7 +194,9 @@ class DeviceComposer: if network_instance_name != DEFAULT_NETWORK_INSTANCE: json_config_rule(*_network_instance(network_instance_name, 'L3VRF')) for endpoint in self.endpoints.values(): config_rules.extend(endpoint.get_config_rules(service_vlan_id, delete=delete)) config_rules.extend(endpoint.get_config_rules( service_vlan_id, access_vlan_tagged=access_vlan_tagged, delete=delete )) for vlan_id in sorted(self.vlan_ids): vlan_name = 'tfs-vlan-{:s}'.format(str(vlan_id)) config_rules.append(json_config_rule(*_network_instance_vlan( Loading Loading @@ -206,6 +226,7 @@ class ConfigRuleComposer: self.aliases : Dict[str, str] = dict() # device_name => device_uuid self.devices : Dict[str, DeviceComposer] = dict() # device_uuid => DeviceComposer self.vlan_id = None self.access_vlan_tagged = False def set_device_alias(self, device_name : str, device_uuid : str) -> None: self.aliases[device_name] = device_uuid Loading Loading @@ -234,6 +255,18 @@ class ConfigRuleComposer: MSG = 'Invalid VLAN ID value in service settings: {:s}' raise Exception(MSG.format(str(json_settings))) access_vlan_tagged = json_settings.get('access_vlan_tagged', json_settings.get('access-vlan-tagged')) if access_vlan_tagged is None: self.access_vlan_tagged = False else: parsed = _safe_bool(access_vlan_tagged) if parsed is None: MSG = 'Invalid access_vlan_tagged value in service settings: {:s}' LOGGER.warning(MSG.format(str(access_vlan_tagged))) self.access_vlan_tagged = False else: self.access_vlan_tagged = parsed def get_config_rules( self, network_instance_name : str = NETWORK_INSTANCE, delete : bool = False ) -> Dict[str, List[Dict]]: Loading @@ -241,7 +274,10 @@ class ConfigRuleComposer: raise Exception('VLAN ID must be configured at service level before composing rules') return { device_uuid : device.get_config_rules(network_instance_name, self.vlan_id, delete=delete) device_uuid : device.get_config_rules( network_instance_name, self.vlan_id, access_vlan_tagged=self.access_vlan_tagged, delete=delete ) for device_uuid, device in self.devices.items() } Loading @@ -252,4 +288,5 @@ class ConfigRuleComposer: for device_uuid, device in self.devices.items() }, 'vlan_id': self.vlan_id, 'access_vlan_tagged': self.access_vlan_tagged, }