diff --git a/src/service/service/service_handlers/l3nm_gnmi_openconfig/ConfigRuleComposer.py b/src/service/service/service_handlers/l3nm_gnmi_openconfig/ConfigRuleComposer.py
index be314a8c1ee5112dd9d321dd2c1ee1dc6173aca4..5db2c5b2f2f5a73520bcea8a3fb419e0a91a577f 100644
--- a/src/service/service/service_handlers/l3nm_gnmi_openconfig/ConfigRuleComposer.py
+++ b/src/service/service/service_handlers/l3nm_gnmi_openconfig/ConfigRuleComposer.py
@@ -12,33 +12,62 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-from typing import Dict, List, Optional, Tuple
-from common.proto.context_pb2 import Device, EndPoint
+import json, netaddr, re
+from typing import Dict, List, Optional, Set, Tuple
+from common.DeviceTypes import DeviceTypeEnum
+from common.proto.context_pb2 import ConfigActionEnum, Device, EndPoint, Service
from common.tools.object_factory.ConfigRule import json_config_rule_delete, json_config_rule_set
-
from service.service.service_handler_api.AnyTreeTools import TreeNode
-def _interface(if_name, sif_index, ipv4_address, ipv4_prefix, enabled) -> Tuple[str, Dict]:
- str_path = '/interface[{:s}]'.format(if_name)
- str_data = {'name': if_name, 'enabled': enabled, 'sub_if_index': sif_index,
- 'sub_if_enabled': enabled, 'sub_if_ipv4_enabled': enabled,
- 'sub_if_ipv4_address': ipv4_address, 'sub_if_ipv4_prefix': ipv4_prefix}
- return str_path, str_data
+NETWORK_INSTANCE = 'teraflowsdn'
+
+RE_IF = re.compile(r'^\/interface\[([^\]]+)\]\/subinterface\[([^\]]+)\]$')
+RE_SR = re.compile(r'^\/network_instance\[([^\]]+)\]\/protocols\[STATIC\]/route\[ ([^\:]+)\:([^\]]+)\]$')
+
+def _interface(
+ interface : str, if_type : Optional[str] = 'l3ipvlan', index : int = 0, vlan_id : Optional[int] = None,
+ address_ip : Optional[str] = None, address_prefix : Optional[int] = None, mtu : Optional[int] = None,
+ enabled : bool = True
+) -> Tuple[str, Dict]:
+ path = '/interface[{:s}]/subinterface[{:d}]'.format(interface, index)
+ data = {'name': interface, 'type': if_type, 'index': index, 'enabled': enabled}
+ if if_type is not None: data['type'] = if_type
+ if vlan_id is not None: data['vlan_id'] = vlan_id
+ if address_ip is not None: data['address_ip'] = address_ip
+ if address_prefix is not None: data['address_prefix'] = address_prefix
+ if mtu is not None: data['mtu'] = mtu
+ return path, data
-def _network_instance(ni_name, ni_type) -> Tuple[str, Dict]:
- str_path = '/network_instance[{:s}]'.format(ni_name)
- str_data = {'name': ni_name, 'type': ni_type}
- return str_path, str_data
+def _network_instance(ni_name : str, ni_type : str) -> Tuple[str, Dict]:
+ path = '/network_instance[{:s}]'.format(ni_name)
+ data = {'name': ni_name, 'type': ni_type}
+ return path, data
-def _network_instance_static_route(ni_name, prefix, next_hop, next_hop_index=0) -> Tuple[str, Dict]:
- str_path = '/network_instance[{:s}]/static_route[{:s}]'.format(ni_name, prefix)
- str_data = {'name': ni_name, 'prefix': prefix, 'next_hop': next_hop, 'next_hop_index': next_hop_index}
- return str_path, str_data
+def _network_instance_protocol(ni_name : str, protocol : str) -> Tuple[str, Dict]:
+ path = '/network_instance[{:s}]/protocols[{:s}]'.format(ni_name, protocol)
+ data = {'name': ni_name, 'identifier': protocol, 'protocol_name': protocol}
+ return path, data
-def _network_instance_interface(ni_name, if_name, sif_index) -> Tuple[str, Dict]:
- str_path = '/network_instance[{:s}]/interface[{:s}.{:d}]'.format(ni_name, if_name, sif_index)
- str_data = {'name': ni_name, 'if_name': if_name, 'sif_index': sif_index}
- return str_path, str_data
+def _network_instance_protocol_static(ni_name : str) -> Tuple[str, Dict]:
+ return _network_instance_protocol(ni_name, 'STATIC')
+
+def _network_instance_protocol_static_route(
+ ni_name : str, prefix : str, next_hop : str, index : int = 0, metric : Optional[int] = None
+) -> Tuple[str, Dict]:
+ protocol = 'STATIC'
+ path = '/network_instance[{:s}]/protocols[{:s}]/static_route[{:s}:{:d}]'.format(ni_name, protocol, prefix, index)
+ data = {
+ 'name': ni_name, 'identifier': protocol, 'protocol_name': protocol,
+ 'prefix': prefix, 'index': index, 'next_hop': next_hop
+ }
+ if metric is not None: data['metric'] = metric
+ return path, data
+
+def _network_instance_interface(ni_name : str, interface : str, sub_interface_index : int) -> Tuple[str, Dict]:
+ sub_interface_name = '{:s}.{:d}'.format(interface, sub_interface_index)
+ path = '/network_instance[{:s}]/interface[{:s}]'.format(ni_name, sub_interface_name)
+ data = {'name': ni_name, 'id': sub_interface_name, 'interface': interface, 'subinterface': sub_interface_index}
+ return path, data
class EndpointComposer:
def __init__(self, endpoint_uuid : str) -> None:
@@ -46,33 +75,47 @@ class EndpointComposer:
self.objekt : Optional[EndPoint] = None
self.sub_interface_index = 0
self.ipv4_address = None
- self.ipv4_prefix = None
+ self.ipv4_prefix_len = None
- def configure(self, endpoint_obj : EndPoint, settings : Optional[TreeNode]) -> None:
- self.objekt = endpoint_obj
+ def configure(self, endpoint_obj : Optional[EndPoint], settings : Optional[TreeNode]) -> None:
+ if endpoint_obj is not None:
+ self.objekt = endpoint_obj
if settings is None: return
json_settings : Dict = settings.value
self.ipv4_address = json_settings['ipv4_address']
- self.ipv4_prefix = json_settings['ipv4_prefix']
+ self.ipv4_prefix_len = json_settings['ipv4_prefix_len']
self.sub_interface_index = json_settings['sub_interface_index']
def get_config_rules(self, network_instance_name : str, delete : bool = False) -> List[Dict]:
+ if self.ipv4_address is None: return []
+ if self.ipv4_prefix_len is None: return []
json_config_rule = json_config_rule_delete if delete else json_config_rule_set
return [
json_config_rule(*_interface(
- self.objekt.name, self.sub_interface_index, self.ipv4_address, self.ipv4_prefix, True
+ self.objekt.name, index=self.sub_interface_index, address_ip=self.ipv4_address,
+ address_prefix=self.ipv4_prefix_len, enabled=True
)),
json_config_rule(*_network_instance_interface(
network_instance_name, self.objekt.name, self.sub_interface_index
)),
]
+ def dump(self) -> Dict:
+ return {
+ 'sub_interface_index' : self.sub_interface_index,
+ 'ipv4_address' : self.ipv4_address,
+ 'ipv4_prefix_len' : self.ipv4_prefix_len,
+ }
+
class DeviceComposer:
def __init__(self, device_uuid : str) -> None:
self.uuid = device_uuid
self.objekt : Optional[Device] = None
self.endpoints : Dict[str, EndpointComposer] = dict()
- self.static_routes : Dict[str, str] = dict()
+ self.connected : Set[str] = set()
+
+ # {prefix => {index => (next_hop, metric)}}
+ self.static_routes : Dict[str, Dict[int, Tuple[str, Optional[int]]]] = dict()
def get_endpoint(self, endpoint_uuid : str) -> EndpointComposer:
if endpoint_uuid not in self.endpoints:
@@ -81,39 +124,108 @@ class DeviceComposer:
def configure(self, device_obj : Device, settings : Optional[TreeNode]) -> None:
self.objekt = device_obj
+ for endpoint_obj in device_obj.device_endpoints:
+ self.get_endpoint(endpoint_obj.name).configure(endpoint_obj, None)
+
+ for config_rule in device_obj.device_config.config_rules:
+ if config_rule.action != ConfigActionEnum.CONFIGACTION_SET: continue
+ if config_rule.WhichOneof('config_rule') != 'custom': continue
+ config_rule_custom = config_rule.custom
+
+ match = RE_IF.match(config_rule_custom.resource_key)
+ if match is not None:
+ if_name, subif_index = match.groups()
+ resource_value = json.loads(config_rule_custom.resource_value)
+ ipv4_network = str(resource_value['address_ip'])
+ ipv4_prefix_len = int(resource_value['address_prefix'])
+ endpoint = self.get_endpoint(if_name)
+ endpoint.ipv4_address = ipv4_network
+ endpoint.ipv4_prefix_len = ipv4_prefix_len
+ endpoint.sub_interface_index = int(subif_index)
+ endpoint_ip_network = netaddr.IPNetwork('{:s}/{:d}'.format(ipv4_network, ipv4_prefix_len))
+ self.connected.add(str(endpoint_ip_network.cidr))
+
+ match = RE_SR.match(config_rule_custom.resource_key)
+ if match is not None:
+ ni_name, prefix, index = match.groups()
+ if ni_name != NETWORK_INSTANCE: continue
+ resource_value : Dict = json.loads(config_rule_custom.resource_value)
+ next_hop = resource_value['next_hop']
+ metric = resource_value.get('metric')
+ self.static_routes.setdefault(prefix, dict())[index] = (next_hop, metric)
+
if settings is None: return
json_settings : Dict = settings.value
- static_routes = json_settings.get('static_routes', [])
+ static_routes : List[Dict] = json_settings.get('static_routes', [])
for static_route in static_routes:
prefix = static_route['prefix']
+ index = static_route.get('index', 0)
next_hop = static_route['next_hop']
- self.static_routes[prefix] = next_hop
+ metric = static_route.get('metric')
+ self.static_routes.setdefault(prefix, dict())[index] = (next_hop, metric)
def get_config_rules(self, network_instance_name : str, delete : bool = False) -> List[Dict]:
+ SELECTED_DEVICES = {DeviceTypeEnum.PACKET_ROUTER.value, DeviceTypeEnum.EMULATED_PACKET_ROUTER.value}
+ if self.objekt.device_type not in SELECTED_DEVICES: return []
+
json_config_rule = json_config_rule_delete if delete else json_config_rule_set
config_rules = [
json_config_rule(*_network_instance(network_instance_name, 'L3VRF'))
]
for endpoint in self.endpoints.values():
config_rules.extend(endpoint.get_config_rules(network_instance_name, delete=delete))
- for prefix, next_hop in self.static_routes.items():
+ if len(self.static_routes) > 0:
config_rules.append(
- json_config_rule(*_network_instance_static_route(network_instance_name, prefix, next_hop))
+ json_config_rule(*_network_instance_protocol_static(network_instance_name))
)
+ for prefix, indexed_static_rule in self.static_routes.items():
+ for index, (next_hop, metric) in indexed_static_rule.items():
+ config_rules.append(
+ json_config_rule(*_network_instance_protocol_static_route(
+ network_instance_name, prefix, next_hop, index=index, metric=metric
+ ))
+ )
if delete: config_rules = list(reversed(config_rules))
return config_rules
+ def dump(self) -> Dict:
+ return {
+ 'endpoints' : {
+ endpoint_uuid : endpoint.dump()
+ for endpoint_uuid, endpoint in self.endpoints.items()
+ },
+ 'connected' : list(self.connected),
+ 'static_routes' : self.static_routes,
+ }
+
class ConfigRuleComposer:
def __init__(self) -> None:
+ self.objekt : Optional[Service] = None
self.devices : Dict[str, DeviceComposer] = dict()
+ def configure(self, service_obj : Service, settings : Optional[TreeNode]) -> None:
+ self.objekt = service_obj
+ if settings is None: return
+ #json_settings : Dict = settings.value
+ # For future use
+
def get_device(self, device_uuid : str) -> DeviceComposer:
if device_uuid not in self.devices:
self.devices[device_uuid] = DeviceComposer(device_uuid)
return self.devices[device_uuid]
- def get_config_rules(self, network_instance_name : str, delete : bool = False) -> Dict[str, List[Dict]]:
+ def get_config_rules(
+ self, network_instance_name : str = NETWORK_INSTANCE, delete : bool = False
+ ) -> Dict[str, List[Dict]]:
return {
device_uuid : device.get_config_rules(network_instance_name, delete=delete)
for device_uuid, device in self.devices.items()
}
+
+ def dump(self) -> Dict:
+ return {
+ 'devices' : {
+ device_uuid : device.dump()
+ for device_uuid, device in self.devices.items()
+ }
+ }
diff --git a/src/service/service/service_handlers/l3nm_gnmi_openconfig/L3NMGnmiOpenConfigServiceHandler.py b/src/service/service/service_handlers/l3nm_gnmi_openconfig/L3NMGnmiOpenConfigServiceHandler.py
index 5856b5f61893174a92ce02a303ae9ad30be16005..9142c9d1e32c698aef8cc1c461d6212a64b303dc 100644
--- a/src/service/service/service_handlers/l3nm_gnmi_openconfig/L3NMGnmiOpenConfigServiceHandler.py
+++ b/src/service/service/service_handlers/l3nm_gnmi_openconfig/L3NMGnmiOpenConfigServiceHandler.py
@@ -18,11 +18,12 @@ from common.method_wrappers.Decorator import MetricsPool, metered_subclass_metho
from common.proto.context_pb2 import ConfigRule, DeviceId, Service
from common.tools.object_factory.Device import json_device_id
from common.type_checkers.Checkers import chk_type
-from service.service.service_handler_api.Tools import get_device_endpoint_uuids, get_endpoint_matching
from service.service.service_handler_api._ServiceHandler import _ServiceHandler
from service.service.service_handler_api.SettingsHandler import SettingsHandler
+from service.service.service_handler_api.Tools import get_device_endpoint_uuids, get_endpoint_matching
from service.service.task_scheduler.TaskExecutor import TaskExecutor
from .ConfigRuleComposer import ConfigRuleComposer
+from .StaticRouteGenerator import StaticRouteGenerator
LOGGER = logging.getLogger(__name__)
@@ -35,16 +36,22 @@ class L3NMGnmiOpenConfigServiceHandler(_ServiceHandler):
self.__service = service
self.__task_executor = task_executor
self.__settings_handler = SettingsHandler(service.service_config, **settings)
- self.__composer = ConfigRuleComposer()
- self.__endpoint_map : Dict[Tuple[str, str], str] = dict()
+ self.__config_rule_composer = ConfigRuleComposer()
+ self.__static_route_generator = StaticRouteGenerator(self.__config_rule_composer)
+ self.__endpoint_map : Dict[Tuple[str, str], Tuple[str, str]] = dict()
def _compose_config_rules(self, endpoints : List[Tuple[str, str, Optional[str]]]) -> None:
+ if len(endpoints) % 2 != 0: raise Exception('Number of endpoints should be even')
+
+ service_settings = self.__settings_handler.get_service_settings()
+ self.__config_rule_composer.configure(self.__service, service_settings)
+
for endpoint in endpoints:
device_uuid, endpoint_uuid = get_device_endpoint_uuids(endpoint)
device_obj = self.__task_executor.get_device(DeviceId(**json_device_id(device_uuid)))
device_settings = self.__settings_handler.get_device_settings(device_obj)
- _device = self.__composer.get_device(device_obj.name)
+ _device = self.__config_rule_composer.get_device(device_obj.name)
_device.configure(device_obj, device_settings)
endpoint_obj = get_endpoint_matching(device_obj, endpoint_uuid)
@@ -52,7 +59,9 @@ class L3NMGnmiOpenConfigServiceHandler(_ServiceHandler):
_endpoint = _device.get_endpoint(endpoint_obj.name)
_endpoint.configure(endpoint_obj, endpoint_settings)
- self.__endpoint_map[(device_uuid, endpoint_uuid)] = device_obj.name
+ self.__endpoint_map[(device_uuid, endpoint_uuid)] = (device_obj.name, endpoint_obj.name)
+
+ self.__static_route_generator.compose(endpoints)
def _do_configurations(
self, config_rules_per_device : Dict[str, List[Dict]], endpoints : List[Tuple[str, str, Optional[str]]],
@@ -62,7 +71,7 @@ class L3NMGnmiOpenConfigServiceHandler(_ServiceHandler):
results_per_device = dict()
for device_name,json_config_rules in config_rules_per_device.items():
try:
- device_obj = self.__composer.get_device(device_name).objekt
+ device_obj = self.__config_rule_composer.get_device(device_name).objekt
if len(json_config_rules) == 0: continue
del device_obj.device_config.config_rules[:]
for json_config_rule in json_config_rules:
@@ -78,7 +87,8 @@ class L3NMGnmiOpenConfigServiceHandler(_ServiceHandler):
results = []
for endpoint in endpoints:
device_uuid, endpoint_uuid = get_device_endpoint_uuids(endpoint)
- device_name = self.__endpoint_map[(device_uuid, endpoint_uuid)]
+ device_name, _ = self.__endpoint_map[(device_uuid, endpoint_uuid)]
+ if device_name not in results_per_device: continue
results.append(results_per_device[device_name])
return results
@@ -88,12 +98,14 @@ class L3NMGnmiOpenConfigServiceHandler(_ServiceHandler):
) -> List[Union[bool, Exception]]:
chk_type('endpoints', endpoints, list)
if len(endpoints) == 0: return []
- service_uuid = self.__service.service_id.service_uuid.uuid
- #settings = self.__settings_handler.get('/settings')
+ #service_uuid = self.__service.service_id.service_uuid.uuid
self._compose_config_rules(endpoints)
- network_instance_name = service_uuid.split('-')[0]
- config_rules_per_device = self.__composer.get_config_rules(network_instance_name, delete=False)
+ #network_instance_name = service_uuid.split('-')[0]
+ #config_rules_per_device = self.__config_rule_composer.get_config_rules(network_instance_name, delete=False)
+ config_rules_per_device = self.__config_rule_composer.get_config_rules(delete=False)
+ LOGGER.debug('config_rules_per_device={:s}'.format(str(config_rules_per_device)))
results = self._do_configurations(config_rules_per_device, endpoints)
+ LOGGER.debug('results={:s}'.format(str(results)))
return results
@metered_subclass_method(METRICS_POOL)
@@ -102,12 +114,14 @@ class L3NMGnmiOpenConfigServiceHandler(_ServiceHandler):
) -> List[Union[bool, Exception]]:
chk_type('endpoints', endpoints, list)
if len(endpoints) == 0: return []
- service_uuid = self.__service.service_id.service_uuid.uuid
- #settings = self.__settings_handler.get('/settings')
+ #service_uuid = self.__service.service_id.service_uuid.uuid
self._compose_config_rules(endpoints)
- network_instance_name = service_uuid.split('-')[0]
- config_rules_per_device = self.__composer.get_config_rules(network_instance_name, delete=True)
+ #network_instance_name = service_uuid.split('-')[0]
+ #config_rules_per_device = self.__config_rule_composer.get_config_rules(network_instance_name, delete=True)
+ config_rules_per_device = self.__config_rule_composer.get_config_rules(delete=True)
+ LOGGER.debug('config_rules_per_device={:s}'.format(str(config_rules_per_device)))
results = self._do_configurations(config_rules_per_device, endpoints, delete=True)
+ LOGGER.debug('results={:s}'.format(str(results)))
return results
@metered_subclass_method(METRICS_POOL)
diff --git a/src/service/service/service_handlers/l3nm_gnmi_openconfig/StaticRouteGenerator.py b/src/service/service/service_handlers/l3nm_gnmi_openconfig/StaticRouteGenerator.py
new file mode 100644
index 0000000000000000000000000000000000000000..6479a07fe98db313cf1d0e5d0ee371ae8e599388
--- /dev/null
+++ b/src/service/service/service_handlers/l3nm_gnmi_openconfig/StaticRouteGenerator.py
@@ -0,0 +1,183 @@
+# Copyright 2022-2024 ETSI OSG/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 json, logging, netaddr
+from typing import List, Optional, Tuple
+from .ConfigRuleComposer import ConfigRuleComposer
+
+LOGGER = logging.getLogger(__name__)
+
+# Used to infer routing networks for adjacent ports when there is no hint in device/endpoint settings
+ROOT_NEIGHBOR_ROUTING_NETWORK = netaddr.IPNetwork('10.254.254.0/16')
+NEIGHBOR_ROUTING_NETWORKS_PREFIX_LEN = 30
+NEIGHBOR_ROUTING_NETWORKS = set(ROOT_NEIGHBOR_ROUTING_NETWORK.subnet(NEIGHBOR_ROUTING_NETWORKS_PREFIX_LEN))
+
+def _generate_neighbor_addresses() -> Tuple[netaddr.IPAddress, netaddr.IPAddress, int]:
+ ip_network = NEIGHBOR_ROUTING_NETWORKS.pop()
+ ip_addresses = list(ip_network.iter_hosts())
+ ip_addresses.append(NEIGHBOR_ROUTING_NETWORKS_PREFIX_LEN)
+ return ip_addresses
+
+def _compute_gateway(ip_network : netaddr.IPNetwork, gateway_host=1) -> netaddr.IPAddress:
+ return netaddr.IPAddress(ip_network.cidr.first + gateway_host)
+
+def _compose_ipv4_network(ipv4_network, ipv4_prefix_len) -> netaddr.IPNetwork:
+ return netaddr.IPNetwork('{:s}/{:d}'.format(str(ipv4_network), int(ipv4_prefix_len)))
+
+class StaticRouteGenerator:
+ def __init__(self, config_rule_composer : ConfigRuleComposer) -> None:
+ self._config_rule_composer = config_rule_composer
+
+ def compose(self, connection_hop_list : List[Tuple[str, str, Optional[str]]]) -> None:
+ link_endpoints = self._compute_link_endpoints(connection_hop_list)
+ LOGGER.debug('link_endpoints = {:s}'.format(str(link_endpoints)))
+
+ self._compute_link_addresses(link_endpoints)
+ LOGGER.debug('config_rule_composer = {:s}'.format(json.dumps(self._config_rule_composer.dump())))
+
+ self._discover_connected_networks(connection_hop_list)
+ LOGGER.debug('config_rule_composer = {:s}'.format(json.dumps(self._config_rule_composer.dump())))
+
+ # Compute and propagate static routes forward (service_endpoint_a => service_endpoint_b)
+ self._compute_static_routes(link_endpoints)
+
+ # Compute and propagate static routes backward (service_endpoint_b => service_endpoint_a)
+ reversed_endpoints = list(reversed(connection_hop_list))
+ reversed_link_endpoints = self._compute_link_endpoints(reversed_endpoints)
+ LOGGER.debug('reversed_link_endpoints = {:s}'.format(str(reversed_link_endpoints)))
+ self._compute_static_routes(reversed_link_endpoints)
+
+ LOGGER.debug('config_rule_composer = {:s}'.format(json.dumps(self._config_rule_composer.dump())))
+
+ def _compute_link_endpoints(
+ self, connection_hop_list : List[Tuple[str, str, Optional[str]]]
+ ) -> List[Tuple[Tuple[str, str, Optional[str]], Tuple[str, str, Optional[str]]]]:
+ num_connection_hops = len(connection_hop_list)
+ if num_connection_hops % 2 != 0: raise Exception('Number of connection hops must be even')
+ if num_connection_hops < 4: raise Exception('Number of connection hops must be >= 4')
+
+ # Skip service endpoints (first and last)
+ it_connection_hops = iter(connection_hop_list[1:-1])
+ return list(zip(it_connection_hops, it_connection_hops))
+
+ def _compute_link_addresses(
+ self, link_endpoints_list : List[Tuple[Tuple[str, str, Optional[str]], Tuple[str, str, Optional[str]]]]
+ ) -> None:
+ for link_endpoints in link_endpoints_list:
+ device_endpoint_a, device_endpoint_b = link_endpoints
+
+ device_uuid_a, endpoint_uuid_a = device_endpoint_a[0:2]
+ endpoint_a = self._config_rule_composer.get_device(device_uuid_a).get_endpoint(endpoint_uuid_a)
+
+ device_uuid_b, endpoint_uuid_b = device_endpoint_b[0:2]
+ endpoint_b = self._config_rule_composer.get_device(device_uuid_b).get_endpoint(endpoint_uuid_b)
+
+ if endpoint_a.ipv4_address is None and endpoint_b.ipv4_address is None:
+ ip_endpoint_a, ip_endpoint_b, prefix_len = _generate_neighbor_addresses()
+ endpoint_a.ipv4_address = str(ip_endpoint_a)
+ endpoint_a.ipv4_prefix_len = prefix_len
+ endpoint_b.ipv4_address = str(ip_endpoint_b)
+ endpoint_b.ipv4_prefix_len = prefix_len
+ elif endpoint_a.ipv4_address is not None and endpoint_b.ipv4_address is None:
+ prefix_len = endpoint_a.ipv4_prefix_len
+ ip_network_a = _compose_ipv4_network(endpoint_a.ipv4_address, prefix_len)
+ if prefix_len > 30:
+ MSG = 'Unsupported prefix_len for {:s}: {:s}'
+ raise Exception(MSG.format(str(endpoint_a), str(prefix_len)))
+ ip_endpoint_b = _compute_gateway(ip_network_a, gateway_host=1)
+ if ip_endpoint_b == ip_network_a.ip:
+ ip_endpoint_b = _compute_gateway(ip_network_a, gateway_host=2)
+ endpoint_b.ipv4_address = str(ip_endpoint_b)
+ endpoint_b.ipv4_prefix_len = prefix_len
+ elif endpoint_a.ipv4_address is None and endpoint_b.ipv4_address is not None:
+ prefix_len = endpoint_b.ipv4_prefix_len
+ ip_network_b = _compose_ipv4_network(endpoint_b.ipv4_address, prefix_len)
+ if prefix_len > 30:
+ MSG = 'Unsupported prefix_len for {:s}: {:s}'
+ raise Exception(MSG.format(str(endpoint_b), str(prefix_len)))
+ ip_endpoint_a = _compute_gateway(ip_network_b, gateway_host=1)
+ if ip_endpoint_a == ip_network_b.ip:
+ ip_endpoint_a = _compute_gateway(ip_network_b, gateway_host=2)
+ endpoint_a.ipv4_address = str(ip_endpoint_a)
+ endpoint_a.ipv4_prefix_len = prefix_len
+ elif endpoint_a.ipv4_address is not None and endpoint_b.ipv4_address is not None:
+ ip_network_a = _compose_ipv4_network(endpoint_a.ipv4_address, endpoint_a.ipv4_prefix_len)
+ ip_network_b = _compose_ipv4_network(endpoint_b.ipv4_address, endpoint_b.ipv4_prefix_len)
+ if ip_network_a.cidr != ip_network_b.cidr:
+ MSG = 'Incompatible CIDRs: endpoint_a({:s})=>{:s} endpoint_b({:s})=>{:s}'
+ raise Exception(MSG.format(str(endpoint_a), str(ip_network_a), str(endpoint_b), str(ip_network_b)))
+ if ip_network_a.ip == ip_network_b.ip:
+ MSG = 'Duplicated IP: endpoint_a({:s})=>{:s} endpoint_b({:s})=>{:s}'
+ raise Exception(MSG.format(str(endpoint_a), str(ip_network_a), str(endpoint_b), str(ip_network_b)))
+
+ def _discover_connected_networks(self, connection_hop_list : List[Tuple[str, str, Optional[str]]]) -> None:
+ for connection_hop in connection_hop_list:
+ device_uuid, endpoint_uuid = connection_hop[0:2]
+ device = self._config_rule_composer.get_device(device_uuid)
+ endpoint = device.get_endpoint(endpoint_uuid)
+
+ if endpoint.ipv4_address is None: continue
+ ip_network = _compose_ipv4_network(endpoint.ipv4_address, endpoint.ipv4_prefix_len)
+
+ device.connected.add(str(ip_network.cidr))
+
+ def _compute_static_routes(
+ self, link_endpoints_list : List[Tuple[Tuple[str, str, Optional[str]], Tuple[str, str, Optional[str]]]]
+ ) -> None:
+ for link_endpoints in link_endpoints_list:
+ device_endpoint_a, device_endpoint_b = link_endpoints
+
+ device_uuid_a, endpoint_uuid_a = device_endpoint_a[0:2]
+ device_a = self._config_rule_composer.get_device(device_uuid_a)
+ endpoint_a = device_a.get_endpoint(endpoint_uuid_a)
+
+ device_uuid_b, endpoint_uuid_b = device_endpoint_b[0:2]
+ device_b = self._config_rule_composer.get_device(device_uuid_b)
+ endpoint_b = device_b.get_endpoint(endpoint_uuid_b)
+
+ # Compute static routes from networks connected in device_a
+ for ip_network_a in device_a.connected:
+ if ip_network_a in device_b.connected: continue
+ if ip_network_a in device_b.static_routes: continue
+ if ip_network_a in ROOT_NEIGHBOR_ROUTING_NETWORK: continue
+ endpoint_a_ip_network = _compose_ipv4_network(endpoint_a.ipv4_address, endpoint_a.ipv4_prefix_len)
+ next_hop = str(endpoint_a_ip_network.ip)
+ device_b.static_routes.setdefault(ip_network_a, dict())[0] = (next_hop, None)
+
+ # Compute static routes from networks connected in device_b
+ for ip_network_b in device_b.connected:
+ if ip_network_b in device_a.connected: continue
+ if ip_network_b in device_a.static_routes: continue
+ if ip_network_b in ROOT_NEIGHBOR_ROUTING_NETWORK: continue
+ endpoint_b_ip_network = _compose_ipv4_network(endpoint_b.ipv4_address, endpoint_b.ipv4_prefix_len)
+ next_hop = str(endpoint_b_ip_network.ip)
+ device_a.static_routes.setdefault(ip_network_b, dict())[0] = (next_hop, None)
+
+ # Propagate static routes from networks connected in device_a
+ for ip_network_a in device_a.static_routes.keys():
+ if ip_network_a in device_b.connected: continue
+ if ip_network_a in device_b.static_routes: continue
+ if ip_network_a in ROOT_NEIGHBOR_ROUTING_NETWORK: continue
+ endpoint_a_ip_network = _compose_ipv4_network(endpoint_a.ipv4_address, endpoint_a.ipv4_prefix_len)
+ next_hop = str(endpoint_a_ip_network.ip)
+ device_b.static_routes.setdefault(ip_network_a, dict())[0] = (next_hop, None)
+
+ # Propagate static routes from networks connected in device_b
+ for ip_network_b in device_b.static_routes.keys():
+ if ip_network_b in device_a.connected: continue
+ if ip_network_b in device_a.static_routes: continue
+ if ip_network_b in ROOT_NEIGHBOR_ROUTING_NETWORK: continue
+ endpoint_b_ip_network = _compose_ipv4_network(endpoint_b.ipv4_address, endpoint_b.ipv4_prefix_len)
+ next_hop = str(endpoint_b_ip_network.ip)
+ device_a.static_routes.setdefault(ip_network_b, dict())[0] = (next_hop, None)
diff --git a/src/service/tests/test_l3nm_gnmi_static_rule_gen/MockServiceHandler.py b/src/service/tests/test_l3nm_gnmi_static_rule_gen/MockServiceHandler.py
new file mode 100644
index 0000000000000000000000000000000000000000..9b3f76566c9d8e5b2c8bdfb05f4b2448c29b7eae
--- /dev/null
+++ b/src/service/tests/test_l3nm_gnmi_static_rule_gen/MockServiceHandler.py
@@ -0,0 +1,160 @@
+# Copyright 2022-2024 ETSI OSG/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 json, logging
+from typing import Any, Dict, List, Optional, Tuple, Union
+from common.proto.context_pb2 import ConfigRule, DeviceId, Service
+from common.tools.object_factory.Device import json_device_id
+from common.type_checkers.Checkers import chk_type
+from service.service.service_handler_api._ServiceHandler import _ServiceHandler
+from service.service.service_handler_api.SettingsHandler import SettingsHandler
+from service.service.service_handler_api.Tools import get_device_endpoint_uuids, get_endpoint_matching
+from .MockTaskExecutor import MockTaskExecutor
+from service.service.service_handlers.l3nm_gnmi_openconfig.ConfigRuleComposer import ConfigRuleComposer
+from service.service.service_handlers.l3nm_gnmi_openconfig.StaticRouteGenerator import StaticRouteGenerator
+
+LOGGER = logging.getLogger(__name__)
+
+class MockServiceHandler(_ServiceHandler):
+ def __init__( # pylint: disable=super-init-not-called
+ self, service : Service, task_executor : MockTaskExecutor, **settings
+ ) -> None:
+ self.__service = service
+ self.__task_executor = task_executor
+ self.__settings_handler = SettingsHandler(service.service_config, **settings)
+ self.__config_rule_composer = ConfigRuleComposer()
+ self.__static_route_generator = StaticRouteGenerator(self.__config_rule_composer)
+ self.__endpoint_map : Dict[Tuple[str, str], Tuple[str, str]] = dict()
+
+ def _compose_config_rules(self, endpoints : List[Tuple[str, str, Optional[str]]]) -> None:
+ if len(endpoints) % 2 != 0: raise Exception('Number of endpoints should be even')
+
+ service_settings = self.__settings_handler.get_service_settings()
+ self.__config_rule_composer.configure(self.__service, service_settings)
+
+ for endpoint in endpoints:
+ device_uuid, endpoint_uuid = get_device_endpoint_uuids(endpoint)
+
+ device_obj = self.__task_executor.get_device(DeviceId(**json_device_id(device_uuid)))
+ device_settings = self.__settings_handler.get_device_settings(device_obj)
+ _device = self.__config_rule_composer.get_device(device_obj.name)
+ _device.configure(device_obj, device_settings)
+
+ endpoint_obj = get_endpoint_matching(device_obj, endpoint_uuid)
+ endpoint_settings = self.__settings_handler.get_endpoint_settings(device_obj, endpoint_obj)
+ _endpoint = _device.get_endpoint(endpoint_obj.name)
+ _endpoint.configure(endpoint_obj, endpoint_settings)
+
+ self.__endpoint_map[(device_uuid, endpoint_uuid)] = (device_obj.name, endpoint_obj.name)
+
+ self.__static_route_generator.compose(endpoints)
+
+ def _do_configurations(
+ self, config_rules_per_device : Dict[str, List[Dict]], endpoints : List[Tuple[str, str, Optional[str]]],
+ delete : bool = False
+ ) -> List[Union[bool, Exception]]:
+ # Configuration is done atomically on each device, all OK / all KO per device
+ results_per_device = dict()
+ for device_name,json_config_rules in config_rules_per_device.items():
+ try:
+ device_obj = self.__config_rule_composer.get_device(device_name).objekt
+ if len(json_config_rules) == 0: continue
+ del device_obj.device_config.config_rules[:]
+ for json_config_rule in json_config_rules:
+ device_obj.device_config.config_rules.append(ConfigRule(**json_config_rule))
+ self.__task_executor.configure_device(device_obj)
+ results_per_device[device_name] = True
+ except Exception as e: # pylint: disable=broad-exception-caught
+ verb = 'deconfigure' if delete else 'configure'
+ MSG = 'Unable to {:s} Device({:s}) : ConfigRules({:s})'
+ LOGGER.exception(MSG.format(verb, str(device_name), str(json_config_rules)))
+ results_per_device[device_name] = e
+
+ results = []
+ for endpoint in endpoints:
+ device_uuid, endpoint_uuid = get_device_endpoint_uuids(endpoint)
+ device_name, _ = self.__endpoint_map[(device_uuid, endpoint_uuid)]
+ if device_name not in results_per_device: continue
+ results.append(results_per_device[device_name])
+ return results
+
+ def SetEndpoint(
+ self, endpoints : List[Tuple[str, str, Optional[str]]], connection_uuid : Optional[str] = None
+ ) -> List[Union[bool, Exception]]:
+ chk_type('endpoints', endpoints, list)
+ if len(endpoints) == 0: return []
+ self._compose_config_rules(endpoints)
+ config_rules_per_device = self.__config_rule_composer.get_config_rules(delete=False)
+ LOGGER.debug('config_rules_per_device={:s}'.format(str(config_rules_per_device)))
+ results = self._do_configurations(config_rules_per_device, endpoints)
+ LOGGER.debug('results={:s}'.format(str(results)))
+ return results
+
+ def DeleteEndpoint(
+ self, endpoints : List[Tuple[str, str, Optional[str]]], connection_uuid : Optional[str] = None
+ ) -> List[Union[bool, Exception]]:
+ chk_type('endpoints', endpoints, list)
+ if len(endpoints) == 0: return []
+ self._compose_config_rules(endpoints)
+ config_rules_per_device = self.__config_rule_composer.get_config_rules(delete=True)
+ LOGGER.debug('config_rules_per_device={:s}'.format(str(config_rules_per_device)))
+ results = self._do_configurations(config_rules_per_device, endpoints, delete=True)
+ LOGGER.debug('results={:s}'.format(str(results)))
+ return results
+
+ def SetConstraint(self, constraints : List[Tuple[str, Any]]) -> List[Union[bool, Exception]]:
+ chk_type('constraints', constraints, list)
+ if len(constraints) == 0: return []
+
+ msg = '[SetConstraint] Method not implemented. Constraints({:s}) are being ignored.'
+ LOGGER.warning(msg.format(str(constraints)))
+ return [True for _ in range(len(constraints))]
+
+ def DeleteConstraint(self, constraints : List[Tuple[str, Any]]) -> List[Union[bool, Exception]]:
+ chk_type('constraints', constraints, list)
+ if len(constraints) == 0: return []
+
+ msg = '[DeleteConstraint] Method not implemented. Constraints({:s}) are being ignored.'
+ LOGGER.warning(msg.format(str(constraints)))
+ return [True for _ in range(len(constraints))]
+
+ def SetConfig(self, resources : List[Tuple[str, Any]]) -> List[Union[bool, Exception]]:
+ chk_type('resources', resources, list)
+ if len(resources) == 0: return []
+
+ results = []
+ for resource in resources:
+ try:
+ resource_value = json.loads(resource[1])
+ self.__settings_handler.set(resource[0], resource_value)
+ results.append(True)
+ except Exception as e: # pylint: disable=broad-except
+ LOGGER.exception('Unable to SetConfig({:s})'.format(str(resource)))
+ results.append(e)
+
+ return results
+
+ def DeleteConfig(self, resources : List[Tuple[str, Any]]) -> List[Union[bool, Exception]]:
+ chk_type('resources', resources, list)
+ if len(resources) == 0: return []
+
+ results = []
+ for resource in resources:
+ try:
+ self.__settings_handler.delete(resource[0])
+ except Exception as e: # pylint: disable=broad-except
+ LOGGER.exception('Unable to DeleteConfig({:s})'.format(str(resource)))
+ results.append(e)
+
+ return results
diff --git a/src/service/tests/test_l3nm_gnmi_static_rule_gen/MockTaskExecutor.py b/src/service/tests/test_l3nm_gnmi_static_rule_gen/MockTaskExecutor.py
new file mode 100644
index 0000000000000000000000000000000000000000..765b04477efdf06bfef934e96329887e898aa1b4
--- /dev/null
+++ b/src/service/tests/test_l3nm_gnmi_static_rule_gen/MockTaskExecutor.py
@@ -0,0 +1,57 @@
+# Copyright 2022-2024 ETSI OSG/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 logging
+from enum import Enum
+from typing import Dict, Optional, Union
+from common.method_wrappers.ServiceExceptions import NotFoundException
+from common.proto.context_pb2 import Connection, Device, DeviceId, Service
+from service.service.tools.ObjectKeys import get_device_key
+
+LOGGER = logging.getLogger(__name__)
+
+CacheableObject = Union[Connection, Device, Service]
+
+class CacheableObjectType(Enum):
+ CONNECTION = 'connection'
+ DEVICE = 'device'
+ SERVICE = 'service'
+
+class MockTaskExecutor:
+ def __init__(self) -> None:
+ self._grpc_objects_cache : Dict[str, CacheableObject] = dict()
+
+ # ----- Common methods ---------------------------------------------------------------------------------------------
+
+ def _load_grpc_object(self, object_type : CacheableObjectType, object_key : str) -> Optional[CacheableObject]:
+ object_key = '{:s}:{:s}'.format(object_type.value, object_key)
+ return self._grpc_objects_cache.get(object_key)
+
+ def _store_grpc_object(self, object_type : CacheableObjectType, object_key : str, grpc_object) -> None:
+ object_key = '{:s}:{:s}'.format(object_type.value, object_key)
+ self._grpc_objects_cache[object_key] = grpc_object
+
+ def _delete_grpc_object(self, object_type : CacheableObjectType, object_key : str) -> None:
+ object_key = '{:s}:{:s}'.format(object_type.value, object_key)
+ self._grpc_objects_cache.pop(object_key, None)
+
+ def get_device(self, device_id : DeviceId) -> Device:
+ device_key = get_device_key(device_id)
+ device = self._load_grpc_object(CacheableObjectType.DEVICE, device_key)
+ if device is None: raise NotFoundException('Device', device_key)
+ return device
+
+ def configure_device(self, device : Device) -> None:
+ device_key = get_device_key(device.device_id)
+ self._store_grpc_object(CacheableObjectType.DEVICE, device_key, device)
diff --git a/src/service/tests/test_l3nm_gnmi_static_rule_gen/__init__.py b/src/service/tests/test_l3nm_gnmi_static_rule_gen/__init__.py
new file mode 100644
index 0000000000000000000000000000000000000000..3ee6f7071f145e06c3aeaefc09a43ccd88e619e3
--- /dev/null
+++ b/src/service/tests/test_l3nm_gnmi_static_rule_gen/__init__.py
@@ -0,0 +1,14 @@
+# Copyright 2022-2024 ETSI OSG/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.
+
diff --git a/src/service/tests/test_l3nm_gnmi_static_rule_gen/test_unitary.py b/src/service/tests/test_l3nm_gnmi_static_rule_gen/test_unitary.py
new file mode 100644
index 0000000000000000000000000000000000000000..43709b036b8158ddfc59453aa798fa2d303906e0
--- /dev/null
+++ b/src/service/tests/test_l3nm_gnmi_static_rule_gen/test_unitary.py
@@ -0,0 +1,147 @@
+# Copyright 2022-2024 ETSI OSG/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.
+
+# Run with:
+# $ PYTHONPATH=./src python -m service.tests.test_l3nm_gnmi_static_rule_gen.test_unitary
+
+import logging
+from typing import List, Optional, Tuple
+from common.DeviceTypes import DeviceTypeEnum
+from common.proto.context_pb2 import Device, DeviceOperationalStatusEnum, Service
+from common.tools.object_factory.ConfigRule import json_config_rule_set
+from common.tools.object_factory.Device import json_device, json_device_id
+from common.tools.object_factory.EndPoint import json_endpoint, json_endpoint_id
+from common.tools.object_factory.Service import json_service_l3nm_planned
+from .MockServiceHandler import MockServiceHandler
+from .MockTaskExecutor import CacheableObjectType, MockTaskExecutor
+
+logging.basicConfig(level=logging.DEBUG)
+LOGGER = logging.getLogger(__name__)
+
+SERVICE_DC1_DC2 = Service(**json_service_l3nm_planned(
+ 'svc-dc1-dc2-uuid',
+ endpoint_ids=[
+ json_endpoint_id(json_device_id('DC1'), 'int'),
+ json_endpoint_id(json_device_id('DC2'), 'int'),
+ ],
+ config_rules=[
+ json_config_rule_set('/device[DC1]/endpoint[eth0]/settings', {
+ 'ipv4_address': '192.168.10.10', 'ipv4_prefix_len': 24, 'sub_interface_index': 0
+ }),
+ json_config_rule_set('/device[R1]/endpoint[1/2]/settings', {
+ 'ipv4_address': '10.0.1.1', 'ipv4_prefix_len': 24, 'sub_interface_index': 0
+ }),
+ #json_config_rule_set('/device[R2]/endpoint[1/2]/settings', {
+ # 'ipv4_address': '10.0.2.1', 'ipv4_prefix_len': 24, 'sub_interface_index': 0
+ #}),
+ json_config_rule_set('/device[DC2]/endpoint[eth0]/settings', {
+ 'ipv4_address': '192.168.20.10', 'ipv4_prefix_len': 24, 'sub_interface_index': 0
+ }),
+ ]
+))
+
+SERVICE_DC1_DC3 = Service(**json_service_l3nm_planned(
+ 'svc-dc1-dc3-uuid',
+ endpoint_ids=[
+ json_endpoint_id(json_device_id('DC1'), 'int'),
+ json_endpoint_id(json_device_id('DC3'), 'int'),
+ ],
+ config_rules=[
+ json_config_rule_set('/device[DC1]/endpoint[eth0]/settings', {
+ 'ipv4_address': '192.168.10.10', 'ipv4_prefix_len': 24, 'sub_interface_index': 0
+ }),
+ #json_config_rule_set('/device[R1]/endpoint[1/2]/settings', {
+ # 'ipv4_address': '10.0.1.1', 'ipv4_prefix_len': 24, 'sub_interface_index': 0
+ #}),
+ json_config_rule_set('/device[R4]/endpoint[1/1]/settings', {
+ 'ipv4_address': '10.0.4.1', 'ipv4_prefix_len': 24, 'sub_interface_index': 0
+ }),
+ json_config_rule_set('/device[DC3]/endpoint[eth0]/settings', {
+ 'ipv4_address': '192.168.30.10', 'ipv4_prefix_len': 24, 'sub_interface_index': 0
+ }),
+ ]
+))
+
+CONNECTION_ENDPOINTS_DC1_DC2 : List[Tuple[str, str, Optional[str]]] = [
+ ('DC1', 'int', None), ('DC1', 'eth0', None),
+ ('R1', '1/1', None), ('R1', '1/2', None),
+ ('R2', '1/1', None), ('R2', '1/2', None),
+ ('R3', '1/1', None), ('R3', '1/2', None),
+ ('DC2', 'eth0', None), ('DC2', 'int', None),
+]
+
+CONNECTION_ENDPOINTS_DC1_DC3 : List[Tuple[str, str, Optional[str]]] = [
+ ('DC1', 'int', None), ('DC1', 'eth0', None),
+ ('R1', '1/1', None), ('R1', '1/2', None),
+ ('R2', '1/1', None), ('R2', '1/3', None),
+ ('R4', '1/1', None), ('R4', '1/2', None),
+ ('DC3', 'eth0', None), ('DC3', 'int', None),
+]
+
+def test_l3nm_gnmi_static_rule_gen() -> None:
+ dev_op_st_enabled = DeviceOperationalStatusEnum.DEVICEOPERATIONALSTATUS_ENABLED
+
+ mock_task_executor = MockTaskExecutor()
+ mock_task_executor._store_grpc_object(CacheableObjectType.DEVICE, 'DC1', Device(**json_device(
+ 'uuid-DC1', DeviceTypeEnum.EMULATED_DATACENTER.value, dev_op_st_enabled, name='DC1', endpoints=[
+ json_endpoint(json_device_id('uuid-DC1'), 'uuid-int', 'packet', name='int' ),
+ json_endpoint(json_device_id('uuid-DC1'), 'uuid-eth0', 'packet', name='eth0'),
+ ]
+ )))
+ mock_task_executor._store_grpc_object(CacheableObjectType.DEVICE, 'DC2', Device(**json_device(
+ 'uuid-DC2', DeviceTypeEnum.EMULATED_DATACENTER.value, dev_op_st_enabled, name='DC2', endpoints=[
+ json_endpoint(json_device_id('uuid-DC2'), 'uuid-int', 'packet', name='int' ),
+ json_endpoint(json_device_id('uuid-DC2'), 'uuid-eth0', 'packet', name='eth0'),
+ ]
+ )))
+ mock_task_executor._store_grpc_object(CacheableObjectType.DEVICE, 'DC3', Device(**json_device(
+ 'uuid-DC3', DeviceTypeEnum.EMULATED_DATACENTER.value, dev_op_st_enabled, name='DC3', endpoints=[
+ json_endpoint(json_device_id('uuid-DC3'), 'uuid-int', 'packet', name='int' ),
+ json_endpoint(json_device_id('uuid-DC3'), 'uuid-eth0', 'packet', name='eth0'),
+ ]
+ )))
+ mock_task_executor._store_grpc_object(CacheableObjectType.DEVICE, 'R1', Device(**json_device(
+ 'uuid-R1', DeviceTypeEnum.EMULATED_PACKET_ROUTER.value, dev_op_st_enabled, name='R1', endpoints=[
+ json_endpoint(json_device_id('uuid-R1'), 'uuid-1/1', 'packet', name='1/1'),
+ json_endpoint(json_device_id('uuid-R1'), 'uuid-1/2', 'packet', name='1/2'),
+ ]
+ )))
+ mock_task_executor._store_grpc_object(CacheableObjectType.DEVICE, 'R2', Device(**json_device(
+ 'uuid-R2', DeviceTypeEnum.EMULATED_PACKET_ROUTER.value, dev_op_st_enabled, name='R2', endpoints=[
+ json_endpoint(json_device_id('uuid-R2'), 'uuid-1/1', 'packet', name='1/1'),
+ json_endpoint(json_device_id('uuid-R2'), 'uuid-1/2', 'packet', name='1/2'),
+ json_endpoint(json_device_id('uuid-R2'), 'uuid-1/3', 'packet', name='1/3'),
+ ]
+ )))
+ mock_task_executor._store_grpc_object(CacheableObjectType.DEVICE, 'R3', Device(**json_device(
+ 'uuid-R3', DeviceTypeEnum.EMULATED_PACKET_ROUTER.value, dev_op_st_enabled, name='R3', endpoints=[
+ json_endpoint(json_device_id('uuid-R3'), 'uuid-1/1', 'packet', name='1/1'),
+ json_endpoint(json_device_id('uuid-R3'), 'uuid-1/2', 'packet', name='1/2'),
+ ]
+ )))
+ mock_task_executor._store_grpc_object(CacheableObjectType.DEVICE, 'R4', Device(**json_device(
+ 'uuid-R4', DeviceTypeEnum.EMULATED_PACKET_ROUTER.value, dev_op_st_enabled, name='R4', endpoints=[
+ json_endpoint(json_device_id('uuid-R4'), 'uuid-1/1', 'packet', name='1/1'),
+ json_endpoint(json_device_id('uuid-R4'), 'uuid-1/2', 'packet', name='1/2'),
+ ]
+ )))
+
+ mock_service_handler = MockServiceHandler(SERVICE_DC1_DC2, mock_task_executor)
+ mock_service_handler.SetEndpoint(CONNECTION_ENDPOINTS_DC1_DC2)
+
+ mock_service_handler = MockServiceHandler(SERVICE_DC1_DC3, mock_task_executor)
+ mock_service_handler.SetEndpoint(CONNECTION_ENDPOINTS_DC1_DC3)
+
+if __name__ == '__main__':
+ test_l3nm_gnmi_static_rule_gen()