Service component - L3NM gNMI OpenConfig Service Handler:

- Added StaticRouteGenerator class
- ConfigRuleComposer: added dump methods to composer classes
- ConfigRuleComposer: extended configure methods to load previous configurations in devices
- ConfigRuleComposer: generalized to a single network instance for all static routing-based services
- ConfigRuleComposer: Updated static route organization with indexes and metrics
- ConfigRuleComposer: Added tracking of connected networks
- ConfigRuleComposer: Corrected config rule composer methods
- Integrated StaticRouteGenerator into Service Handler
- Multiple cosmetic improvements
- Added code for a future unitary test

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()
+            }
+        }

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)

src/service/service/service_handlers/l3nm_gnmi_openconfig/StaticRouteGenerator.py

+# 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)

src/service/tests/test_l3nm_gnmi_static_rule_gen/MockServiceHandler.py

+# 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

src/service/tests/test_l3nm_gnmi_static_rule_gen/MockTaskExecutor.py

+# 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)

src/service/tests/test_l3nm_gnmi_static_rule_gen/__init__.py

+# 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.
+

src/service/tests/test_l3nm_gnmi_static_rule_gen/test_unitary.py

+# 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()