src/device/tests/gnmi_openconfig/storage/StorageEndpoints.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 re
+from typing import Dict, List, Tuple
+from .Tools import compose_resources
+
+RE_RESKEY_ENDPOINT = re.compile(r'^\/endpoints\/endpoint\[([^\]]+)\]$')
+
+ENDPOINT_PACKET_SAMPLE_TYPES : Dict[int, str] = {
+    101: '/openconfig-interfaces:interfaces/interface[name={:s}]/state/counters/out-pkts',
+    102: '/openconfig-interfaces:interfaces/interface[name={:s}]/state/counters/in-pkts',
+    201: '/openconfig-interfaces:interfaces/interface[name={:s}]/state/counters/out-octets',
+    202: '/openconfig-interfaces:interfaces/interface[name={:s}]/state/counters/in-octets',
+}
+
+class Endpoints:
+    STRUCT : List[Tuple[str, List[str]]] = [
+        ('/endpoints/endpoint[{:s}]', ['uuid', 'type', 'sample_types']),
+    ]
+
+    def __init__(self) -> None:
+        self._items : Dict[str, Dict] = dict()
+
+    def add(self, ep_uuid : str, resource_value : Dict) -> None:
+        item = self._items.setdefault(ep_uuid, dict())
+        item['uuid'] = ep_uuid
+
+        for _, field_names in Endpoints.STRUCT:
+            field_names = set(field_names)
+            item.update({k:v for k,v in resource_value.items() if k in field_names})
+
+        item['sample_types'] = {
+            sample_type_id : sample_type_path.format(ep_uuid)
+            for sample_type_id, sample_type_path in ENDPOINT_PACKET_SAMPLE_TYPES.items()
+        }
+
+    def get(self, ep_uuid : str) -> Dict:
+        return self._items.get(ep_uuid)
+
+    def remove(self, ep_uuid : str) -> None:
+        self._items.pop(ep_uuid, None)
+    
+    def compose_resources(self) -> List[Dict]:
+        return compose_resources(self._items, Endpoints.STRUCT)
+
+class StorageEndpoints:
+    def __init__(self) -> None:
+        self.endpoints = Endpoints()
+
+    def populate(self, resources : List[Tuple[str, Dict]]) -> None:
+        for resource_key, resource_value in resources:
+            match = RE_RESKEY_ENDPOINT.match(resource_key)
+            if match is not None:
+                self.endpoints.add(match.group(1), resource_value)
+                continue
+
+            MSG = 'Unhandled Resource Key: {:s} => {:s}'
+            raise Exception(MSG.format(str(resource_key), str(resource_value)))
+
+    def get_expected_config(self) -> List[Tuple[str, Dict]]:
+        expected_config = list()
+        expected_config.extend(self.endpoints.compose_resources())
+        return expected_config

src/device/tests/gnmi_openconfig/storage/StorageInterface copy.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 re
+from typing import Dict, List, Tuple
+from .Tools import compose_resources
+
+PREFIX = r'^\/interface\[([^\]]+)\]'
+RE_RESKEY_INTERFACE    = re.compile(PREFIX + r'$')
+RE_RESKEY_ETHERNET     = re.compile(PREFIX + r'\/ethernet$')
+RE_RESKEY_SUBINTERFACE = re.compile(PREFIX + r'\/subinterface\[([^\]]+)\]$')
+#RE_RESKEY_IPV4_ADDRESS = re.compile(PREFIX + r'\/subinterface\[([^\]]+)\]\/ipv4\[([^\]]+)\]$')
+
+class Interfaces:
+    STRUCT : List[Tuple[str, List[str]]] = [
+        ('/interface[{:s}]', ['name', 'type', 'admin-status', 'oper-status', 'management', 'mtu', 'ifindex',
+                              'hardware-port', 'transceiver']),
+        ('/interface[{:s}]/ethernet', ['port-speed', 'negotiated-port-speed', 'mac-address', 'hw-mac-address']),
+    ]
+
+    def __init__(self) -> None:
+        self._items : Dict[str, Dict] = dict()
+
+    def add(self, if_name : str, resource_value : Dict) -> None:
+        item = self._items.setdefault(if_name, dict())
+        item['name'] = if_name
+        for _, field_names in Interfaces.STRUCT:
+            field_names = set(field_names)
+            item.update({k:v for k,v in resource_value.items() if k in field_names})
+
+    def get(self, if_name : str) -> Dict:
+        return self._items.get(if_name)
+
+    def remove(self, if_name : str) -> None:
+        self._items.pop(if_name, None)
+    
+    def compose_resources(self) -> List[Dict]:
+        return compose_resources(self._items, Interfaces.STRUCT)
+
+class SubInterfaces:
+    STRUCT : List[Tuple[str, List[str]]] = [
+        ('/interface[{:s}]/subinterface[{:d}]', ['index']),
+    ]
+
+    def __init__(self) -> None:
+        self._items : Dict[Tuple[str, int], Dict] = dict()
+
+    def add(self, if_name : str, subif_index : int) -> None:
+        item = self._items.setdefault((if_name, subif_index), dict())
+        item['index'] = subif_index
+
+    def get(self, if_name : str, subif_index : int) -> Dict:
+        return self._items.get((if_name, subif_index))
+
+    def remove(self, if_name : str, subif_index : int) -> None:
+        self._items.pop((if_name, subif_index), None)
+    
+    def compose_resources(self) -> List[Dict]:
+        return compose_resources(self._items, SubInterfaces.STRUCT)
+
+class IPv4Addresses:
+    STRUCT : List[Tuple[str, List[str]]] = [
+        ('/interface[{:s}]/subinterface[{:d}]', ['index', 'address_ip', 'address_prefix', 'origin']),
+    ]
+
+    def __init__(self) -> None:
+        self._items : Dict[Tuple[str, int], Dict] = dict()
+
+    def add(self, if_name : str, subif_index : int, ipv4_address : str, resource_value : Dict) -> None:
+        item = self._items.setdefault((if_name, subif_index), dict())
+        item['index'         ] = subif_index
+        item['address_ip'    ] = ipv4_address
+        item['origin'        ] = resource_value.get('origin')
+        item['address_prefix'] = resource_value.get('prefix')
+
+    def get(self, if_name : str, subif_index : int, ipv4_address : str) -> Dict:
+        return self._items.get((if_name, subif_index))
+
+    def remove(self, if_name : str, subif_index : int, ipv4_address : str) -> None:
+        self._items.pop((if_name, subif_index), None)
+
+    def compose_resources(self) -> List[Dict]:
+        return compose_resources(self._items, IPv4Addresses.STRUCT)
+
+class StorageInterface:
+    def __init__(self) -> None:
+        self.interfaces     = Interfaces()
+        self.subinterfaces  = SubInterfaces()
+        self.ipv4_addresses = IPv4Addresses()
+
+    def populate(self, resources : List[Tuple[str, Dict]]) -> None:
+        for resource_key, resource_value in resources:
+            match = RE_RESKEY_INTERFACE.match(resource_key)
+            if match is not None:
+                self.interfaces.add(match.group(1), resource_value)
+                continue
+
+            match = RE_RESKEY_ETHERNET.match(resource_key)
+            if match is not None:
+                self.interfaces.add(match.group(1), resource_value)
+                continue
+
+            match = RE_RESKEY_SUBINTERFACE.match(resource_key)
+            if match is not None:
+                self.subinterfaces.add(match.group(1), int(match.group(2)))
+                address_ip = resource_value.get('address_ip')
+                self.ipv4_addresses.add(match.group(1), int(match.group(2)), address_ip, resource_value)
+                continue
+
+            #match = RE_RESKEY_IPV4_ADDRESS.match(resource_key)
+            #if match is not None:
+            #    self.ipv4_addresses.add(match.group(1), int(match.group(2)), match.group(3), resource_value)
+            #    continue
+
+            MSG = 'Unhandled Resource Key: {:s} => {:s}'
+            raise Exception(MSG.format(str(resource_key), str(resource_value)))
+
+    def get_expected_config(self) -> List[Tuple[str, Dict]]:
+        expected_config = list()
+        expected_config.extend(self.interfaces.compose_resources())
+        #expected_config.extend(self.subinterfaces.compose_resources())
+        expected_config.extend(self.ipv4_addresses.compose_resources())
+        return expected_config

src/device/tests/gnmi_openconfig/storage/StorageInterface.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 re
+from typing import Dict, List, Tuple
+from .Tools import compose_resources
+
+PREFIX = r'^\/interface\[([^\]]+)\]'
+RE_RESKEY_INTERFACE    = re.compile(PREFIX + r'$')
+RE_RESKEY_ETHERNET     = re.compile(PREFIX + r'\/ethernet$')
+RE_RESKEY_SUBINTERFACE = re.compile(PREFIX + r'\/subinterface\[([^\]]+)\]$')
+RE_RESKEY_IPV4_ADDRESS = re.compile(PREFIX + r'\/subinterface\[([^\]]+)\]\/ipv4\[([^\]]+)\]$')
+
+class Interfaces:
+    STRUCT : List[Tuple[str, List[str]]] = [
+        ('/interface[{:s}]', ['name', 'type', 'admin-status', 'oper-status', 'management', 'mtu', 'ifindex',
+                              'hardware-port', 'transceiver']),
+        ('/interface[{:s}]/ethernet', ['port-speed', 'negotiated-port-speed', 'mac-address', 'hw-mac-address']),
+    ]
+
+    def __init__(self) -> None:
+        self._items : Dict[str, Dict] = dict()
+
+    def add(self, if_name : str, resource_value : Dict) -> None:
+        item = self._items.setdefault(if_name, dict())
+        item['name'] = if_name
+        for _, field_names in Interfaces.STRUCT:
+            field_names = set(field_names)
+            item.update({k:v for k,v in resource_value.items() if k in field_names})
+
+    def get(self, if_name : str) -> Dict:
+        return self._items.get(if_name)
+
+    def remove(self, if_name : str) -> None:
+        self._items.pop(if_name, None)
+    
+    def compose_resources(self) -> List[Dict]:
+        return compose_resources(self._items, Interfaces.STRUCT)
+
+class SubInterfaces:
+    STRUCT : List[Tuple[str, List[str]]] = [
+        ('/interface[{:s}]/subinterface[{:d}]', ['index']),
+    ]
+
+    def __init__(self) -> None:
+        self._items : Dict[Tuple[str, int], Dict] = dict()
+
+    def add(self, if_name : str, subif_index : int) -> None:
+        item = self._items.setdefault((if_name, subif_index), dict())
+        item['index'] = subif_index
+
+    def get(self, if_name : str, subif_index : int) -> Dict:
+        return self._items.get((if_name, subif_index))
+
+    def remove(self, if_name : str, subif_index : int) -> None:
+        self._items.pop((if_name, subif_index), None)
+    
+    def compose_resources(self) -> List[Dict]:
+        return compose_resources(self._items, SubInterfaces.STRUCT)
+
+class IPv4Addresses:
+    STRUCT : List[Tuple[str, List[str]]] = [
+        ('/interface[{:s}]/subinterface[{:d}]/ipv4[{:s}]', ['ip', 'origin', 'prefix']),
+    ]
+
+    def __init__(self) -> None:
+        self._items : Dict[Tuple[str, int, str], Dict] = dict()
+
+    def add(self, if_name : str, subif_index : int, ipv4_address : str, resource_value : Dict) -> None:
+        item = self._items.setdefault((if_name, subif_index, ipv4_address), dict())
+        item['ip'    ] = ipv4_address
+        item['origin'] = resource_value.get('origin')
+        item['prefix'] = resource_value.get('prefix')
+
+    def get(self, if_name : str, subif_index : int, ipv4_address : str) -> Dict:
+        return self._items.get((if_name, subif_index, ipv4_address))
+
+    def remove(self, if_name : str, subif_index : int, ipv4_address : str) -> None:
+        self._items.pop((if_name, subif_index, ipv4_address), None)
+
+    def compose_resources(self) -> List[Dict]:
+        return compose_resources(self._items, IPv4Addresses.STRUCT)
+
+class StorageInterface:
+    def __init__(self) -> None:
+        self.interfaces     = Interfaces()
+        self.subinterfaces  = SubInterfaces()
+        self.ipv4_addresses = IPv4Addresses()
+
+    def populate(self, resources : List[Tuple[str, Dict]]) -> None:
+        for resource_key, resource_value in resources:
+            match = RE_RESKEY_INTERFACE.match(resource_key)
+            if match is not None:
+                self.interfaces.add(match.group(1), resource_value)
+                continue
+
+            match = RE_RESKEY_ETHERNET.match(resource_key)
+            if match is not None:
+                self.interfaces.add(match.group(1), resource_value)
+                continue
+
+            match = RE_RESKEY_SUBINTERFACE.match(resource_key)
+            if match is not None:
+                self.subinterfaces.add(match.group(1), int(match.group(2)))
+                continue
+
+            match = RE_RESKEY_IPV4_ADDRESS.match(resource_key)
+            if match is not None:
+                self.ipv4_addresses.add(match.group(1), int(match.group(2)), match.group(3), resource_value)
+                continue
+
+            MSG = 'Unhandled Resource Key: {:s} => {:s}'
+            raise Exception(MSG.format(str(resource_key), str(resource_value)))
+
+    def get_expected_config(self) -> List[Tuple[str, Dict]]:
+        expected_config = list()
+        expected_config.extend(self.interfaces.compose_resources())
+        expected_config.extend(self.subinterfaces.compose_resources())
+        expected_config.extend(self.ipv4_addresses.compose_resources())
+        return expected_config

src/device/tests/gnmi_openconfig/storage/StorageNetworkInstance.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 re
+from typing import Dict, List, Tuple
+from .Tools import compose_resources
+
+PREFIX = r'^\/network\_instance\[([^\]]+)\]'
+RE_RESKEY_NET_INST     = re.compile(PREFIX + r'$')
+RE_RESKEY_INTERFACE    = re.compile(PREFIX + r'\/interface\[([^\]]+)\]$')
+RE_RESKEY_PROTOCOL     = re.compile(PREFIX + r'\/protocol\[([^\]]+)\]$')
+RE_RESKEY_PROTO_STATIC = re.compile(PREFIX + r'\/protocol\[([^\]]+)\]\/static\_routes\[([^\]]+)\]$')
+RE_RESKEY_TABLE        = re.compile(PREFIX + r'\/table\[([^\,]+)\,([^\]]+)\]$')
+RE_RESKEY_VLAN         = re.compile(PREFIX + r'\/vlan\[([^\]]+)\]$')
+
+class NetworkInstances:
+    STRUCT : List[Tuple[str, List[str]]] = [
+        ('/network_instance[{:s}]', ['name', 'type']),
+    ]
+
+    def __init__(self) -> None:
+        self._items : Dict[str, Dict] = dict()
+
+    def add(self, ni_name : str, resource_value : Dict) -> None:
+        item = self._items.setdefault(ni_name, dict())
+        item['name'] = ni_name
+        item['type'] = resource_value.get('type')
+
+    def get(self, ni_name : str) -> Dict:
+        return self._items.get(ni_name)
+
+    def remove(self, ni_name : str) -> None:
+        self._items.pop(ni_name, None)
+    
+    def compose_resources(self) -> List[Dict]:
+        return compose_resources(self._items, NetworkInstances.STRUCT)
+
+class Interfaces:
+    STRUCT : List[Tuple[str, List[str]]] = [
+        ('/network_instance[{:s}]/interface[{:s}.{:d}]', ['name', 'id', 'if_name', 'sif_index']),
+    ]
+
+    def __init__(self) -> None:
+        self._items : Dict[Tuple[str, str], Dict] = dict()
+
+    def add(self, ni_name : str, if_name : str, sif_index : int) -> None:
+        item = self._items.setdefault((ni_name, if_name, sif_index), dict())
+        item['name'     ] = ni_name
+        item['id'       ] = '{:s}.{:d}'.format(if_name, sif_index)
+        item['if_name'  ] = if_name
+        item['sif_index'] = sif_index
+
+    def get(self, ni_name : str, if_name : str, sif_index : int) -> Dict:
+        return self._items.get((ni_name, if_name, sif_index))
+
+    def remove(self, ni_name : str, if_name : str, sif_index : int) -> None:
+        self._items.pop((ni_name, if_name, sif_index), None)
+
+    def compose_resources(self) -> List[Dict]:
+        return compose_resources(self._items, Interfaces.STRUCT)
+
+class Protocols:
+    STRUCT : List[Tuple[str, List[str]]] = [
+        ('/network_instance[{:s}]/protocol[{:s}]', ['id', 'name']),
+    ]
+
+    def __init__(self) -> None:
+        self._items : Dict[Tuple[str, str], Dict] = dict()
+
+    def add(self, ni_name : str, protocol : str) -> None:
+        item = self._items.setdefault((ni_name, protocol), dict())
+        item['id'  ] = protocol
+        item['name'] = protocol
+
+    def get(self, ni_name : str, protocol : str) -> Dict:
+        return self._items.get((ni_name, protocol))
+
+    def remove(self, ni_name : str, protocol : str) -> None:
+        self._items.pop((ni_name, protocol), None)
+
+    def compose_resources(self) -> List[Dict]:
+        return compose_resources(self._items, Protocols.STRUCT)
+
+class StaticRoutes:
+    STRUCT : List[Tuple[str, List[str]]] = [
+        ('/network_instance[{:s}]/protocol[{:s}]/static_routes[{:s}]', ['prefix', 'next_hops']),
+    ]
+
+    def __init__(self) -> None:
+        self._items : Dict[Tuple[str, str, str], Dict] = dict()
+
+    def add(self, ni_name : str, protocol : str, prefix : str, resource_value : Dict) -> None:
+        item = self._items.setdefault((ni_name, protocol, prefix), dict())
+        item['prefix'   ] = prefix
+        item['next_hops'] = resource_value.get('next_hops')
+
+    def get(self, ni_name : str, protocol : str, prefix : str) -> Dict:
+        return self._items.get((ni_name, protocol, prefix))
+
+    def remove(self, ni_name : str, protocol : str, prefix : str) -> None:
+        self._items.pop((ni_name, protocol, prefix), None)
+
+    def compose_resources(self) -> List[Dict]:
+        return compose_resources(self._items, StaticRoutes.STRUCT)
+
+class Tables:
+    STRUCT : List[Tuple[str, List[str]]] = [
+        ('/network_instance[{:s}]/table[{:s},{:s}]', ['protocol', 'address_family']),
+    ]
+
+    def __init__(self) -> None:
+        self._items : Dict[Tuple[str, str, str], Dict] = dict()
+
+    def add(self, ni_name : str, protocol : str, address_family : str) -> None:
+        item = self._items.setdefault((ni_name, protocol, address_family), dict())
+        item['protocol'      ] = protocol
+        item['address_family'] = address_family
+
+    def get(self, ni_name : str, protocol : str, address_family : str) -> Dict:
+        return self._items.get((ni_name, protocol, address_family))
+
+    def remove(self, ni_name : str, protocol : str, address_family : str) -> None:
+        self._items.pop((ni_name, protocol, address_family), None)
+
+    def compose_resources(self) -> List[Dict]:
+        return compose_resources(self._items, Tables.STRUCT)
+
+class Vlans:
+    STRUCT : List[Tuple[str, List[str]]] = [
+        ('/network_instance[{:s}]/vlan[{:d}]', ['vlan_id', 'name', 'members']),
+    ]
+
+    def __init__(self) -> None:
+        self._items : Dict[Tuple[str, int], Dict] = dict()
+
+    def add(self, ni_name : str, vlan_id : int, resource_value : Dict) -> None:
+        item = self._items.setdefault((ni_name, vlan_id), dict())
+        item['vlan_id'] = vlan_id
+        item['name'   ] = resource_value.get('name')
+        item['members'] = sorted(resource_value.get('members'))
+
+    def get(self, ni_name : str, vlan_id : int) -> Dict:
+        return self._items.get((ni_name, vlan_id))
+
+    def remove(self, ni_name : str, vlan_id : int) -> None:
+        self._items.pop((ni_name, vlan_id), None)
+
+    def compose_resources(self) -> List[Dict]:
+        return compose_resources(self._items, Vlans.STRUCT)
+
+class StorageNetworkInstance:
+    def __init__(self) -> None:
+        self.network_instances = NetworkInstances()
+        self.interfaces        = Interfaces()
+        self.protocols         = Protocols()
+        self.protocol_static   = StaticRoutes()
+        self.tables            = Tables()
+        self.vlans             = Vlans()
+
+    def populate(self, resources : List[Tuple[str, Dict]]) -> None:
+        for resource_key, resource_value in resources:
+            match = RE_RESKEY_NET_INST.match(resource_key)
+            if match is not None:
+                self.network_instances.add(match.group(1), resource_value)
+                continue
+
+            match = RE_RESKEY_INTERFACE.match(resource_key)
+            if match is not None:
+                if_id = match.group(2)
+                if_id_parts = if_id.split('.')
+                if_name = if_id_parts[0]
+                sif_index = 0 if len(if_id_parts) == 1 else int(if_id_parts[1])
+                self.interfaces.add(match.group(1), if_name, sif_index)
+                continue
+
+            match = RE_RESKEY_PROTOCOL.match(resource_key)
+            if match is not None:
+                self.protocols.add(match.group(1), match.group(2))
+                continue
+
+            match = RE_RESKEY_PROTO_STATIC.match(resource_key)
+            if match is not None:
+                self.protocol_static.add(match.group(1), match.group(2), match.group(3), resource_value)
+                continue
+
+            match = RE_RESKEY_TABLE.match(resource_key)
+            if match is not None:
+                self.tables.add(match.group(1), match.group(2), match.group(3))
+                continue
+
+            match = RE_RESKEY_VLAN.match(resource_key)
+            if match is not None:
+                self.vlans.add(match.group(1), int(match.group(2)), resource_value)
+                continue
+
+            MSG = 'Unhandled Resource Key: {:s} => {:s}'
+            raise Exception(MSG.format(str(resource_key), str(resource_value)))
+
+    def get_expected_config(self) -> List[Tuple[str, Dict]]:
+        expected_config = list()
+        expected_config.extend(self.network_instances.compose_resources())
+        expected_config.extend(self.interfaces.compose_resources())
+        expected_config.extend(self.protocols.compose_resources())
+        expected_config.extend(self.protocol_static.compose_resources())
+        expected_config.extend(self.tables.compose_resources())
+        expected_config.extend(self.vlans.compose_resources())
+        return expected_config

src/device/tests/gnmi_openconfig/storage/Tools.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.
+
+from typing import Dict, List, Tuple
+
+def compose_resources(
+    storage : Dict[Tuple, Dict], config_struct : List[Tuple[str, List[str]]]
+) -> List[Dict]:
+    expected_config = list()
+
+    for resource_key_fields, resource_value_data in storage.items():
+        for resource_key_template, resource_key_field_names in config_struct:
+            if isinstance(resource_key_fields, (str, int, float, bool)): resource_key_fields = (resource_key_fields,)
+            resource_key = resource_key_template.format(*resource_key_fields)
+            resource_value = {
+                field_name : resource_value_data[field_name]
+                for field_name in resource_key_field_names
+                if field_name in resource_value_data and resource_value_data[field_name] is not None
+            }
+            expected_config.append((resource_key, resource_value))
+
+    return expected_config

src/device/tests/gnmi_openconfig/storage/__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/device/tests/gnmi_openconfig/test_unitary_gnmi_openconfig.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 os
+os.environ['DEVICE_EMULATED_ONLY'] = 'YES'
+
+# pylint: disable=wrong-import-position
+import logging, pytest, time
+from typing import Dict, List
+from device.service.driver_api._Driver import (
+    RESOURCE_ENDPOINTS, RESOURCE_INTERFACES, RESOURCE_NETWORK_INSTANCES,
+    RESOURCE_ROUTING_POLICIES, RESOURCE_SERVICES
+)
+from device.service.drivers.gnmi_openconfig.GnmiOpenConfigDriver import GnmiOpenConfigDriver
+from .storage.Storage import Storage
+from .tools.manage_config import (
+    check_config_endpoints, check_config_interfaces, check_config_network_instances, del_config, get_config, set_config
+)
+from .tools.check_updates import check_updates
+from .tools.request_composers import (
+    interface, network_instance, network_instance_interface, network_instance_static_route
+)
+
+logging.basicConfig(level=logging.DEBUG)
+LOGGER = logging.getLogger(__name__)
+LOGGER.setLevel(logging.DEBUG)
+
+
+##### DRIVER FIXTURE ###################################################################################################
+
+DRIVER_SETTING_ADDRESS  = '172.20.20.101'
+DRIVER_SETTING_PORT     = 6030
+DRIVER_SETTING_USERNAME = 'admin'
+DRIVER_SETTING_PASSWORD = 'admin'
+DRIVER_SETTING_USE_TLS  = False
+
+@pytest.fixture(scope='session')
+def driver() -> GnmiOpenConfigDriver:
+    _driver = GnmiOpenConfigDriver(
+        DRIVER_SETTING_ADDRESS, DRIVER_SETTING_PORT,
+        username=DRIVER_SETTING_USERNAME,
+        password=DRIVER_SETTING_PASSWORD,
+        use_tls=DRIVER_SETTING_USE_TLS,
+    )
+    _driver.Connect()
+    yield _driver
+    time.sleep(1)
+    _driver.Disconnect()
+
+
+##### STORAGE FIXTURE ##################################################################################################
+
+@pytest.fixture(scope='session')
+def storage() -> Dict:
+    yield Storage()
+
+
+##### NETWORK INSTANCE DETAILS #########################################################################################
+
+NETWORK_INSTANCES = [
+    {
+        'name': 'test-l3-svc',
+        'type': 'L3VRF',
+        'interfaces': [
+            {'name': 'Ethernet1',  'index': 0, 'ipv4_addr': '192.168.1.1',  'ipv4_prefix': 24, 'enabled': True},
+            {'name': 'Ethernet10', 'index': 0, 'ipv4_addr': '192.168.10.1', 'ipv4_prefix': 24, 'enabled': True},
+        ],
+        'static_routes': [
+            {'prefix': '172.0.0.0/24', 'next_hop': '172.16.0.2', 'metric': 1},
+            {'prefix': '172.2.0.0/24', 'next_hop': '172.16.0.3', 'metric': 1},
+        ]
+    },
+    #{
+    #    'name': 'test-l2-svc',
+    #    'type': 'L2VSI',
+    #    'interfaces': [
+    #        {'name': 'Ethernet2', 'index': 0, 'ipv4_addr': '192.168.1.1',  'ipv4_prefix': 24, 'enabled': True},
+    #        {'name': 'Ethernet4', 'index': 0, 'ipv4_addr': '192.168.10.1', 'ipv4_prefix': 24, 'enabled': True},
+    #    ],
+    #    'static_routes': [
+    #        {'prefix': '172.0.0.0/24', 'next_hop': '172.16.0.2', 'metric': 1},
+    #        {'prefix': '172.2.0.0/24', 'next_hop': '172.16.0.3', 'metric': 1},
+    #    ]
+    #}
+]
+
+
+##### TEST METHODS #####################################################################################################
+
+def test_get_endpoints(
+    driver : GnmiOpenConfigDriver,  # pylint: disable=redefined-outer-name
+    storage : Storage,              # pylint: disable=redefined-outer-name
+) -> None:
+    results_getconfig = get_config(driver, [RESOURCE_ENDPOINTS])
+    storage.endpoints.populate(results_getconfig)
+    check_config_endpoints(driver, storage)
+
+
+def test_get_interfaces(
+    driver : GnmiOpenConfigDriver,  # pylint: disable=redefined-outer-name
+    storage : Storage,              # pylint: disable=redefined-outer-name
+) -> None:
+    results_getconfig = get_config(driver, [RESOURCE_INTERFACES])
+    storage.interfaces.populate(results_getconfig)
+    check_config_interfaces(driver, storage)
+
+
+def test_get_network_instances(
+    driver : GnmiOpenConfigDriver,  # pylint: disable=redefined-outer-name
+    storage : Storage,              # pylint: disable=redefined-outer-name
+) -> None:
+    results_getconfig = get_config(driver, [RESOURCE_NETWORK_INSTANCES])
+    storage.network_instances.populate(results_getconfig)
+    check_config_network_instances(driver, storage)
+
+
+def test_set_network_instances(
+    driver : GnmiOpenConfigDriver,  # pylint: disable=redefined-outer-name
+    storage : Storage,              # pylint: disable=redefined-outer-name
+) -> None:
+    check_config_interfaces(driver, storage)
+    check_config_network_instances(driver, storage)
+
+    resources_to_set = list()
+    ni_names = list()
+    for ni in NETWORK_INSTANCES:
+        ni_name = ni['name']
+        ni_type = ni['type']
+        resources_to_set.append(network_instance(ni_name, ni_type))
+        ni_names.append(ni_name)
+        storage.network_instances.network_instances.add(ni_name, {'type': ni_type})
+        storage.network_instances.protocols.add(ni_name, 'DIRECTLY_CONNECTED')
+        storage.network_instances.tables.add(ni_name, 'DIRECTLY_CONNECTED', 'IPV4')
+        storage.network_instances.tables.add(ni_name, 'DIRECTLY_CONNECTED', 'IPV6')
+
+    results_setconfig = set_config(driver, resources_to_set)
+    check_updates(results_setconfig, '/network_instance[{:s}]', ni_names)
+
+    check_config_interfaces(driver, storage, max_retries=10, retry_delay=2.0)
+    check_config_network_instances(driver, storage, max_retries=10, retry_delay=2.0)
+
+
+def test_add_interfaces_to_network_instance(
+    driver : GnmiOpenConfigDriver,  # pylint: disable=redefined-outer-name
+    storage : Storage,              # pylint: disable=redefined-outer-name
+) -> None:
+    check_config_interfaces(driver, storage)
+    check_config_network_instances(driver, storage)
+
+    resources_to_set = list()
+    ni_if_names = list()
+    for ni in NETWORK_INSTANCES:
+        ni_name = ni['name']
+        for ni_if in ni.get('interfaces', list()):
+            if_name     = ni_if['name' ]
+            subif_index = ni_if['index']
+            resources_to_set.append(network_instance_interface(ni_name, if_name, subif_index))
+            ni_if_names.append((ni_name, '{:s}.{:d}'.format(if_name, subif_index)))
+            storage.network_instances.interfaces.add(ni_name, if_name, subif_index)
+
+    results_setconfig = set_config(driver, resources_to_set)
+    check_updates(results_setconfig, '/network_instance[{:s}]/interface[{:s}]', ni_if_names)
+
+    check_config_interfaces(driver, storage, max_retries=10, retry_delay=2.0)
+    check_config_network_instances(driver, storage, max_retries=10, retry_delay=2.0)
+
+
+def test_set_interfaces(
+    driver : GnmiOpenConfigDriver,  # pylint: disable=redefined-outer-name
+    storage : Storage,              # pylint: disable=redefined-outer-name
+) -> None:
+    check_config_interfaces(driver, storage)
+    check_config_network_instances(driver, storage)
+
+    resources_to_set = list()
+    if_names = list()
+    for ni in NETWORK_INSTANCES:
+        for ni_if in ni.get('interfaces', list()):
+            if_name      = ni_if['name'       ]
+            subif_index  = ni_if['index'      ]
+            ipv4_address = ni_if['ipv4_addr'  ]
+            ipv4_prefix  = ni_if['ipv4_prefix']
+            enabled      = ni_if['enabled'    ]
+            resources_to_set.append(interface(
+                if_name, subif_index, ipv4_address, ipv4_prefix, enabled
+            ))
+            if_names.append(if_name)
+            storage.interfaces.ipv4_addresses.add(if_name, subif_index, ipv4_address, {
+                'origin' : 'STATIC', 'prefix': ipv4_prefix
+            })
+            default_vlan = storage.network_instances.vlans.get('default', 1)
+            default_vlan_members : List[str] = default_vlan.setdefault('members', list())
+            if if_name in default_vlan_members: default_vlan_members.remove(if_name)
+
+    results_setconfig = set_config(driver, resources_to_set)
+    check_updates(results_setconfig, '/interface[{:s}]', if_names)
+
+    check_config_interfaces(driver, storage, max_retries=10, retry_delay=2.0)
+    check_config_network_instances(driver, storage, max_retries=10, retry_delay=2.0)
+
+
+def test_set_network_instance_static_routes(
+    driver : GnmiOpenConfigDriver,  # pylint: disable=redefined-outer-name
+    storage : Storage,              # pylint: disable=redefined-outer-name
+) -> None:
+    check_config_interfaces(driver, storage)
+    check_config_network_instances(driver, storage)
+
+    resources_to_set = list()
+    ni_sr_prefixes = list()
+    for ni in NETWORK_INSTANCES:
+        ni_name = ni['name']
+        for ni_sr in ni.get('static_routes', list()):
+            ni_sr_prefix   = ni_sr['prefix'  ]
+            ni_sr_next_hop = ni_sr['next_hop']
+            ni_sr_metric   = ni_sr['metric'  ]
+            ni_sr_next_hop_index = 'AUTO_{:d}_{:s}'.format(ni_sr_metric, '-'.join(ni_sr_next_hop.split('.')))
+            resources_to_set.append(network_instance_static_route(
+                ni_name, ni_sr_prefix, ni_sr_next_hop_index, ni_sr_next_hop, metric=ni_sr_metric
+            ))
+            ni_sr_prefixes.append((ni_name, ni_sr_prefix))
+            storage.network_instances.protocols.add(ni_name, 'STATIC')
+            storage.network_instances.protocol_static.add(ni_name, 'STATIC', ni_sr_prefix, {
+                'prefix': ni_sr_prefix, 'next_hops': {
+                    ni_sr_next_hop_index: {'next_hop': ni_sr_next_hop, 'metric': ni_sr_metric}
+                }
+            })
+            storage.network_instances.tables.add(ni_name, 'STATIC', 'IPV4')
+            storage.network_instances.tables.add(ni_name, 'STATIC', 'IPV6')
+
+    results_setconfig = set_config(driver, resources_to_set)
+    check_updates(results_setconfig, '/network_instance[{:s}]/static_route[{:s}]', ni_sr_prefixes)
+
+    check_config_interfaces(driver, storage, max_retries=10, retry_delay=2.0)
+    check_config_network_instances(driver, storage, max_retries=10, retry_delay=2.0)
+
+
+def test_del_network_instance_static_routes(
+    driver : GnmiOpenConfigDriver,  # pylint: disable=redefined-outer-name
+    storage : Storage,              # pylint: disable=redefined-outer-name
+) -> None:
+    check_config_interfaces(driver, storage)
+    check_config_network_instances(driver, storage)
+
+    resources_to_delete = list()
+    ni_sr_prefixes = list()
+    for ni in NETWORK_INSTANCES:
+        ni_name = ni['name']
+        for ni_sr in ni.get('static_routes', list()):
+            ni_sr_prefix   = ni_sr['prefix'  ]
+            ni_sr_next_hop = ni_sr['next_hop']
+            ni_sr_metric   = ni_sr['metric'  ]
+            ni_sr_next_hop_index = 'AUTO_{:d}_{:s}'.format(ni_sr_metric, '-'.join(ni_sr_next_hop.split('.')))
+            resources_to_delete.append(network_instance_static_route(
+                ni_name, ni_sr_prefix, ni_sr_next_hop_index, ni_sr_next_hop, metric=ni_sr_metric
+            ))
+            ni_sr_prefixes.append((ni_name, ni_sr_prefix))
+
+            storage.network_instances.protocols.remove(ni_name, 'STATIC')
+            storage.network_instances.protocol_static.remove(ni_name, 'STATIC', ni_sr_prefix)
+            storage.network_instances.tables.remove(ni_name, 'STATIC', 'IPV4')
+            storage.network_instances.tables.remove(ni_name, 'STATIC', 'IPV6')
+
+    results_deleteconfig = del_config(driver, resources_to_delete)
+    check_updates(results_deleteconfig, '/network_instance[{:s}]/static_route[{:s}]', ni_sr_prefixes)
+
+    check_config_interfaces(driver, storage, max_retries=10, retry_delay=2.0)
+    #check_config_network_instances(driver, storage, max_retries=10, retry_delay=2.0)
+
+
+def test_del_interfaces(
+    driver : GnmiOpenConfigDriver,  # pylint: disable=redefined-outer-name
+    storage : Storage,              # pylint: disable=redefined-outer-name
+) -> None:
+    check_config_interfaces(driver, storage)
+    #check_config_network_instances(driver, storage)
+
+    resources_to_delete = list()
+    if_names = list()
+    for ni in NETWORK_INSTANCES:
+        for ni_if in ni.get('interfaces', list()):
+            if_name      = ni_if['name'       ]
+            subif_index  = ni_if['index'      ]
+            ipv4_address = ni_if['ipv4_addr'  ]
+            ipv4_prefix  = ni_if['ipv4_prefix']
+            enabled      = ni_if['enabled'    ]
+            resources_to_delete.append(interface(if_name, subif_index, ipv4_address, ipv4_prefix, enabled))
+            if_names.append(if_name)
+            storage.interfaces.ipv4_addresses.remove(if_name, subif_index, ipv4_address)
+            default_vlan = storage.network_instances.vlans.get('default', 1)
+            default_vlan_members : List[str] = default_vlan.setdefault('members', list())
+            if if_name not in default_vlan_members: default_vlan_members.append(if_name)
+
+    results_deleteconfig = del_config(driver, resources_to_delete)
+    check_updates(results_deleteconfig, '/interface[{:s}]', if_names)
+
+    check_config_interfaces(driver, storage, max_retries=10, retry_delay=2.0)
+    #check_config_network_instances(driver, storage, max_retries=10, retry_delay=2.0)
+
+
+def test_del_interfaces_from_network_instance(
+    driver : GnmiOpenConfigDriver,  # pylint: disable=redefined-outer-name
+    storage : Storage,              # pylint: disable=redefined-outer-name
+) -> None:
+    check_config_interfaces(driver, storage)
+    #check_config_network_instances(driver, storage)
+
+    resources_to_delete = list()
+    ni_if_names = list()
+    for ni in NETWORK_INSTANCES:
+        ni_name = ni['name']
+        for ni_if in ni.get('interfaces', list()):
+            if_name     = ni_if['name' ]
+            subif_index = ni_if['index']
+            resources_to_delete.append(network_instance_interface(ni_name, if_name, subif_index))
+            ni_if_names.append((ni_name, '{:s}.{:d}'.format(if_name, subif_index)))
+            storage.network_instances.interfaces.remove(ni_name, if_name, subif_index)
+
+    results_deleteconfig = del_config(driver, resources_to_delete)
+    check_updates(results_deleteconfig, '/network_instance[{:s}]/interface[{:s}]', ni_if_names)
+    
+    check_config_interfaces(driver, storage, max_retries=10, retry_delay=2.0)
+    #check_config_network_instances(driver, storage, max_retries=10, retry_delay=2.0)
+
+
+def test_del_network_instances(
+    driver : GnmiOpenConfigDriver,  # pylint: disable=redefined-outer-name
+    storage : Storage,              # pylint: disable=redefined-outer-name
+) -> None:
+    check_config_interfaces(driver, storage)
+    #check_config_network_instances(driver, storage)
+
+    resources_to_delete = list()
+    ni_names = list()
+    for ni in NETWORK_INSTANCES:
+        ni_name = ni['name']
+        ni_type = ni['type']
+        resources_to_delete.append(network_instance(ni_name, ni_type))
+        ni_names.append(ni_name)
+        storage.network_instances.network_instances.remove(ni_name)
+        storage.network_instances.protocols.remove(ni_name, 'DIRECTLY_CONNECTED')
+        storage.network_instances.tables.remove(ni_name, 'DIRECTLY_CONNECTED', 'IPV4')
+        storage.network_instances.tables.remove(ni_name, 'DIRECTLY_CONNECTED', 'IPV6')
+
+    results_deleteconfig = del_config(driver, resources_to_delete)
+    check_updates(results_deleteconfig, '/network_instance[{:s}]', ni_names)
+
+    check_config_interfaces(driver, storage, max_retries=10, retry_delay=2.0)
+    check_config_network_instances(driver, storage, max_retries=10, retry_delay=2.0)

src/device/tests/gnmi_openconfig/tools/__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/device/tests/gnmi_openconfig/tools/check_updates.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.
+
+from typing import Iterable, List, Tuple
+
+def check_updates(results : Iterable[Tuple[str, bool]], format_str : str, item_ids : List[Tuple]) -> None:
+    results = set(results)
+    assert len(results) == len(item_ids)
+    for item_id_fields in item_ids:
+        if isinstance(item_id_fields, (str, int, float, bool)): item_id_fields = (item_id_fields,)
+        assert (format_str.format(*item_id_fields), True) in results

src/device/tests/gnmi_openconfig/tools/manage_config.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 copy, deepdiff, logging, time
+from typing import Callable, Dict, List, Tuple, Union
+from device.service.driver_api._Driver import (
+    RESOURCE_ENDPOINTS, RESOURCE_INTERFACES, RESOURCE_NETWORK_INSTANCES,
+    RESOURCE_ROUTING_POLICIES, RESOURCE_SERVICES
+)
+from device.service.drivers.gnmi_openconfig.GnmiOpenConfigDriver import GnmiOpenConfigDriver
+from device.tests.gnmi_openconfig.storage.Storage import Storage
+from .result_config_adapters import adapt_endpoint, adapt_interface, adapt_network_instance
+
+LOGGER = logging.getLogger(__name__)
+
+def get_config(driver : GnmiOpenConfigDriver, resources_to_get : List[str]) -> List[Tuple[str, Dict]]:
+    LOGGER.info('[get_config] resources_to_get = {:s}'.format(str(resources_to_get)))
+    results_getconfig = driver.GetConfig(resources_to_get)
+    LOGGER.info('[get_config] results_getconfig = {:s}'.format(str(results_getconfig)))
+    return results_getconfig
+
+def set_config(
+    driver : GnmiOpenConfigDriver, resources_to_set : List[Tuple[str, Dict]]
+) -> List[Tuple[str, Union[bool, Exception]]]:
+    LOGGER.info('[set_config] resources_to_set = {:s}'.format(str(resources_to_set)))
+    results_setconfig = driver.SetConfig(resources_to_set)
+    LOGGER.info('[set_config] results_setconfig = {:s}'.format(str(results_setconfig)))
+    return results_setconfig
+
+def del_config(
+    driver : GnmiOpenConfigDriver, resources_to_delete : List[Tuple[str, Dict]]
+) -> List[Tuple[str, Union[bool, Exception]]]:
+    LOGGER.info('resources_to_delete = {:s}'.format(str(resources_to_delete)))
+    results_deleteconfig = driver.DeleteConfig(resources_to_delete)
+    LOGGER.info('results_deleteconfig = {:s}'.format(str(results_deleteconfig)))
+    return results_deleteconfig
+
+def check_expected_config(
+    driver : GnmiOpenConfigDriver, resources_to_get : List[str], expected_config : List[Dict],
+    func_adapt_returned_config : Callable[[Tuple[str, Dict]], Tuple[str, Dict]] = lambda x: x,
+    max_retries : int = 1, retry_delay : float = 0.5
+) -> List[Dict]:
+    LOGGER.info('expected_config = {:s}'.format(str(expected_config)))
+
+    num_retry = 0
+    return_data = None
+    while num_retry < max_retries:
+        results_getconfig = get_config(driver, resources_to_get)
+        return_data = copy.deepcopy(results_getconfig)
+
+        results_getconfig = [
+            func_adapt_returned_config(resource_key, resource_value)
+            for resource_key, resource_value in results_getconfig
+        ]
+
+        diff_data = deepdiff.DeepDiff(sorted(expected_config), sorted(results_getconfig))
+        num_diffs = len(diff_data)
+        if num_diffs == 0: break
+        # let the device take some time to reconfigure
+        time.sleep(retry_delay)
+        num_retry += 1
+
+    if num_diffs > 0: LOGGER.error('Differences[{:d}]:\n{:s}'.format(num_diffs, str(diff_data.pretty())))
+    assert num_diffs == 0
+    return return_data
+
+def check_config_endpoints(
+    driver : GnmiOpenConfigDriver, storage : Storage,
+    max_retries : int = 1, retry_delay : float = 0.5
+) -> List[Dict]:
+    return check_expected_config(
+        driver, [RESOURCE_ENDPOINTS], storage.endpoints.get_expected_config(),
+        adapt_endpoint, max_retries=max_retries, retry_delay=retry_delay
+    )
+
+def check_config_interfaces(
+    driver : GnmiOpenConfigDriver, storage : Storage,
+    max_retries : int = 1, retry_delay : float = 0.5
+) -> List[Dict]:
+    return check_expected_config(
+        driver, [RESOURCE_INTERFACES], storage.interfaces.get_expected_config(),
+        adapt_interface, max_retries=max_retries, retry_delay=retry_delay
+    )
+
+def check_config_network_instances(
+    driver : GnmiOpenConfigDriver, storage : Storage,
+    max_retries : int = 1, retry_delay : float = 0.5
+) -> List[Dict]:
+    return check_expected_config(
+        driver, [RESOURCE_NETWORK_INSTANCES], storage.network_instances.get_expected_config(),
+        adapt_network_instance, max_retries=max_retries, retry_delay=retry_delay
+    )

src/device/tests/gnmi_openconfig/tools/request_composers copy.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.
+
+from typing import Dict, Tuple
+
+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
+
+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_static_route(ni_name, prefix, next_hop_index, next_hop, metric=1) -> Tuple[str, Dict]:
+    str_path = '/network_instance[{:s}]/static_route[{:s}]'.format(ni_name, prefix)
+    str_data = {
+        'name': ni_name, 'identifier': 'STATIC', 'protocol_name': 'STATIC',
+        'prefix': prefix, 'index': next_hop_index, 'next_hop': next_hop, 'metric': metric
+    }
+    return str_path, str_data
+
+def network_instance_interface(ni_name, if_name, sif_index) -> Tuple[str, Dict]:
+    ni_if_id = '{:s}.{:d}'.format(if_name, sif_index)
+    str_path = '/network_instance[{:s}]/interface[{:s}]'.format(ni_name, ni_if_id)
+    str_data = {
+        'name': ni_name, 'id': ni_if_id, 'interface': if_name, 'subinterface': sif_index
+    }
+    return str_path, str_data

src/device/tests/gnmi_openconfig/tools/request_composers.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.
+
+from typing import Dict, Tuple
+
+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
+
+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_static_route(ni_name, prefix, next_hop_index, next_hop, metric=1) -> Tuple[str, Dict]:
+    str_path = '/network_instance[{:s}]/static_route[{:s}]'.format(ni_name, prefix)
+    str_data = {
+        'name': ni_name, 'prefix': prefix, 'next_hop_index': next_hop_index, 'next_hop': next_hop, 'metric': metric
+    }
+    return str_path, str_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

src/device/tests/gnmi_openconfig/tools/result_config_adapters.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 re
+from typing import Dict, Tuple
+
+def adapt_endpoint(resource_key : str, resource_value : Dict) -> Tuple[str, Dict]:
+    return resource_key, resource_value
+
+def adapt_interface(resource_key : str, resource_value : Dict) -> Tuple[str, Dict]:
+    return resource_key, resource_value
+
+def adapt_network_instance(resource_key : str, resource_value : Dict) -> Tuple[str, Dict]:
+    match = re.match(r'^\/network\_instance\[([^\]]+)\]\/vlan\[([^\]]+)\]$', resource_key)
+    if match is not None:
+        members = resource_value.get('members')
+        if len(members) > 0: resource_value['members'] = sorted(members)
+    return resource_key, resource_value

src/device/tests/test_gnmi.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, os, sys, time
-from typing import Dict, Tuple
-os.environ['DEVICE_EMULATED_ONLY'] = 'YES'
-from device.service.drivers.gnmi_openconfig.GnmiOpenConfigDriver import GnmiOpenConfigDriver # pylint: disable=wrong-import-position
-#from device.service.driver_api._Driver import (
-#    RESOURCE_ENDPOINTS, RESOURCE_INTERFACES, RESOURCE_NETWORK_INSTANCES, RESOURCE_ROUTING_POLICIES, RESOURCE_SERVICES
-#)
-
-logging.basicConfig(level=logging.DEBUG)
-LOGGER = logging.getLogger(__name__)
-LOGGER.setLevel(logging.DEBUG)
-
-# +---+---------------------------+--------------+---------------------------------+-------+---------+--------------------+--------------+
-# | # |           Name            | Container ID |              Image              | Kind  |  State  |    IPv4 Address    | IPv6 Address |
-# +---+---------------------------+--------------+---------------------------------+-------+---------+--------------------+--------------+
-# | 1 | clab-tfs-scenario-client1 | a8d48ec3265a | ghcr.io/hellt/network-multitool | linux | running | 172.100.100.201/24 | N/A          |
-# | 2 | clab-tfs-scenario-client2 | fc88436d2b32 | ghcr.io/hellt/network-multitool | linux | running | 172.100.100.202/24 | N/A          |
-# | 3 | clab-tfs-scenario-srl1    | b995b9bdadda | ghcr.io/nokia/srlinux           | srl   | running | 172.100.100.101/24 | N/A          |
-# | 4 | clab-tfs-scenario-srl2    | aacfc38cc376 | ghcr.io/nokia/srlinux           | srl   | running | 172.100.100.102/24 | N/A          |
-# +---+---------------------------+--------------+---------------------------------+-------+---------+--------------------+--------------+
-
-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
-
-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_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_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 main():
-    driver_settings = {
-        'protocol': 'gnmi',
-        'username': 'admin',
-        'password': 'NokiaSrl1!',
-        'use_tls' : True,
-    }
-    driver = GnmiOpenConfigDriver('172.100.100.102', 57400, **driver_settings)
-    driver.Connect()
-
-    #resources_to_get = []
-    #resources_to_get = [RESOURCE_ENDPOINTS]
-    #resources_to_get = [RESOURCE_INTERFACES]
-    #resources_to_get = [RESOURCE_NETWORK_INSTANCES]
-    #resources_to_get = [RESOURCE_ROUTING_POLICIES]
-    #resources_to_get = [RESOURCE_SERVICES]
-    #LOGGER.info('resources_to_get = {:s}'.format(str(resources_to_get)))
-    #results_getconfig = driver.GetConfig(resources_to_get)
-    #LOGGER.info('results_getconfig = {:s}'.format(str(results_getconfig)))
-
-    #resources_to_set = [
-    #    network_instance('test-svc', 'L3VRF'),
-    #
-    #    interface('ethernet-1/1', 0, '172.16.0.1', 24, True),
-    #    network_instance_interface('test-svc', 'ethernet-1/1', 0),
-    #
-    #    interface('ethernet-1/2', 0, '172.0.0.1', 24, True),
-    #    network_instance_interface('test-svc', 'ethernet-1/2', 0),
-    #
-    #    network_instance_static_route('test-svc', '172.0.0.0/24', '172.16.0.2'),
-    #    network_instance_static_route('test-svc', '172.2.0.0/24', '172.16.0.3'),
-    #]
-    #LOGGER.info('resources_to_set = {:s}'.format(str(resources_to_set)))
-    #results_setconfig = driver.SetConfig(resources_to_set)
-    #LOGGER.info('results_setconfig = {:s}'.format(str(results_setconfig)))
-
-    resources_to_delete = [
-        #network_instance_static_route('d35fc1d9', '172.0.0.0/24', '172.16.0.2'),
-        #network_instance_static_route('d35fc1d9', '172.2.0.0/24', '172.16.0.3'),
-    
-        #network_instance_interface('d35fc1d9', 'ethernet-1/1', 0),
-        #network_instance_interface('d35fc1d9', 'ethernet-1/2', 0),
-    
-        interface('ethernet-1/1', 0, '172.16.1.1', 24, True),
-        interface('ethernet-1/2', 0, '172.0.0.2', 24, True),
-    
-        network_instance('20f66fb5', 'L3VRF'),
-    ]
-    LOGGER.info('resources_to_delete = {:s}'.format(str(resources_to_delete)))
-    results_deleteconfig = driver.DeleteConfig(resources_to_delete)
-    LOGGER.info('results_deleteconfig = {:s}'.format(str(results_deleteconfig)))
-
-    time.sleep(1)
-
-    driver.Disconnect()
-    return 0
-
-if __name__ == '__main__':
-    sys.exit(main())

src/e2e_orchestrator/.gitlab-ci.yml

@@ -13,9 +13,9 @@
 # limitations under the License.
 
 # build, tag and push the Docker image to the gitlab registry
-build e2eorchestrator:
+build e2e_orchestrator:
   variables:
-    IMAGE_NAME: 'e2eorchestrator' # name of the microservice
+    IMAGE_NAME: 'e2e_orchestrator' # name of the microservice
     IMAGE_TAG: 'latest' # tag of the container image (production, development, etc)
   stage: build
   before_script:

src/e2e_orchestrator/Dockerfile

@@ -77,8 +77,11 @@ RUN pip-compile --quiet --output-file=e2e_orchestrator/requirements.txt e2e_orch
 RUN python3 -m pip install -r e2e_orchestrator/requirements.txt
 
 # Add component files into working directory
-COPY --chown=teraflow:teraflow ./src/context/. context
-COPY --chown=teraflow:teraflow ./src/e2e_orchestrator/. e2e_orchestrator
+COPY src/context/__init__.py context/__init__.py
+COPY src/context/client/. context/client/
+COPY src/service/__init__.py service/__init__.py
+COPY src/service/client/. service/client/
+COPY src/e2e_orchestrator/. e2e_orchestrator/
 
 # Start the service
 ENTRYPOINT ["python", "-m", "e2e_orchestrator.service"]

src/e2e_orchestrator/requirements.in

@@ -12,4 +12,5 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-networkx
\ No newline at end of file
+networkx
+websockets==12.0

src/e2e_orchestrator/service/E2EOrchestratorServiceServicerImpl.py

@@ -12,33 +12,181 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import logging
-
-import networkx as nx
-import grpc
 import copy
-
-from common.Constants import ServiceNameEnum
-from common.method_wrappers.Decorator import (MetricsPool, MetricTypeEnum, safe_and_metered_rpc_method)
+from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method
 from common.proto.e2eorchestrator_pb2 import E2EOrchestratorRequest, E2EOrchestratorReply
-from common.proto.context_pb2 import Empty, Connection, EndPointId
+from common.proto.context_pb2 import (
+    Empty, Connection, EndPointId, Link, LinkId, TopologyDetails, Topology, Context, Service, ServiceId,
+    ServiceTypeEnum, ServiceStatusEnum)
 from common.proto.e2eorchestrator_pb2_grpc import E2EOrchestratorServiceServicer
+from common.Settings import get_setting
 from context.client.ContextClient import ContextClient
+from service.client.ServiceClient import ServiceClient
 from context.service.database.uuids.EndPoint import endpoint_get_uuid
+from context.service.database.uuids.Device import device_get_uuid
+from common.proto.vnt_manager_pb2 import VNTSubscriptionRequest
+from common.tools.grpc.Tools import grpc_message_to_json_string
+import grpc
+import json
+import logging
+import networkx as nx
+from threading import Thread
+from websockets.sync.client import connect
+from websockets.sync.server import serve
+from common.Constants import DEFAULT_CONTEXT_NAME
 
 
 LOGGER = logging.getLogger(__name__)
+logging.getLogger("websockets").propagate = True
 
 METRICS_POOL = MetricsPool("E2EOrchestrator", "RPC")
 
+
 context_client: ContextClient = ContextClient()
+service_client: ServiceClient = ServiceClient()
+
+EXT_HOST = str(get_setting('WS_IP_HOST'))
+EXT_PORT = str(get_setting('WS_IP_PORT'))
+
+OWN_HOST = str(get_setting('WS_E2E_HOST'))
+OWN_PORT = str(get_setting('WS_E2E_PORT'))
+
+
+ALL_HOSTS = "0.0.0.0"
+
+class SubscriptionServer(Thread):
+    def __init__(self):
+        Thread.__init__(self)
+            
+    def run(self):
+        url = "ws://" + EXT_HOST + ":" + EXT_PORT
+        request = VNTSubscriptionRequest()
+        request.host = OWN_HOST
+        request.port = OWN_PORT
+        try: 
+            LOGGER.debug("Trying to connect to {}".format(url))
+            websocket = connect(url)
+        except Exception as ex:
+            LOGGER.error('Error connecting to {}'.format(url))
+        else:
+            with websocket:
+                LOGGER.debug("Connected to {}".format(url))
+                send = grpc_message_to_json_string(request)
+                websocket.send(send)
+                LOGGER.debug("Sent: {}".format(send))
+                try:
+                    message = websocket.recv()
+                    LOGGER.debug("Received message from WebSocket: {}".format(message))
+                except Exception as ex:
+                    LOGGER.error('Exception receiving from WebSocket: {}'.format(ex))
+
+            self._events_server()
+
+
+    def _events_server(self):
+        all_hosts = "0.0.0.0"
+
+        try:
+            server = serve(self._event_received, all_hosts, int(OWN_PORT))
+        except Exception as ex:
+            LOGGER.error('Error starting server on {}:{}'.format(all_hosts, OWN_PORT))
+            LOGGER.error('Exception!: {}'.format(ex))
+        else:
+            with server:
+                LOGGER.info("Running events server...: {}:{}".format(all_hosts, OWN_PORT))
+                server.serve_forever()
+
+
+    def _event_received(self, connection):
+        LOGGER.info("EVENT received!")
+        for message in connection:
+            message_json = json.loads(message)
+            # LOGGER.info("message_json: {}".format(message_json))
+
+            # Link creation
+            if 'link_id' in message_json:
+                link = Link(**message_json)
+
+                service = Service()
+                service.service_id.service_uuid.uuid = link.link_id.link_uuid.uuid
+                service.service_id.context_id.context_uuid.uuid = DEFAULT_CONTEXT_NAME
+                service.service_type = ServiceTypeEnum.SERVICETYPE_OPTICAL_CONNECTIVITY
+                service.service_status.service_status = ServiceStatusEnum.SERVICESTATUS_PLANNED
+                service_client.CreateService(service)
+
+                links = context_client.ListLinks(Empty()).links
+                a_device_uuid = device_get_uuid(link.link_endpoint_ids[0].device_id)
+                a_endpoint_uuid = endpoint_get_uuid(link.link_endpoint_ids[0])[2]
+                z_device_uuid = device_get_uuid(link.link_endpoint_ids[1].device_id)
+                z_endpoint_uuid = endpoint_get_uuid(link.link_endpoint_ids[1])[2]
+
+                for _link in links:
+                    for _endpoint_id in _link.link_endpoint_ids:
+                        if _endpoint_id.device_id.device_uuid.uuid == a_device_uuid and \
+                        _endpoint_id.endpoint_uuid.uuid == a_endpoint_uuid:
+                            a_ep_id = _endpoint_id
+                        elif _endpoint_id.device_id.device_uuid.uuid == z_device_uuid and \
+                        _endpoint_id.endpoint_uuid.uuid == z_endpoint_uuid:
+                            z_ep_id = _endpoint_id
+
+                if (not 'a_ep_id' in locals()) or (not 'z_ep_id' in locals()):
+                    error_msg = 'Could not get VNT link endpoints'
+                    LOGGER.error(error_msg)
+                    connection.send(error_msg)
+                    return
+
+                service.service_endpoint_ids.append(copy.deepcopy(a_ep_id))
+                service.service_endpoint_ids.append(copy.deepcopy(z_ep_id))
+
+                # service_client.UpdateService(service)
+                connection.send(grpc_message_to_json_string(link))
+            # Link removal
+            elif 'link_uuid' in message_json:
+                LOGGER.info('REMOVING VIRTUAL LINK')
+                link_id = LinkId(**message_json)
+
+                service_id = ServiceId()
+                service_id.service_uuid.uuid = link_id.link_uuid.uuid
+                service_id.context_id.context_uuid.uuid = DEFAULT_CONTEXT_NAME
+                # service_client.DeleteService(service_id)
+                connection.send(grpc_message_to_json_string(link_id))
+                context_client.RemoveLink(link_id)
+            # Topology received
+            else:
+                LOGGER.info('TOPOLOGY')
+                topology_details = TopologyDetails(**message_json)
+
+                context = Context()
+                context.context_id.context_uuid.uuid = topology_details.topology_id.context_id.context_uuid.uuid
+                context_client.SetContext(context)
+
+                topology = Topology()
+                topology.topology_id.context_id.CopyFrom(context.context_id)
+                topology.topology_id.topology_uuid.uuid = topology_details.topology_id.topology_uuid.uuid
+                context_client.SetTopology(topology)
+
+                for device in topology_details.devices:
+                    context_client.SetDevice(device)
+
+                for link in topology_details.links:
+                    context_client.SetLink(link)
+
 
 
 class E2EOrchestratorServiceServicerImpl(E2EOrchestratorServiceServicer):
     def __init__(self):
         LOGGER.debug("Creating Servicer...")
-        LOGGER.debug("Servicer Created")
+        try:
+            LOGGER.debug("Requesting subscription")
+            sub_server = SubscriptionServer()
+            sub_server.start()
+            LOGGER.debug("Servicer Created")
+
+        except Exception as ex:
+            LOGGER.info("Exception!: {}".format(ex))
+
 
+        
     @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
     def Compute(self, request: E2EOrchestratorRequest, context: grpc.ServicerContext) -> E2EOrchestratorReply:
         endpoints_ids = []

src/e2e_orchestrator/service/__main__.py

@@ -43,13 +43,6 @@ def main():
     logging.basicConfig(level=log_level)
     LOGGER = logging.getLogger(__name__)
 
-    wait_for_environment_variables(
-        [
-            get_env_var_name(ServiceNameEnum.E2EORCHESTRATOR, ENVVAR_SUFIX_SERVICE_HOST),
-            get_env_var_name(ServiceNameEnum.E2EORCHESTRATOR, ENVVAR_SUFIX_SERVICE_PORT_GRPC),
-        ]
-    )
-
     signal.signal(signal.SIGINT, signal_handler)
     signal.signal(signal.SIGTERM, signal_handler)