diff --git a/src/device/tests/gnmi_openconfig/storage/StorageNetworkInstance.py b/src/device/tests/gnmi_openconfig/storage/StorageNetworkInstance.py
index 558cc032cdf0be46733f565f7c9143bdcc401e78..fa336488317776a10e20b283acfc1dcfbc3d7b8a 100644
--- a/src/device/tests/gnmi_openconfig/storage/StorageNetworkInstance.py
+++ b/src/device/tests/gnmi_openconfig/storage/StorageNetworkInstance.py
@@ -45,16 +45,14 @@ class NetworkInstances:
class Interfaces:
STRUCT : List[Tuple[str, List[str]]] = [
- ('/network_instance[{:s}]/interface[{:s}]', ['ni_name', 'if_name']),
+ ('/network_instance[{:s}]/interface[{:s}]', []),
]
def __init__(self) -> None:
self._items : Dict[Tuple[str, str], Dict] = dict()
def add(self, ni_name : str, if_name : str) -> None:
- item = self._items.setdefault((ni_name, if_name), dict())
- item['ni_name'] = ni_name
- item['if_name'] = if_name
+ self._items.setdefault((ni_name, if_name), dict())
def remove(self, ni_name : str, if_name : str) -> None:
self._items.pop((ni_name, if_name), None)
@@ -86,6 +84,11 @@ class StaticRoutes:
('/network_instance[{:s}]/protocol[{:s}]/static_routes[{:s}]', ['prefix', 'next_hops']),
]
+ #('/network_instance[{:s}]/static_route[{:s}]'.format(ni_name, static_route['prefix']), {
+ # 'name': ni_name, 'prefix': static_route['prefix'], 'next_hop': static_route['gateway'],
+ # 'next_hop_index': 0, 'metric': static_route['metric']
+ #})
+
def __init__(self) -> None:
self._items : Dict[Tuple[str, str, str], Dict] = dict()
diff --git a/src/device/tests/gnmi_openconfig/test_unitary_gnmi_openconfig.py b/src/device/tests/gnmi_openconfig/test_unitary_gnmi_openconfig.py
index dd0561a2b82b81ff2fe89e8369a829a1950e61b4..3970e65a63ba5325da37cc3ea20baf42936ad947 100644
--- a/src/device/tests/gnmi_openconfig/test_unitary_gnmi_openconfig.py
+++ b/src/device/tests/gnmi_openconfig/test_unitary_gnmi_openconfig.py
@@ -16,21 +16,17 @@ import os
os.environ['DEVICE_EMULATED_ONLY'] = 'YES'
# pylint: disable=wrong-import-position
-import itertools, logging, pytest, time
+import logging, pytest, time
from typing import Dict
-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 .tools.check_config import check_config_endpoints, check_config_interfaces, check_config_network_instances
-from .tools.check_updates import check_updates
-from .tools.expected_config_composers import (
- compose_expected_config__interface, compose_expected_config__network_instance
+from .storage.Storage import Storage
+from .tools.check_config import (
+ check_config_endpoints, check_config_interfaces, check_config_network_instances
)
+from .tools.check_updates import check_updates
from .tools.request_composers import (
interface, network_instance, network_instance_interface, network_instance_static_route
)
-from .storage.Storage import Storage
logging.basicConfig(level=logging.DEBUG)
LOGGER = logging.getLogger(__name__)
@@ -200,7 +196,7 @@ def test_add_interfaces_to_network_instance(
subif_index = ni_if['sif_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)
+ storage.network_instances.interfaces.add(ni_name, if_name) # TODO: add subif_index
LOGGER.info('resources_to_set = {:s}'.format(str(resources_to_set)))
results_setconfig = driver.SetConfig(resources_to_set)
@@ -218,22 +214,26 @@ def test_set_network_instance_static_routes(
check_config_interfaces(driver, storage)
check_config_network_instances(driver, storage)
- # TODO: update structure
+ 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_gateway = ni_sr['gateway']
+ ni_sr_metric = ni_sr['metric' ]
+ resources_to_set.append(
+ network_instance_static_route(ni_name, ni_sr_prefix, ni_sr_gateway, metric=ni_sr_metric)
+ )
+ ni_sr_prefixes.append((ni_name, ni_sr_prefix))
+ storage.network_instances.protocol_static.add(ni_name, 'STATIC', ni_sr_prefix, {
+ 'prefix': ni_sr_prefix,
+ })
- resources_to_set = list(itertools.chain(*[
- [
- network_instance_static_route(ni['name'], ni_sr['prefix'], ni_sr['gateway'], metric=ni_sr['metric'])
- for ni_sr in ni.get('static_routes', list())
- ]
- for ni in NETWORK_INSTANCES
- ]))
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)))
- check_updates(results_setconfig, '/network_instance[{:s}]/static_route[{:s}]', list(itertools.chain(*[
- [(ni['name'], ni_sr['prefix']) for ni_sr in ni.get('static_routes', list())]
- for ni in NETWORK_INSTANCES
- ])))
+ check_updates(results_setconfig, '/network_instance[{:s}]/static_route[{:s}]', ni_sr_prefixes)
check_config_interfaces(driver, storage, max_retries=5)
check_config_network_instances(driver, storage, max_retries=5)
@@ -246,22 +246,24 @@ def test_del_network_instance_static_routes(
check_config_interfaces(driver, storage)
check_config_network_instances(driver, storage)
- # TODO: update structure
+ 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_gateway = ni_sr['gateway']
+ ni_sr_metric = ni_sr['metric' ]
+ resources_to_delete.append(
+ network_instance_static_route(ni_name, ni_sr_prefix, ni_sr_gateway, metric=ni_sr_metric)
+ )
+ ni_sr_prefixes.append((ni_name, ni_sr_prefix))
+ storage.network_instances.protocol_static.remove(ni_name, 'STATIC', ni_sr_prefix)
- resources_to_delete = list(itertools.chain(*[
- [
- network_instance_static_route(ni['name'], ni_sr['prefix'], ni_sr['gateway'], metric=ni_sr['metric'])
- for ni_sr in ni.get('static_routes', list())
- ]
- for ni in NETWORK_INSTANCES
- ]))
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)))
- check_updates(results_deleteconfig, '/network_instance[{:s}]/static_route[{:s}]', list(itertools.chain(*[
- [(ni['name'], ni_sr['prefix']) for ni_sr in ni.get('static_routes', list())]
- for ni in NETWORK_INSTANCES
- ])))
+ check_updates(results_deleteconfig, '/network_instance[{:s}]/static_route[{:s}]', ni_sr_prefixes)
check_config_interfaces(driver, storage, max_retries=5)
check_config_network_instances(driver, storage, max_retries=5)
@@ -274,26 +276,22 @@ def test_del_interfaces_from_network_instance(
check_config_interfaces(driver, storage)
check_config_network_instances(driver, storage)
- # TODO: update structure
+ 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['if_name']
+ subif_index = ni_if['sif_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) # TODO: add subif_index
- resources_to_delete = list(itertools.chain(*[
- [
- network_instance_interface(ni['name'], ni_if['if_name'], ni_if['subif_index'])
- for ni_if in ni.get('interfaces', list())
- ]
- for ni in NETWORK_INSTANCES
- ]))
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)))
- check_updates(results_deleteconfig, '/network_instance[{:s}]/interface[{:s}]', list(itertools.chain(*[
- [
- (ni['name'], '{:s}.{:d}'.format(ni_if['if_name'], ni_if['subif_index']))
- for ni_if in ni.get('interfaces', list())
- ]
- for ni in NETWORK_INSTANCES
- ])))
-
+ check_updates(results_deleteconfig, '/network_instance[{:s}]/interface[{:s}]', ni_if_names)
+
check_config_interfaces(driver, storage, max_retries=5)
check_config_network_instances(driver, storage, max_retries=5)
@@ -305,22 +303,24 @@ def test_del_interfaces(
check_config_interfaces(driver, storage)
check_config_network_instances(driver, storage)
- # TODO: update structure
+ resources_to_delete = list()
+ if_names = list()
+ for ni in NETWORK_INSTANCES:
+ ni_name = ni['name']
+ for ni_if in ni.get('interfaces', list()):
+ if_name = ni_if['if_name']
+ subif_index = ni_if['sif_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(ni_name)
+ storage.interfaces.ipv4_addresses.remove(if_name, subif_index, ipv4_address)
- resources_to_delete = list(itertools.chain(*[
- [
- interface(ni_if['if_name'], ni_if['sif_index'], ni_if['ipv4_addr'], ni_if['ipv4_prefix'], ni_if['enabled'])
- for ni_if in ni.get('interfaces', list())
- ]
- for ni in NETWORK_INSTANCES
- ]))
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)))
- check_updates(results_deleteconfig, '/interface[{:s}]', list(itertools.chain(*[
- [ni_if['name'] for ni_if in ni.get('interfaces', list())]
- for ni in NETWORK_INSTANCES
- ])))
+ check_updates(results_deleteconfig, '/interface[{:s}]', if_names)
check_config_interfaces(driver, storage, max_retries=5)
check_config_network_instances(driver, storage, max_retries=5)
@@ -333,16 +333,19 @@ def test_del_network_instances(
check_config_interfaces(driver, storage)
check_config_network_instances(driver, storage)
- # TODO: update structure
+ 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)
- resources_to_delete = [
- network_instance(ni['name'], ni['type'])
- for ni in NETWORK_INSTANCES
- ]
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)))
- check_updates(results_deleteconfig, '/network_instance[{:s}]', [ni['name'] for ni in NETWORK_INSTANCES])
+ check_updates(results_deleteconfig, '/network_instance[{:s}]', ni_names)
check_config_interfaces(driver, storage, max_retries=5)
check_config_network_instances(driver, storage, max_retries=5)
diff --git a/src/device/tests/gnmi_openconfig/tools/check_config.py b/src/device/tests/gnmi_openconfig/tools/check_config.py
index 017a7038e3c264cba9a32fda2bcd4cb7ead38589..974acdeba4c02b1197c2af74a229f537118a4577 100644
--- a/src/device/tests/gnmi_openconfig/tools/check_config.py
+++ b/src/device/tests/gnmi_openconfig/tools/check_config.py
@@ -75,7 +75,6 @@ def check_config_network_instances(
driver : GnmiOpenConfigDriver, storage : Storage,
max_retries : int = 1, retry_delay : float = 0.5
) -> List[Dict]:
- expected_config =
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
diff --git a/src/device/tests/gnmi_openconfig/tools/expected_config_composers.py b/src/device/tests/gnmi_openconfig/tools/expected_config_composers.py
deleted file mode 100644
index 487476c016fd86991c50f66fb8dc9ec9b112d643..0000000000000000000000000000000000000000
--- a/src/device/tests/gnmi_openconfig/tools/expected_config_composers.py
+++ /dev/null
@@ -1,58 +0,0 @@
-# Copyright 2022-2023 ETSI TeraFlowSDN - TFS OSG (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
-
-
-def compose_expected_config__network_instance(
- network_instances : List[Dict], include_interfaces : bool = False, include_static_routes : bool = False
-) -> List[Dict]:
- expected_config = list()
- for network_instance in network_instances:
- ni_name = network_instance['name']
- ni_type = network_instance['type']
-
- expected_config.extend([
- ('/network_instance[{:s}]'.format(ni_name), {
- 'name': ni_name, 'type': ni_type
- }),
- ('/network_instance[{:s}]/protocol[DIRECTLY_CONNECTED]'.format(ni_name), {
- 'id': 'DIRECTLY_CONNECTED', 'name': 'DIRECTLY_CONNECTED'
- }),
- ('/network_instance[{:s}]/table[DIRECTLY_CONNECTED,IPV4]'.format(ni_name), {
- 'protocol': 'DIRECTLY_CONNECTED', 'address_family': 'IPV4'
- }),
- ('/network_instance[{:s}]/table[DIRECTLY_CONNECTED,IPV6]'.format(ni_name), {
- 'protocol': 'DIRECTLY_CONNECTED', 'address_family': 'IPV6'
- })
- ])
-
- if include_interfaces:
- expected_config.extend([
- ('/network_instance[{:s}]/interface[{:s}]'.format(ni_name, interface['name']), {
-
- })
- for interface in network_instance.get('interfaces', list())
- ])
-
- if include_static_routes:
- expected_config.extend([
- ('/network_instance[{:s}]/static_route[{:s}]'.format(ni_name, static_route['prefix']), {
- 'name': ni_name, 'prefix': static_route['prefix'], 'next_hop': static_route['gateway'],
- 'next_hop_index': 0, 'metric': static_route['metric']
- })
- for static_route in network_instance.get('static_routes', list())
- ])
-
- return expected_config