diff --git a/scripts/run_tests_locally-device-gnmi-openconfig.sh b/scripts/run_tests_locally-device-gnmi-openconfig.sh
index d81684da10fd259fbd4f8c052a5b0218f71b2021..7183b4104a13894afe7ed8b86b0c70e0251e7312 100755
--- a/scripts/run_tests_locally-device-gnmi-openconfig.sh
+++ b/scripts/run_tests_locally-device-gnmi-openconfig.sh
@@ -22,4 +22,4 @@ RCFILE=$PROJECTDIR/coverage/.coveragerc
# Run unitary tests and analyze coverage of code at same time
# helpful pytest flags: --log-level=INFO -o log_cli=true --verbose --maxfail=1 --durations=0
coverage run --rcfile=$RCFILE --append -m pytest --log-level=INFO --verbose \
- device/tests/test_unitary_gnmi_openconfig.py
+ device/tests/gnmi_openconfig/test_unitary_gnmi_openconfig.py
diff --git a/src/device/service/drivers/gnmi_openconfig/handlers/InterfaceCounter.py b/src/device/service/drivers/gnmi_openconfig/handlers/InterfaceCounter.py
index 1c2cfc17aea4498cbcac7c3460cc53862847cf70..d4701826cc43c9418ade2ba677adf22f0983a11c 100644
--- a/src/device/service/drivers/gnmi_openconfig/handlers/InterfaceCounter.py
+++ b/src/device/service/drivers/gnmi_openconfig/handlers/InterfaceCounter.py
@@ -12,11 +12,10 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import json, logging
+import json, libyang, logging
from typing import Any, Dict, List, Tuple
-import pyangbind.lib.pybindJSON as pybindJSON
-from . import openconfig
from ._Handler import _Handler
+from .YangHandler import YangHandler
LOGGER = logging.getLogger(__name__)
@@ -25,40 +24,41 @@ class InterfaceCounterHandler(_Handler):
def get_resource_key(self) -> str: return '/interface/counters'
def get_path(self) -> str: return '/openconfig-interfaces:interfaces/interface/state/counters'
- def parse(self, json_data : Dict) -> List[Tuple[str, Dict[str, Any]]]:
+ def parse(
+ self, json_data : Dict, yang_handler : YangHandler
+ ) -> List[Tuple[str, Dict[str, Any]]]:
LOGGER.info('json_data = {:s}'.format(json.dumps(json_data)))
- oc_interfaces = pybindJSON.loads_ietf(json_data, openconfig.interfaces, 'interfaces')
- LOGGER.info('oc_interfaces = {:s}'.format(pybindJSON.dumps(oc_interfaces, mode='ietf')))
- counters = []
- for interface_key, oc_interface in oc_interfaces.interface.items():
- LOGGER.info('interface_key={:s} oc_interfaces={:s}'.format(
- interface_key, pybindJSON.dumps(oc_interface, mode='ietf')
- ))
+ yang_interfaces_path = self.get_path()
+ json_data_valid = yang_handler.parse_to_dict(yang_interfaces_path, json_data, fmt='json')
- interface = {}
- interface['name'] = oc_interface.name
-
- interface_counters = oc_interface.state.counters
- interface['in-broadcast-pkts' ] = interface_counters.in_broadcast_pkts
- interface['in-discards' ] = interface_counters.in_discards
- interface['in-errors' ] = interface_counters.in_errors
- interface['in-fcs-errors' ] = interface_counters.in_fcs_errors
- interface['in-multicast-pkts' ] = interface_counters.in_multicast_pkts
- interface['in-octets' ] = interface_counters.in_octets
- interface['in-pkts' ] = interface_counters.in_pkts
- interface['in-unicast-pkts' ] = interface_counters.in_unicast_pkts
- interface['out-broadcast-pkts'] = interface_counters.out_broadcast_pkts
- interface['out-discards' ] = interface_counters.out_discards
- interface['out-errors' ] = interface_counters.out_errors
- interface['out-multicast-pkts'] = interface_counters.out_multicast_pkts
- interface['out-octets' ] = interface_counters.out_octets
- interface['out-pkts' ] = interface_counters.out_pkts
- interface['out-unicast-pkts' ] = interface_counters.out_unicast_pkts
+ entries = []
+ for interface in json_data_valid['interfaces']['interface']:
+ LOGGER.info('interface={:s}'.format(str(interface)))
+ interface_name = interface['name']
+ interface_counters = interface.get('state', {}).get('counters', {})
+ _interface = {
+ 'name' : interface_name,
+ 'in-broadcast-pkts' : interface_counters['in_broadcast_pkts' ],
+ 'in-discards' : interface_counters['in_discards' ],
+ 'in-errors' : interface_counters['in_errors' ],
+ 'in-fcs-errors' : interface_counters['in_fcs_errors' ],
+ 'in-multicast-pkts' : interface_counters['in_multicast_pkts' ],
+ 'in-octets' : interface_counters['in_octets' ],
+ 'in-pkts' : interface_counters['in_pkts' ],
+ 'in-unicast-pkts' : interface_counters['in_unicast_pkts' ],
+ 'out-broadcast-pkts': interface_counters['out_broadcast_pkts'],
+ 'out-discards' : interface_counters['out_discards' ],
+ 'out-errors' : interface_counters['out_errors' ],
+ 'out-multicast-pkts': interface_counters['out_multicast_pkts'],
+ 'out-octets' : interface_counters['out_octets' ],
+ 'out-pkts' : interface_counters['out_pkts' ],
+ 'out-unicast-pkts' : interface_counters['out_unicast_pkts' ],
+ }
LOGGER.info('interface = {:s}'.format(str(interface)))
- if len(interface) == 0: continue
- counters.append(('/interface[{:s}]'.format(interface['name']), interface))
+ entry_interface_key = '/interface[{:s}]'.format(interface_name)
+ entries.append((entry_interface_key, _interface))
- return counters
+ return entries
diff --git a/src/device/service/drivers/gnmi_openconfig/handlers/NetworkInstance.py b/src/device/service/drivers/gnmi_openconfig/handlers/NetworkInstance.py
index 0b4d157452b66d74efadc883b5f59c49b4bee47b..1efed024c1c9e8f5d0aad0494c4a1e4d51bebf0e 100644
--- a/src/device/service/drivers/gnmi_openconfig/handlers/NetworkInstance.py
+++ b/src/device/service/drivers/gnmi_openconfig/handlers/NetworkInstance.py
@@ -12,11 +12,10 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import json, libyang, logging
-import operator
+import json, libyang, logging, operator
from typing import Any, Dict, List, Tuple
from ._Handler import _Handler
-from .Tools import get_bool, get_int, get_str
+from .Tools import get_str
from .YangHandler import YangHandler
LOGGER = logging.getLogger(__name__)
@@ -45,7 +44,7 @@ class NetworkInstanceHandler(_Handler):
def compose(
self, resource_key : str, resource_value : Dict, yang_handler : YangHandler, delete : bool = False
) -> Tuple[str, str]:
- ni_name = get_str(resource_value, 'name') # test-svc
+ ni_name = get_str(resource_value, 'name') # test-svc
if delete:
PATH_TMPL = '/network-instances/network-instance[name={:s}]'
@@ -56,21 +55,32 @@ class NetworkInstanceHandler(_Handler):
ni_type = get_str(resource_value, 'type') # L3VRF / L2VSI / ...
ni_type = MAP_NETWORK_INSTANCE_TYPE.get(ni_type, ni_type)
- # 'DIRECTLY_CONNECTED' is implicitly added
+ str_path = '/network-instances/network-instance[name={:s}]'.format(ni_name)
+ #str_data = json.dumps({
+ # 'name': ni_name,
+ # 'config': {'name': ni_name, 'type': ni_type},
+ #})
+ yang_nis : libyang.DContainer = yang_handler.get_data_path('/openconfig-network-instance:network-instances')
+ yang_ni_path = 'network-instance[name="{:s}"]'.format(ni_name)
+ yang_ni : libyang.DContainer = yang_nis.create_path(yang_ni_path)
+ yang_ni.create_path('config/name', ni_name)
+ yang_ni.create_path('config/type', ni_type)
- str_path = '/network-instances/network-instance[name={:s}]'.format(ni_name)
- str_data = json.dumps({
- 'name': ni_name,
- 'config': {'name': ni_name, 'type': ni_type},
- #'protocols': {'protocol': protocols},
- })
+ # 'DIRECTLY_CONNECTED' is implicitly added
+ #'protocols': {'protocol': protocols},
+
+ str_data = yang_ni.print_mem('json')
+ LOGGER.warning('str_data = {:s}'.format(str(str_data)))
+ json_data = json.loads(str_data)
+ json_data = json_data['openconfig-network-instance:network-instance'][0]
+ str_data = json.dumps(json_data)
return str_path, str_data
def parse(
self, json_data : Dict, yang_handler : YangHandler
) -> List[Tuple[str, Dict[str, Any]]]:
- LOGGER.debug('json_data = {:s}'.format(json.dumps(json_data)))
+ LOGGER.info('json_data = {:s}'.format(json.dumps(json_data)))
# Arista Parsing Fixes:
# - Default instance comes with mpls/signaling-protocols/rsvp-te/global/hellos/state/hello-interval set to 0
diff --git a/src/device/service/drivers/gnmi_openconfig/handlers/NetworkInstanceInterface.py b/src/device/service/drivers/gnmi_openconfig/handlers/NetworkInstanceInterface.py
index 205373fca870ea7338a3c9c043c60306b535c1c0..ab105c2b0e8f8fa086805b8485651891c12bfd5f 100644
--- a/src/device/service/drivers/gnmi_openconfig/handlers/NetworkInstanceInterface.py
+++ b/src/device/service/drivers/gnmi_openconfig/handlers/NetworkInstanceInterface.py
@@ -12,21 +12,31 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import json, logging
+import json, libyang, logging
from typing import Any, Dict, List, Tuple
from ._Handler import _Handler
+from .Tools import get_int, get_str
+from .YangHandler import YangHandler
LOGGER = logging.getLogger(__name__)
+IS_CEOS = True
+
class NetworkInstanceInterfaceHandler(_Handler):
def get_resource_key(self) -> str: return '/network_instance/interface'
- def get_path(self) -> str: return '/network-instances/network-instance/interfaces'
+ def get_path(self) -> str: return '/openconfig-network-instance:network-instances/network-instance/interfaces'
- def compose(self, resource_key : str, resource_value : Dict, delete : bool = False) -> Tuple[str, str]:
- ni_name = str(resource_value['name' ]) # test-svc
- if_name = str(resource_value['if_name' ]) # ethernet-1/1
- sif_index = int(resource_value['sif_index']) # 0
- if_id = '{:s}.{:d}'.format(if_name, sif_index)
+ def compose(
+ self, resource_key : str, resource_value : Dict, yang_handler : YangHandler, delete : bool = False
+ ) -> Tuple[str, str]:
+ ni_name = get_str(resource_value, 'name' ) # test-svc
+ if_name = get_str(resource_value, 'if_name' ) # ethernet-1/1
+ sif_index = get_int(resource_value, 'sif_index', 0) # 0
+
+ if IS_CEOS:
+ if_id = if_name
+ else:
+ if_id = '{:s}.{:d}'.format(if_name, sif_index)
if delete:
PATH_TMPL = '/network-instances/network-instance[name={:s}]/interfaces/interface[id={:s}]'
@@ -35,12 +45,30 @@ class NetworkInstanceInterfaceHandler(_Handler):
return str_path, str_data
str_path = '/network-instances/network-instance[name={:s}]/interfaces/interface[id={:s}]'.format(ni_name, if_id)
- str_data = json.dumps({
- 'id': if_id,
- 'config': {'id': if_id, 'interface': if_name, 'subinterface': sif_index},
- })
+ #str_data = json.dumps({
+ # 'id': if_id,
+ # 'config': {'id': if_id, 'interface': if_name, 'subinterface': sif_index},
+ #})
+
+ yang_nis : libyang.DContainer = yang_handler.get_data_path('/openconfig-network-instance:network-instances')
+ yang_ni : libyang.DContainer = yang_nis.create_path('network-instance[name="{:s}"]'.format(ni_name))
+ yang_ni_ifs : libyang.DContainer = yang_ni.create_path('interfaces')
+ yang_ni_if_path = 'interface[id="{:s}"]'.format(if_id)
+ yang_ni_if : libyang.DContainer = yang_ni_ifs.create_path(yang_ni_if_path)
+ yang_ni_if.create_path('config/id', if_id)
+ yang_ni_if.create_path('config/interface', if_name)
+ yang_ni_if.create_path('config/subinterface', sif_index)
+
+ str_data = yang_ni_if.print_mem('json')
+ LOGGER.warning('[compose] str_data = {:s}'.format(str(str_data)))
+ json_data = json.loads(str_data)
+ json_data = json_data['openconfig-network-instance:interface'][0]
+ str_data = json.dumps(json_data)
return str_path, str_data
- def parse(self, json_data : Dict) -> List[Tuple[str, Dict[str, Any]]]:
+ def parse(
+ self, json_data : Dict, yang_handler : YangHandler
+ ) -> List[Tuple[str, Dict[str, Any]]]:
+ LOGGER.warning('[parse] json_data = {:s}'.format(str(json_data)))
response = []
return response
diff --git a/src/device/service/drivers/gnmi_openconfig/handlers/NetworkInstanceStaticRoute.py b/src/device/service/drivers/gnmi_openconfig/handlers/NetworkInstanceStaticRoute.py
index 9d75e9ac66e023c8f7be44d892cb6eec647761eb..0343e3cbaab3f554f45590c3832bd89b6b552aa8 100644
--- a/src/device/service/drivers/gnmi_openconfig/handlers/NetworkInstanceStaticRoute.py
+++ b/src/device/service/drivers/gnmi_openconfig/handlers/NetworkInstanceStaticRoute.py
@@ -12,21 +12,25 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import json, logging
+import json, libyang, logging
from typing import Any, Dict, List, Tuple
from ._Handler import _Handler
+from .Tools import get_int, get_str
+from .YangHandler import YangHandler
LOGGER = logging.getLogger(__name__)
class NetworkInstanceStaticRouteHandler(_Handler):
def get_resource_key(self) -> str: return '/network_instance/static_route'
- def get_path(self) -> str: return '/network-instances/network-instance/static_route'
+ def get_path(self) -> str: return '/openconfig-network-instance:network-instances/network-instance/static_route'
- def compose(self, resource_key : str, resource_value : Dict, delete : bool = False) -> Tuple[str, str]:
- ni_name = str(resource_value['name' ]) # test-svc
- prefix = str(resource_value['prefix' ]) # '172.0.1.0/24'
+ def compose(
+ self, resource_key : str, resource_value : Dict, yang_handler : YangHandler, delete : bool = False
+ ) -> Tuple[str, str]:
+ ni_name = get_str(resource_value, 'name' ) # test-svc
+ prefix = get_str(resource_value, 'prefix') # '172.0.1.0/24'
- identifier = 'STATIC'
+ identifier = 'openconfig-policy-types:STATIC'
name = 'static'
if delete:
PATH_TMPL = '/network-instances/network-instance[name={:s}]/protocols'
@@ -35,27 +39,56 @@ class NetworkInstanceStaticRouteHandler(_Handler):
str_data = json.dumps({})
return str_path, str_data
- next_hop = str(resource_value['next_hop' ]) # '172.0.0.1'
- next_hop_index = int(resource_value.get('next_hop_index', 0)) # 0
+ next_hop = get_str(resource_value, 'next_hop' ) # '172.0.0.1'
+ next_hop_index = get_int(resource_value, 'next_hop_index', 0) # 0
PATH_TMPL = '/network-instances/network-instance[name={:s}]/protocols/protocol[identifier={:s}][name={:s}]'
str_path = PATH_TMPL.format(ni_name, identifier, name)
- str_data = json.dumps({
- 'identifier': identifier, 'name': name,
- 'config': {'identifier': identifier, 'name': name, 'enabled': True},
- 'static_routes': {'static': [{
- 'prefix': prefix,
- 'config': {'prefix': prefix},
- 'next_hops': {
- 'next-hop': [{
- 'index': next_hop_index,
- 'config': {'index': next_hop_index, 'next_hop': next_hop}
- }]
- }
- }]}
- })
+ #str_data = json.dumps({
+ # 'identifier': identifier, 'name': name,
+ # 'config': {'identifier': identifier, 'name': name, 'enabled': True},
+ # 'static_routes': {'static': [{
+ # 'prefix': prefix,
+ # 'config': {'prefix': prefix},
+ # 'next_hops': {
+ # 'next-hop': [{
+ # 'index': next_hop_index,
+ # 'config': {'index': next_hop_index, 'next_hop': next_hop}
+ # }]
+ # }
+ # }]}
+ #})
+
+ yang_nis : libyang.DContainer = yang_handler.get_data_path('/openconfig-network-instance:network-instances')
+ yang_ni : libyang.DContainer = yang_nis.create_path('network-instance[name="{:s}"]'.format(ni_name))
+ yang_ni_prs : libyang.DContainer = yang_ni.create_path('protocols')
+ yang_ni_pr_path = 'protocol[identifier="{:s}"][name="{:s}"]'.format(identifier, name)
+ yang_ni_pr : libyang.DContainer = yang_ni_prs.create_path(yang_ni_pr_path)
+ yang_ni_pr.create_path('config/identifier', identifier)
+ yang_ni_pr.create_path('config/name', name )
+ yang_ni_pr.create_path('config/enabled', True )
+
+ yang_ni_pr_srs : libyang.DContainer = yang_ni_pr.create_path('static-routes')
+ yang_ni_pr_sr_path = 'static[prefix="{:s}"]'.format(prefix)
+ yang_ni_pr_sr : libyang.DContainer = yang_ni_pr_srs.create_path(yang_ni_pr_sr_path)
+ yang_ni_pr_sr.create_path('config/prefix', prefix)
+
+ yang_ni_pr_sr_nhs : libyang.DContainer = yang_ni_pr_sr.create_path('next-hops')
+ yang_ni_pr_sr_nh_path = 'next-hop[index="{:d}"]'.format(next_hop_index)
+ yang_ni_pr_sr_nh : libyang.DContainer = yang_ni_pr_sr_nhs.create_path(yang_ni_pr_sr_nh_path)
+ yang_ni_pr_sr_nh.create_path('config/index', next_hop_index)
+ yang_ni_pr_sr_nh.create_path('config/next-hop', next_hop)
+
+ str_data = yang_ni_pr.print_mem('json')
+ LOGGER.warning('[compose] str_data = {:s}'.format(str(str_data)))
+ json_data = json.loads(str_data)
+ json_data = json_data['openconfig-network-instance:protocol'][0]
+ str_data = json.dumps(json_data)
return str_path, str_data
- def parse(self, json_data : Dict) -> List[Tuple[str, Dict[str, Any]]]:
+ def parse(
+ self, json_data : Dict, yang_handler : YangHandler
+ ) -> List[Tuple[str, Dict[str, Any]]]:
+ LOGGER.warning('[parse] json_data = {:s}'.format(str(json_data)))
response = []
return response
diff --git a/src/device/tests/gnmi_openconfig/__init__.py b/src/device/tests/gnmi_openconfig/__init__.py
new file mode 100644
index 0000000000000000000000000000000000000000..1549d9811aa5d1c193a44ad45d0d7773236c0612
--- /dev/null
+++ b/src/device/tests/gnmi_openconfig/__init__.py
@@ -0,0 +1,14 @@
+# 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.
+
diff --git a/src/device/tests/gnmi_openconfig/request_composers.py b/src/device/tests/gnmi_openconfig/request_composers.py
new file mode 100644
index 0000000000000000000000000000000000000000..faa8425c8afa04fe5e1e5fa4a3c641052d865d9c
--- /dev/null
+++ b/src/device/tests/gnmi_openconfig/request_composers.py
@@ -0,0 +1,44 @@
+# 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, 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, next_hop_index=0, 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': next_hop, 'next_hop_index': next_hop_index, '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
diff --git a/src/device/tests/gnmi_openconfig/storage.py b/src/device/tests/gnmi_openconfig/storage.py
new file mode 100644
index 0000000000000000000000000000000000000000..4271b002f5bb13f1b950779b3d49a2635f93ebab
--- /dev/null
+++ b/src/device/tests/gnmi_openconfig/storage.py
@@ -0,0 +1,285 @@
+# 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.
+
+import pytest, re
+from typing import Dict, List, Tuple
+
+@pytest.fixture(scope='session')
+def storage() -> Dict:
+ yield dict()
+
+
+##### POPULATE INTERFACE STORAGE #######################################################################################
+
+def populate_interfaces_storage(
+ storage : Dict, # pylint: disable=redefined-outer-name
+ resources : List[Tuple[str, Dict]],
+) -> None:
+ interfaces_storage : Dict = storage.setdefault('interfaces', dict())
+ subinterfaces_storage : Dict = storage.setdefault('interface_subinterfaces', dict())
+ ipv4_addresses_storage : Dict = storage.setdefault('interface_subinterface_ipv4_addresses', dict())
+
+ for resource_key, resource_value in resources:
+ match = re.match(r'^\/interface\[([^\]]+)\]$', resource_key)
+ if match is not None:
+ if_name = match.group(1)
+ if_storage = interfaces_storage.setdefault(if_name, dict())
+ if_storage['name' ] = if_name
+ if_storage['type' ] = resource_value.get('type' )
+ if_storage['admin-status' ] = resource_value.get('admin-status' )
+ if_storage['oper-status' ] = resource_value.get('oper-status' )
+ if_storage['ifindex' ] = resource_value.get('ifindex' )
+ if_storage['mtu' ] = resource_value.get('mtu' )
+ if_storage['management' ] = resource_value.get('management' )
+ if_storage['hardware-port'] = resource_value.get('hardware-port')
+ if_storage['transceiver' ] = resource_value.get('transceiver' )
+ continue
+
+ match = re.match(r'^\/interface\[([^\]]+)\]\/ethernet$', resource_key)
+ if match is not None:
+ if_name = match.group(1)
+ if_storage = interfaces_storage.setdefault(if_name, dict())
+ if_storage['port-speed' ] = resource_value.get('port-speed' )
+ if_storage['negotiated-port-speed'] = resource_value.get('negotiated-port-speed')
+ if_storage['mac-address' ] = resource_value.get('mac-address' )
+ if_storage['hw-mac-address' ] = resource_value.get('hw-mac-address' )
+ continue
+
+ match = re.match(r'^\/interface\[([^\]]+)\]\/subinterface\[([^\]]+)\]$', resource_key)
+ if match is not None:
+ if_name = match.group(1)
+ subif_index = int(match.group(2))
+ subif_storage = subinterfaces_storage.setdefault((if_name, subif_index), dict())
+ subif_storage['index'] = subif_index
+ continue
+
+ match = re.match(r'^\/interface\[([^\]]+)\]\/subinterface\[([^\]]+)\]\/ipv4\[([^\]]+)\]$', resource_key)
+ if match is not None:
+ if_name = match.group(1)
+ subif_index = int(match.group(2))
+ ipv4_addr = match.group(3)
+ ipv4_address_storage = ipv4_addresses_storage.setdefault((if_name, subif_index, ipv4_addr), dict())
+ ipv4_address_storage['ip' ] = ipv4_addr
+ ipv4_address_storage['origin'] = resource_value.get('origin')
+ ipv4_address_storage['prefix'] = resource_value.get('prefix')
+ continue
+
+
+##### POPULATE NETWORK INSTANCE STORAGE ################################################################################
+
+def populate_network_instances_storage(
+ storage : Dict, # pylint: disable=redefined-outer-name
+ resources : List[Tuple[str, Dict]],
+) -> None:
+ network_instances_storage : Dict = storage.setdefault('network_instances', dict())
+ network_instance_protocols_storage : Dict = storage.setdefault('network_instance_protocols', dict())
+ network_instance_protocol_static_storage : Dict = storage.setdefault('network_instance_protocol_static', dict())
+ network_instance_tables_storage : Dict = storage.setdefault('network_instance_tables', dict())
+ network_instance_vlans_storage : Dict = storage.setdefault('network_instance_vlans', dict())
+
+ for resource_key, resource_value in resources:
+ match = re.match(r'^\/network\_instance\[([^\]]+)\]$', resource_key)
+ if match is not None:
+ name = match.group(1)
+ ni_storage = network_instances_storage.setdefault(name, dict())
+ ni_storage['name'] = name
+ ni_storage['type'] = resource_value.get('type')
+ continue
+
+ match = re.match(r'^\/network\_instance\[([^\]]+)\]\/protocol\[([^\]]+)\]$', resource_key)
+ if match is not None:
+ name = match.group(1)
+ protocol = match.group(2)
+ ni_p_storage = network_instance_protocols_storage.setdefault((name, protocol), dict())
+ ni_p_storage['id' ] = protocol
+ ni_p_storage['name'] = protocol
+ continue
+
+ pattern = r'^\/network\_instance\[([^\]]+)\]\/protocol\[([^\]]+)\]\/static\_routes\[([^\]]+)\]$'
+ match = re.match(pattern, resource_key)
+ if match is not None:
+ name = match.group(1)
+ protocol = match.group(2)
+ prefix = match.group(3)
+ ni_p_s_storage = network_instance_protocol_static_storage.setdefault((name, protocol, prefix), dict())
+ ni_p_s_storage['prefix' ] = prefix
+ ni_p_s_storage['next_hops'] = sorted(resource_value.get('next_hops'))
+ continue
+
+ match = re.match(r'^\/network\_instance\[([^\]]+)\]\/table\[([^\,]+)\,([^\]]+)\]$', resource_key)
+ if match is not None:
+ name = match.group(1)
+ protocol = match.group(2)
+ address_family = match.group(3)
+ ni_t_storage = network_instance_tables_storage.setdefault((name, protocol, address_family), dict())
+ ni_t_storage['protocol' ] = protocol
+ ni_t_storage['address_family'] = address_family
+ continue
+
+ match = re.match(r'^\/network\_instance\[([^\]]+)\]\/vlan\[([^\]]+)\]$', resource_key)
+ if match is not None:
+ name = match.group(1)
+ vlan_id = int(match.group(2))
+ ni_v_storage = network_instance_vlans_storage.setdefault((name, vlan_id), dict())
+ ni_v_storage['vlan_id'] = vlan_id
+ ni_v_storage['name' ] = resource_value.get('name')
+ ni_v_storage['members'] = sorted(resource_value.get('members'))
+ continue
+
+
+##### GET EXPECTED INTERFACE CONFIG ####################################################################################
+
+INTERFACE_CONFIG_STRUCTURE : List[Tuple[str, List[str]]] = [
+ ('/interface[{if_name:s}]', [
+ 'name', 'type', 'admin-status', 'oper-status', 'management', 'mtu', 'ifindex', 'hardware-port', 'transceiver'
+ ]),
+ ('/interface[{if_name:s}]/ethernet', [
+ 'port-speed', 'negotiated-port-speed', 'mac-address', 'hw-mac-address'
+ ]),
+]
+
+INTERFACE_SUBINTERFACE_CONFIG_STRUCTURE : List[Tuple[str, List[str]]] = [
+ ('/interface[{if_name:s}]/subinterface[{subif_index:d}]', ['index']),
+]
+
+INTERFACE_SUBINTERFACE_IPV4_ADDRESS_CONFIG_STRUCTURE : List[Tuple[str, List[str]]] = [
+ ('/interface[{if_name:s}]/subinterface[{subif_index:d}]/ipv4[{ipv4_addr:s}]', ['ip', 'origin', 'prefix']),
+]
+
+def get_expected_interface_config(
+ storage : Dict, # pylint: disable=redefined-outer-name
+) -> List[Tuple[str, Dict]]:
+ interfaces_storage : Dict = storage.setdefault('interfaces', dict())
+ subinterfaces_storage : Dict = storage.setdefault('interface_subinterfaces', dict())
+ ipv4_addresses_storage : Dict = storage.setdefault('interface_subinterface_ipv4_addresses', dict())
+
+ expected_interface_config = list()
+ for if_name, if_storage in interfaces_storage.items():
+ for resource_key_template, resource_key_field_names in INTERFACE_CONFIG_STRUCTURE:
+ resource_key = resource_key_template.format(if_name=if_name)
+ resource_value = {
+ field_name : if_storage[field_name]
+ for field_name in resource_key_field_names
+ if field_name in if_storage and if_storage[field_name] is not None
+ }
+ expected_interface_config.append((resource_key, resource_value))
+
+ for (if_name, subif_index), subif_storage in subinterfaces_storage.items():
+ for resource_key_template, resource_key_field_names in INTERFACE_SUBINTERFACE_CONFIG_STRUCTURE:
+ resource_key = resource_key_template.format(if_name=if_name, subif_index=subif_index)
+ resource_value = {
+ field_name : subif_storage[field_name]
+ for field_name in resource_key_field_names
+ if field_name in subif_storage and subif_storage[field_name] is not None
+ }
+ expected_interface_config.append((resource_key, resource_value))
+
+ for (if_name, subif_index, ipv4_addr), ipv4_storage in ipv4_addresses_storage.items():
+ for resource_key_template, resource_key_field_names in INTERFACE_SUBINTERFACE_IPV4_ADDRESS_CONFIG_STRUCTURE:
+ resource_key = resource_key_template.format(if_name=if_name, subif_index=subif_index, ipv4_addr=ipv4_addr)
+ resource_value = {
+ field_name : ipv4_storage[field_name]
+ for field_name in resource_key_field_names
+ if field_name in ipv4_storage and ipv4_storage[field_name] is not None
+ }
+ expected_interface_config.append((resource_key, resource_value))
+
+ return expected_interface_config
+
+
+##### GET EXPECTED NETWORK INSTANCE CONFIG #############################################################################
+
+NETWORK_INSTANCE_CONFIG_STRUCTURE : List[Tuple[str, List[str]]] = [
+ ('/network_instance[{ni_name:s}]', ['name', 'type']),
+]
+
+NETWORK_INSTANCE_PROTOCOL_CONFIG_STRUCTURE : List[Tuple[str, List[str]]] = [
+ ('/network_instance[{ni_name:s}]/protocol[{protocol:s}]', ['id', 'name']),
+]
+
+NETWORK_INSTANCE_PROTOCOL_STATIC_CONFIG_STRUCTURE : List[Tuple[str, List[str]]] = [
+ ('/network_instance[{ni_name:s}]/protocol[{protocol:s}]/static_routes[{prefix:s}]', ['prefix', 'next_hops']),
+]
+
+NETWORK_INSTANCE_TABLE_CONFIG_STRUCTURE : List[Tuple[str, List[str]]] = [
+ ('/network_instance[{ni_name:s}]/table[{protocol:s},{address_family:s}]', ['protocol', 'address_family']),
+]
+
+NETWORK_INSTANCE_VLAN_CONFIG_STRUCTURE : List[Tuple[str, List[str]]] = [
+ ('/network_instance[{ni_name:s}]/vlan[{vlan_id:d}]', ['vlan_id', 'name', 'members']),
+]
+
+def get_expected_network_instance_config(
+ storage : Dict, # pylint: disable=redefined-outer-name
+) -> List[Tuple[str, Dict]]:
+ network_instances_storage : Dict = storage.setdefault('network_instances', dict())
+ network_instance_protocols_storage : Dict = storage.setdefault('network_instance_protocols', dict())
+ network_instance_protocol_static_storage : Dict = storage.setdefault('network_instance_protocol_static', dict())
+ network_instance_tables_storage : Dict = storage.setdefault('network_instance_tables', dict())
+ network_instance_vlans_storage : Dict = storage.setdefault('network_instance_vlans', dict())
+
+ expected_network_instance_config = list()
+ for ni_name, ni_storage in network_instances_storage.items():
+ for resource_key_template, resource_key_field_names in NETWORK_INSTANCE_CONFIG_STRUCTURE:
+ resource_key = resource_key_template.format(ni_name=ni_name)
+ resource_value = {
+ field_name : ni_storage[field_name]
+ for field_name in resource_key_field_names
+ if field_name in ni_storage and ni_storage[field_name] is not None
+ }
+ expected_network_instance_config.append((resource_key, resource_value))
+
+ for (ni_name, protocol), ni_p_storage in network_instance_protocols_storage.items():
+ for resource_key_template, resource_key_field_names in NETWORK_INSTANCE_PROTOCOL_CONFIG_STRUCTURE:
+ resource_key = resource_key_template.format(ni_name=ni_name, protocol=protocol)
+ resource_value = {
+ field_name : ni_p_storage[field_name]
+ for field_name in resource_key_field_names
+ if field_name in ni_p_storage and ni_p_storage[field_name] is not None
+ }
+ expected_network_instance_config.append((resource_key, resource_value))
+
+ for (ni_name, protocol, prefix), ni_p_s_storage in network_instance_protocol_static_storage.items():
+ for resource_key_template, resource_key_field_names in NETWORK_INSTANCE_PROTOCOL_STATIC_CONFIG_STRUCTURE:
+ resource_key = resource_key_template.format(ni_name=ni_name, protocol=protocol, prefix=prefix)
+ resource_value = {
+ field_name : ni_p_s_storage[field_name]
+ for field_name in resource_key_field_names
+ if field_name in ni_p_s_storage and ni_p_s_storage[field_name] is not None
+ }
+ expected_network_instance_config.append((resource_key, resource_value))
+
+ for (ni_name, protocol, address_family), ni_t_storage in network_instance_tables_storage.items():
+ for resource_key_template, resource_key_field_names in NETWORK_INSTANCE_TABLE_CONFIG_STRUCTURE:
+ resource_key = resource_key_template.format(
+ ni_name=ni_name, protocol=protocol, address_family=address_family
+ )
+ resource_value = {
+ field_name : ni_t_storage[field_name]
+ for field_name in resource_key_field_names
+ if field_name in ni_t_storage and ni_t_storage[field_name] is not None
+ }
+ expected_network_instance_config.append((resource_key, resource_value))
+
+ for (ni_name, vlan_id), ni_v_storage in network_instance_vlans_storage.items():
+ for resource_key_template, resource_key_field_names in NETWORK_INSTANCE_VLAN_CONFIG_STRUCTURE:
+ resource_key = resource_key_template.format(ni_name=ni_name, vlan_id=vlan_id)
+ resource_value = {
+ field_name : ni_v_storage[field_name]
+ for field_name in resource_key_field_names
+ if field_name in ni_v_storage and ni_v_storage[field_name] is not None
+ }
+ expected_network_instance_config.append((resource_key, resource_value))
+
+ return expected_network_instance_config
diff --git a/src/device/tests/gnmi_openconfig/test_unitary_gnmi_openconfig.py b/src/device/tests/gnmi_openconfig/test_unitary_gnmi_openconfig.py
new file mode 100644
index 0000000000000000000000000000000000000000..69b7a609ad40f6a3c57c315d3ef0a18a0e4bcdfc
--- /dev/null
+++ b/src/device/tests/gnmi_openconfig/test_unitary_gnmi_openconfig.py
@@ -0,0 +1,556 @@
+# 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.
+
+import deepdiff, logging, os, pytest, re, time
+from typing import Dict
+os.environ['DEVICE_EMULATED_ONLY'] = 'YES'
+
+# pylint: disable=wrong-import-position
+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 .request_composers import interface, network_instance, network_instance_interface, network_instance_static_route
+from .storage import ( # pylint: disable=unused-import
+ storage, # be careful, order of symbols is important here!; storage should be the first one
+ get_expected_interface_config, get_expected_network_instance_config, populate_interfaces_storage,
+ populate_network_instances_storage
+)
+
+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()
+
+
+##### NETWORK INSTANCE DETAILS #########################################################################################
+
+NI_NAME = 'test-l3-svc'
+NI_TYPE = 'L3VRF'
+NI_INTERFACES = [
+ # interface_name, subinterface_index, ipv4 address, ipv4 prefix, enabled
+ ('Ethernet1', 0, '192.168.1.1', 24, True),
+ ('Ethernet10', 0, '192.168.10.1', 24, True),
+]
+NI_STATIC_ROUTES = [
+ # prefix, gateway, metric
+ ('172.0.0.0/24', '172.16.0.2', 1),
+ ('172.2.0.0/24', '172.16.0.3', 1),
+]
+
+
+##### TEST METHODS #####################################################################################################
+
+def test_get_endpoints(
+ driver : GnmiOpenConfigDriver, # pylint: disable=redefined-outer-name
+) -> None:
+ resources_to_get = [RESOURCE_ENDPOINTS]
+ 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)))
+ expected_getconfig = [
+ ('/endpoints/endpoint[ethernet1]', {'uuid': 'ethernet1', 'type': '-', 'sample_types': {
+ 202: '/openconfig-interfaces:interfaces/interface[name=ethernet1]/state/counters/in-octets',
+ 201: '/openconfig-interfaces:interfaces/interface[name=ethernet1]/state/counters/out-octets',
+ 102: '/openconfig-interfaces:interfaces/interface[name=ethernet1]/state/counters/in-pkts',
+ 101: '/openconfig-interfaces:interfaces/interface[name=ethernet1]/state/counters/out-pkts'
+ }}),
+ ('/endpoints/endpoint[ethernet10]', {'uuid': 'ethernet10', 'type': '-', 'sample_types': {
+ 202: '/openconfig-interfaces:interfaces/interface[name=ethernet10]/state/counters/in-octets',
+ 201: '/openconfig-interfaces:interfaces/interface[name=ethernet10]/state/counters/out-octets',
+ 102: '/openconfig-interfaces:interfaces/interface[name=ethernet10]/state/counters/in-pkts',
+ 101: '/openconfig-interfaces:interfaces/interface[name=ethernet10]/state/counters/out-pkts'
+ }})
+ ]
+ diff_data = deepdiff.DeepDiff(sorted(expected_getconfig), sorted(results_getconfig))
+ num_diffs = len(diff_data)
+ if num_diffs > 0: LOGGER.error('Differences[{:d}]:\n{:s}'.format(num_diffs, str(diff_data.pretty())))
+ assert num_diffs == 0
+
+
+def test_get_interfaces(
+ driver : GnmiOpenConfigDriver, # pylint: disable=redefined-outer-name
+ storage : Dict, # pylint: disable=redefined-outer-name
+) -> None:
+ resources_to_get = [RESOURCE_INTERFACES]
+ 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)))
+
+ populate_interfaces_storage(storage, results_getconfig)
+ expected_getconfig = get_expected_interface_config(storage)
+
+ diff_data = deepdiff.DeepDiff(sorted(expected_getconfig), sorted(results_getconfig))
+ num_diffs = len(diff_data)
+ if num_diffs > 0: LOGGER.error('Differences[{:d}]:\n{:s}'.format(num_diffs, str(diff_data.pretty())))
+ assert num_diffs == 0
+
+
+def test_get_network_instances(
+ driver : GnmiOpenConfigDriver, # pylint: disable=redefined-outer-name
+ storage : Dict, # pylint: disable=redefined-outer-name
+) -> None:
+ resources_to_get = [RESOURCE_NETWORK_INSTANCES]
+ 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)))
+
+ populate_network_instances_storage(storage, results_getconfig)
+ expected_getconfig = get_expected_network_instance_config(storage)
+
+ for resource_key, resource_value in results_getconfig:
+ match = re.match(r'^\/network\_instance\[([^\]]+)\]\/vlan\[([^\]]+)\]$', resource_key)
+ if match is None: continue
+ members = resource_value.get('members')
+ if len(members) > 0: resource_value['members'] = sorted(members)
+
+ diff_data = deepdiff.DeepDiff(sorted(expected_getconfig), sorted(results_getconfig))
+ num_diffs = len(diff_data)
+ if num_diffs > 0: LOGGER.error('Differences[{:d}]:\n{:s}'.format(num_diffs, str(diff_data.pretty())))
+ assert num_diffs == 0
+ raise Exception()
+
+
+def test_set_network_instances(
+ driver : GnmiOpenConfigDriver, # pylint: disable=redefined-outer-name
+ storage : Dict, # pylint: disable=redefined-outer-name
+) -> None:
+ resources_to_get = [RESOURCE_NETWORK_INSTANCES]
+ 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(NI_NAME, NI_TYPE),
+ ]
+ 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)))
+
+ network_instances = sorted([NI_NAME])
+ results = set(results_setconfig)
+ assert len(results) == len(network_instances)
+ for ni_name in network_instances:
+ assert ('/network_instance[{:s}]'.format(ni_name), True) in results
+
+ expected_getconfig = get_expected_network_instance_config(storage)
+ expected_getconfig.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'
+ })
+ ])
+ #for resource_key, resource_value in expected_getconfig:
+ # if resource_key == '/network_instance[default]/vlan[1]':
+ # resource_value['members'] = list()
+ LOGGER.info('expected_getconfig = {:s}'.format(str(sorted(expected_getconfig))))
+
+ permitted_retries = 5
+ while permitted_retries > 0:
+ resources_to_get = [RESOURCE_NETWORK_INSTANCES]
+ 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)))
+
+ for resource_key, resource_value in results_getconfig:
+ match = re.match(r'^\/network\_instance\[([^\]]+)\]\/vlan\[([^\]]+)\]$', resource_key)
+ if match is None: continue
+ members = resource_value.get('members')
+ if len(members) > 0: resource_value['members'] = sorted(members)
+
+ diff_data = deepdiff.DeepDiff(sorted(expected_getconfig), sorted(results_getconfig))
+ num_diffs = len(diff_data)
+ if num_diffs == 0: break
+ # let the device take some time to reconfigure
+ time.sleep(0.5)
+ permitted_retries -= 1
+
+ if num_diffs > 0: LOGGER.error('Differences[{:d}]:\n{:s}'.format(num_diffs, str(diff_data.pretty())))
+ assert num_diffs == 0
+
+
+def test_set_interfaces(
+ driver : GnmiOpenConfigDriver, # pylint: disable=redefined-outer-name
+ storage : Dict, # pylint: disable=redefined-outer-name
+) -> None:
+ resources_to_get = [RESOURCE_INTERFACES]
+ 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 = [
+ interface(if_name, sif_index, ipv4_addr, ipv4_prefix, enabled)
+ for if_name, sif_index, ipv4_addr, ipv4_prefix, enabled in NI_INTERFACES
+ ]
+ 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)))
+
+ interfaces = sorted([
+ if_name
+ for if_name, _, _, _, _ in NI_INTERFACES
+ ])
+ results = set(results_setconfig)
+ assert len(results) == len(interfaces)
+ for if_name in interfaces:
+ assert ('/interface[{:s}]'.format(if_name), True) in results
+
+ expected_getconfig = get_expected_interface_config(storage)
+ expected_getconfig.extend([
+ ('/interface[{:s}]/subinterface[{:d}]/ipv4[{:s}]'.format(if_name, sif_index, ipv4_addr), {
+ 'ip': ipv4_addr, 'origin': 'STATIC', 'prefix': ipv4_prefix
+ })
+ for if_name, sif_index, ipv4_addr, ipv4_prefix, _ in NI_INTERFACES
+ ])
+
+ permitted_retries = 5
+ while permitted_retries > 0:
+ resources_to_get = [RESOURCE_INTERFACES]
+ 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)))
+
+ diff_data = deepdiff.DeepDiff(sorted(expected_getconfig), sorted(results_getconfig))
+ num_diffs = len(diff_data)
+ if num_diffs == 0: break
+ # let the device take some time to reconfigure
+ time.sleep(0.5)
+ permitted_retries -= 1
+
+ if num_diffs > 0: LOGGER.error('Differences[{:d}]:\n{:s}'.format(num_diffs, str(diff_data.pretty())))
+ assert num_diffs == 0
+
+
+def test_add_interfaces_to_network_instance(
+ driver : GnmiOpenConfigDriver, # pylint: disable=redefined-outer-name
+ storage : Dict, # pylint: disable=redefined-outer-name
+) -> None:
+ resources_to_get = [RESOURCE_INTERFACES]
+ 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_get = [RESOURCE_NETWORK_INSTANCES]
+ 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_interface(NI_NAME, if_name, sif_index)
+ for if_name, sif_index, _, _, _ in NI_INTERFACES
+ ]
+ 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)))
+
+ #interfaces = sorted(['Ethernet1', 'Ethernet10'])
+ #results = set(results_setconfig)
+ #assert len(results) == len(interfaces)
+ #for if_name in interfaces:
+ # assert ('/interface[{:s}]'.format(if_name), True) in results
+
+ #expected_getconfig = get_expected_interface_config(storage)
+ #expected_getconfig.extend([
+ # ('/interface[Ethernet1]/subinterface[0]/ipv4[192.168.1.1]', {
+ # 'ip': '192.168.1.1', 'origin': 'STATIC', 'prefix': 24
+ # }),
+ # ('/interface[Ethernet10]/subinterface[0]/ipv4[192.168.10.1]', {
+ # 'ip': '192.168.10.1', 'origin': 'STATIC', 'prefix': 24
+ # })
+ #])
+
+ #permitted_retries = 5
+ #while permitted_retries > 0:
+ # resources_to_get = [RESOURCE_INTERFACES]
+ # 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)))
+ #
+ # diff_data = deepdiff.DeepDiff(sorted(expected_getconfig), sorted(results_getconfig))
+ # num_diffs = len(diff_data)
+ # if num_diffs == 0: break
+ # # let the device take some time to reconfigure
+ # time.sleep(0.5)
+ # permitted_retries -= 1
+
+ #if num_diffs > 0: LOGGER.error('Differences[{:d}]:\n{:s}'.format(num_diffs, str(diff_data.pretty())))
+ #assert num_diffs == 0
+ raise Exception()
+
+
+def test_set_network_instance_static_routes(
+ driver : GnmiOpenConfigDriver, # pylint: disable=redefined-outer-name
+ storage : Dict, # pylint: disable=redefined-outer-name
+) -> None:
+ resources_to_get = [RESOURCE_NETWORK_INSTANCES]
+ 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_static_route(NI_NAME, prefix, gateway, metric=metric)
+ for prefix, gateway, metric in NI_STATIC_ROUTES
+ ]
+ 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)))
+
+
+
+
+
+ #interfaces = sorted(['Ethernet1', 'Ethernet10'])
+ #results = set(results_setconfig)
+ #assert len(results) == len(interfaces)
+ #for if_name in interfaces:
+ # assert ('/interface[{:s}]'.format(if_name), True) in results
+
+ #expected_getconfig = get_expected_interface_config(storage)
+ #expected_getconfig.extend([
+ # ('/interface[Ethernet1]/subinterface[0]/ipv4[192.168.1.1]', {
+ # 'ip': '192.168.1.1', 'origin': 'STATIC', 'prefix': 24
+ # }),
+ # ('/interface[Ethernet10]/subinterface[0]/ipv4[192.168.10.1]', {
+ # 'ip': '192.168.10.1', 'origin': 'STATIC', 'prefix': 24
+ # })
+ #])
+
+ #permitted_retries = 5
+ #while permitted_retries > 0:
+ # resources_to_get = [RESOURCE_INTERFACES]
+ # 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)))
+ #
+ # diff_data = deepdiff.DeepDiff(sorted(expected_getconfig), sorted(results_getconfig))
+ # num_diffs = len(diff_data)
+ # if num_diffs == 0: break
+ # # let the device take some time to reconfigure
+ # time.sleep(0.5)
+ # permitted_retries -= 1
+ #
+ #if num_diffs > 0: LOGGER.error('Differences[{:d}]:\n{:s}'.format(num_diffs, str(diff_data.pretty())))
+ #assert num_diffs == 0
+ raise Exception()
+
+
+def test_del_network_instance_static_routes(
+ driver : GnmiOpenConfigDriver, # pylint: disable=redefined-outer-name
+ storage : Dict, # pylint: disable=redefined-outer-name
+) -> None:
+
+ resources_to_get = [RESOURCE_NETWORK_INSTANCES]
+ 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_delete = [
+ network_instance_static_route(NI_NAME, '172.0.0.0/24', '172.16.0.2'),
+ network_instance_static_route(NI_NAME, '172.2.0.0/24', '172.16.0.3'),
+ ]
+ 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)))
+
+ #interfaces = sorted(['Ethernet1', 'Ethernet10'])
+ #results = set(results_deleteconfig)
+ #assert len(results) == len(interfaces)
+ #for if_name in interfaces:
+ # assert ('/interface[{:s}]'.format(if_name), True) in results
+
+ resources_to_get = [RESOURCE_NETWORK_INSTANCES]
+ 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)))
+
+ #expected_getconfig = get_expected_interface_config(storage)
+
+ #diff_data = deepdiff.DeepDiff(sorted(expected_getconfig), sorted(results_getconfig))
+ #num_diffs = len(diff_data)
+ #if num_diffs > 0: LOGGER.error('Differences[{:d}]:\n{:s}'.format(num_diffs, str(diff_data.pretty())))
+ #assert num_diffs == 0
+ raise Exception()
+
+
+def test_del_interfaces_from_network_instance(
+ driver : GnmiOpenConfigDriver, # pylint: disable=redefined-outer-name
+ storage : Dict, # pylint: disable=redefined-outer-name
+) -> None:
+
+ resources_to_get = [RESOURCE_INTERFACES]
+ 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_get = [RESOURCE_NETWORK_INSTANCES]
+ 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_delete = [
+ network_instance_interface(NI_NAME, ni_if_name, ni_sif_index)
+ for ni_if_name, ni_sif_index in NI_INTERFACES
+ ]
+ 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)))
+
+ interface_ids = sorted([
+ '{:s}.{:d}'.format(ni_if_name, ni_sif_index)
+ for ni_if_name, ni_sif_index in NI_INTERFACES
+ ])
+ results = set(results_deleteconfig)
+ assert len(results) == len(interface_ids)
+ for interface_id in interface_ids:
+ assert ('/network_instance[{:s}]/interface[{:s}]'.format(NI_NAME, interface_id), True) in results
+
+ resources_to_get = [RESOURCE_INTERFACES]
+ 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)))
+
+ expected_getconfig = get_expected_interface_config(storage)
+ expected_getconfig.extend([
+ ('/interface[Ethernet1]/subinterface[0]/ipv4[192.168.1.1]', {
+ 'ip': '192.168.1.1', 'origin': 'STATIC', 'prefix': 24
+ }),
+ ('/interface[Ethernet10]/subinterface[0]/ipv4[192.168.10.1]', {
+ 'ip': '192.168.10.1', 'origin': 'STATIC', 'prefix': 24
+ })
+ ])
+
+ diff_data = deepdiff.DeepDiff(sorted(expected_getconfig), sorted(results_getconfig))
+ num_diffs = len(diff_data)
+ if num_diffs > 0: LOGGER.error('Differences[{:d}]:\n{:s}'.format(num_diffs, str(diff_data.pretty())))
+ assert num_diffs == 0
+
+ resources_to_get = [RESOURCE_NETWORK_INSTANCES]
+ 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)))
+
+ expected_getconfig = get_expected_network_instance_config(storage)
+
+ diff_data = deepdiff.DeepDiff(sorted(expected_getconfig), sorted(results_getconfig))
+ num_diffs = len(diff_data)
+ if num_diffs > 0: LOGGER.error('Differences[{:d}]:\n{:s}'.format(num_diffs, str(diff_data.pretty())))
+ assert num_diffs == 0
+ raise Exception()
+
+
+def test_del_interfaces(
+ driver : GnmiOpenConfigDriver, # pylint: disable=redefined-outer-name
+ storage : Dict, # pylint: disable=redefined-outer-name
+) -> None:
+ resources_to_get = [RESOURCE_INTERFACES]
+ 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_delete = [
+ interface('Ethernet1', 0, '192.168.1.1', 24, True),
+ interface('Ethernet10', 0, '192.168.10.1', 24, True),
+ ]
+ 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)))
+
+ interfaces = sorted(['Ethernet1', 'Ethernet10'])
+ results = set(results_deleteconfig)
+ assert len(results) == len(interfaces)
+ for if_name in interfaces:
+ assert ('/interface[{:s}]'.format(if_name), True) in results
+
+ resources_to_get = [RESOURCE_INTERFACES]
+ 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)))
+
+ expected_getconfig = get_expected_interface_config(storage)
+
+ diff_data = deepdiff.DeepDiff(sorted(expected_getconfig), sorted(results_getconfig))
+ num_diffs = len(diff_data)
+ if num_diffs > 0: LOGGER.error('Differences[{:d}]:\n{:s}'.format(num_diffs, str(diff_data.pretty())))
+ assert num_diffs == 0
+
+
+def test_del_network_instances(
+ driver : GnmiOpenConfigDriver, # pylint: disable=redefined-outer-name
+ storage : Dict, # pylint: disable=redefined-outer-name
+) -> None:
+ resources_to_get = [RESOURCE_NETWORK_INSTANCES]
+ 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_delete = [
+ network_instance(NI_NAME, '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)))
+
+ network_instances = sorted([NI_NAME])
+ results = set(results_deleteconfig)
+ assert len(results) == len(network_instances)
+ for ni_name in network_instances:
+ assert ('/network_instance[{:s}]'.format(ni_name), True) in results
+
+ resources_to_get = [RESOURCE_NETWORK_INSTANCES]
+ 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)))
+
+ for resource_key, resource_value in results_getconfig:
+ match = re.match(r'^\/network\_instance\[([^\]]+)\]\/vlan\[([^\]]+)\]$', resource_key)
+ if match is None: continue
+ members = resource_value.get('members')
+ if len(members) > 0: resource_value['members'] = sorted(members)
+
+ expected_getconfig = get_expected_network_instance_config(storage)
+
+ diff_data = deepdiff.DeepDiff(sorted(expected_getconfig), sorted(results_getconfig))
+ num_diffs = len(diff_data)
+ if num_diffs > 0: LOGGER.error('Differences[{:d}]:\n{:s}'.format(num_diffs, str(diff_data.pretty())))
+ assert num_diffs == 0
diff --git a/src/device/tests/test_unitary_gnmi_openconfig.py b/src/device/tests/test_unitary_gnmi_openconfig.py
deleted file mode 100644
index 7d33d1a71e6a999ebdc549d327c27d4357c476d3..0000000000000000000000000000000000000000
--- a/src/device/tests/test_unitary_gnmi_openconfig.py
+++ /dev/null
@@ -1,637 +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.
-
-import deepdiff, logging, os, pytest, re, time
-from typing import Dict, List, Tuple
-os.environ['DEVICE_EMULATED_ONLY'] = 'YES'
-
-# pylint: disable=wrong-import-position
-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
-
-logging.basicConfig(level=logging.DEBUG)
-LOGGER = logging.getLogger(__name__)
-LOGGER.setLevel(logging.DEBUG)
-
-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()
-
-@pytest.fixture(scope='session')
-def storage() -> Dict:
- yield dict()
-
-
-##### STORAGE POPULATORS ###############################################################################################
-
-def populate_interfaces_storage(
- storage : Dict, # pylint: disable=redefined-outer-name
- resources : List[Tuple[str, Dict]],
-) -> None:
- interfaces_storage : Dict = storage.setdefault('interfaces', dict())
- subinterfaces_storage : Dict = storage.setdefault('interface_subinterfaces', dict())
- ipv4_addresses_storage : Dict = storage.setdefault('interface_subinterface_ipv4_addresses', dict())
-
- for resource_key, resource_value in resources:
- match = re.match(r'^\/interface\[([^\]]+)\]$', resource_key)
- if match is not None:
- if_name = match.group(1)
- if_storage = interfaces_storage.setdefault(if_name, dict())
- if_storage['name' ] = if_name
- if_storage['type' ] = resource_value.get('type' )
- if_storage['admin-status' ] = resource_value.get('admin-status' )
- if_storage['oper-status' ] = resource_value.get('oper-status' )
- if_storage['ifindex' ] = resource_value.get('ifindex' )
- if_storage['mtu' ] = resource_value.get('mtu' )
- if_storage['management' ] = resource_value.get('management' )
- if_storage['hardware-port'] = resource_value.get('hardware-port')
- if_storage['transceiver' ] = resource_value.get('transceiver' )
- continue
-
- match = re.match(r'^\/interface\[([^\]]+)\]\/ethernet$', resource_key)
- if match is not None:
- if_name = match.group(1)
- if_storage = interfaces_storage.setdefault(if_name, dict())
- if_storage['port-speed' ] = resource_value.get('port-speed' )
- if_storage['negotiated-port-speed'] = resource_value.get('negotiated-port-speed')
- if_storage['mac-address' ] = resource_value.get('mac-address' )
- if_storage['hw-mac-address' ] = resource_value.get('hw-mac-address' )
- continue
-
- match = re.match(r'^\/interface\[([^\]]+)\]\/subinterface\[([^\]]+)\]$', resource_key)
- if match is not None:
- if_name = match.group(1)
- subif_index = int(match.group(2))
- subif_storage = subinterfaces_storage.setdefault((if_name, subif_index), dict())
- subif_storage['index'] = subif_index
- continue
-
- match = re.match(r'^\/interface\[([^\]]+)\]\/subinterface\[([^\]]+)\]\/ipv4\[([^\]]+)\]$', resource_key)
- if match is not None:
- if_name = match.group(1)
- subif_index = int(match.group(2))
- ipv4_addr = match.group(3)
- ipv4_address_storage = ipv4_addresses_storage.setdefault((if_name, subif_index, ipv4_addr), dict())
- ipv4_address_storage['ip' ] = ipv4_addr
- ipv4_address_storage['origin'] = resource_value.get('origin')
- ipv4_address_storage['prefix'] = resource_value.get('prefix')
- continue
-
-def populate_network_instances_storage(
- storage : Dict, # pylint: disable=redefined-outer-name
- resources : List[Tuple[str, Dict]],
-) -> None:
- network_instances_storage : Dict = storage.setdefault('network_instances', dict())
- network_instance_protocols_storage : Dict = storage.setdefault('network_instance_protocols', dict())
- network_instance_protocol_static_storage : Dict = storage.setdefault('network_instance_protocol_static', dict())
- network_instance_tables_storage : Dict = storage.setdefault('network_instance_tables', dict())
- network_instance_vlans_storage : Dict = storage.setdefault('network_instance_vlans', dict())
-
- for resource_key, resource_value in resources:
- match = re.match(r'^\/network\_instance\[([^\]]+)\]$', resource_key)
- if match is not None:
- name = match.group(1)
- ni_storage = network_instances_storage.setdefault(name, dict())
- ni_storage['name'] = name
- ni_storage['type'] = resource_value.get('type')
- continue
-
- match = re.match(r'^\/network\_instance\[([^\]]+)\]\/protocol\[([^\]]+)\]$', resource_key)
- if match is not None:
- name = match.group(1)
- protocol = match.group(2)
- ni_p_storage = network_instance_protocols_storage.setdefault((name, protocol), dict())
- ni_p_storage['id' ] = protocol
- ni_p_storage['name'] = protocol
- continue
-
- pattern = r'^\/network\_instance\[([^\]]+)\]\/protocol\[([^\]]+)\]\/static\_routes\[([^\]]+)\]$'
- match = re.match(pattern, resource_key)
- if match is not None:
- name = match.group(1)
- protocol = match.group(2)
- prefix = match.group(3)
- ni_p_s_storage = network_instance_protocol_static_storage.setdefault((name, protocol, prefix), dict())
- ni_p_s_storage['prefix' ] = prefix
- ni_p_s_storage['next_hops'] = sorted(resource_value.get('next_hops'))
- continue
-
- match = re.match(r'^\/network\_instance\[([^\]]+)\]\/table\[([^\,]+)\,([^\]]+)\]$', resource_key)
- if match is not None:
- name = match.group(1)
- protocol = match.group(2)
- address_family = match.group(3)
- ni_t_storage = network_instance_tables_storage.setdefault((name, protocol, address_family), dict())
- ni_t_storage['protocol' ] = protocol
- ni_t_storage['address_family'] = address_family
- continue
-
- match = re.match(r'^\/network\_instance\[([^\]]+)\]\/vlan\[([^\]]+)\]$', resource_key)
- if match is not None:
- name = match.group(1)
- vlan_id = int(match.group(2))
- ni_v_storage = network_instance_vlans_storage.setdefault((name, vlan_id), dict())
- ni_v_storage['vlan_id'] = vlan_id
- ni_v_storage['name' ] = resource_value.get('name')
- ni_v_storage['members'] = sorted(resource_value.get('members'))
- continue
-
-
-##### EXPECTED CONFIG COMPOSERS ########################################################################################
-
-INTERFACE_CONFIG_STRUCTURE : List[Tuple[str, List[str]]] = [
- ('/interface[{if_name:s}]', [
- 'name', 'type', 'admin-status', 'oper-status', 'management', 'mtu', 'ifindex', 'hardware-port', 'transceiver'
- ]),
- ('/interface[{if_name:s}]/ethernet', [
- 'port-speed', 'negotiated-port-speed', 'mac-address', 'hw-mac-address'
- ]),
-]
-
-INTERFACE_SUBINTERFACE_CONFIG_STRUCTURE : List[Tuple[str, List[str]]] = [
- ('/interface[{if_name:s}]/subinterface[{subif_index:d}]', ['index']),
-]
-
-INTERFACE_SUBINTERFACE_IPV4_ADDRESS_CONFIG_STRUCTURE : List[Tuple[str, List[str]]] = [
- ('/interface[{if_name:s}]/subinterface[{subif_index:d}]/ipv4[{ipv4_addr:s}]', ['ip', 'origin', 'prefix']),
-]
-
-def get_expected_interface_config(
- storage : Dict, # pylint: disable=redefined-outer-name
-) -> List[Tuple[str, Dict]]:
- interfaces_storage : Dict = storage.setdefault('interfaces', dict())
- subinterfaces_storage : Dict = storage.setdefault('interface_subinterfaces', dict())
- ipv4_addresses_storage : Dict = storage.setdefault('interface_subinterface_ipv4_addresses', dict())
-
- expected_interface_config = list()
- for if_name, if_storage in interfaces_storage.items():
- for resource_key_template, resource_key_field_names in INTERFACE_CONFIG_STRUCTURE:
- resource_key = resource_key_template.format(if_name=if_name)
- resource_value = {
- field_name : if_storage[field_name]
- for field_name in resource_key_field_names
- if field_name in if_storage and if_storage[field_name] is not None
- }
- expected_interface_config.append((resource_key, resource_value))
-
- for (if_name, subif_index), subif_storage in subinterfaces_storage.items():
- for resource_key_template, resource_key_field_names in INTERFACE_SUBINTERFACE_CONFIG_STRUCTURE:
- resource_key = resource_key_template.format(if_name=if_name, subif_index=subif_index)
- resource_value = {
- field_name : subif_storage[field_name]
- for field_name in resource_key_field_names
- if field_name in subif_storage and subif_storage[field_name] is not None
- }
- expected_interface_config.append((resource_key, resource_value))
-
- for (if_name, subif_index, ipv4_addr), ipv4_storage in ipv4_addresses_storage.items():
- for resource_key_template, resource_key_field_names in INTERFACE_SUBINTERFACE_IPV4_ADDRESS_CONFIG_STRUCTURE:
- resource_key = resource_key_template.format(if_name=if_name, subif_index=subif_index, ipv4_addr=ipv4_addr)
- resource_value = {
- field_name : ipv4_storage[field_name]
- for field_name in resource_key_field_names
- if field_name in ipv4_storage and ipv4_storage[field_name] is not None
- }
- expected_interface_config.append((resource_key, resource_value))
-
- return expected_interface_config
-
-NETWORK_INSTANCE_CONFIG_STRUCTURE : List[Tuple[str, List[str]]] = [
- ('/network_instance[{ni_name:s}]', ['name', 'type']),
-]
-
-NETWORK_INSTANCE_PROTOCOL_CONFIG_STRUCTURE : List[Tuple[str, List[str]]] = [
- ('/network_instance[{ni_name:s}]/protocol[{protocol:s}]', ['id', 'name']),
-]
-
-NETWORK_INSTANCE_PROTOCOL_STATIC_CONFIG_STRUCTURE : List[Tuple[str, List[str]]] = [
- ('/network_instance[{ni_name:s}]/protocol[{protocol:s}]/static_routes[{prefix:s}]', ['prefix', 'next_hops']),
-]
-
-NETWORK_INSTANCE_TABLE_CONFIG_STRUCTURE : List[Tuple[str, List[str]]] = [
- ('/network_instance[{ni_name:s}]/table[{protocol:s},{address_family:s}]', ['protocol', 'address_family']),
-]
-
-NETWORK_INSTANCE_VLAN_CONFIG_STRUCTURE : List[Tuple[str, List[str]]] = [
- ('/network_instance[{ni_name:s}]/vlan[{vlan_id:d}]', ['vlan_id', 'name', 'members']),
-]
-
-def get_expected_network_instance_config(
- storage : Dict, # pylint: disable=redefined-outer-name
-) -> List[Tuple[str, Dict]]:
- network_instances_storage : Dict = storage.setdefault('network_instances', dict())
- network_instance_protocols_storage : Dict = storage.setdefault('network_instance_protocols', dict())
- network_instance_protocol_static_storage : Dict = storage.setdefault('network_instance_protocol_static', dict())
- network_instance_tables_storage : Dict = storage.setdefault('network_instance_tables', dict())
- network_instance_vlans_storage : Dict = storage.setdefault('network_instance_vlans', dict())
-
- expected_network_instance_config = list()
- for ni_name, ni_storage in network_instances_storage.items():
- for resource_key_template, resource_key_field_names in NETWORK_INSTANCE_CONFIG_STRUCTURE:
- resource_key = resource_key_template.format(ni_name=ni_name)
- resource_value = {
- field_name : ni_storage[field_name]
- for field_name in resource_key_field_names
- if field_name in ni_storage and ni_storage[field_name] is not None
- }
- expected_network_instance_config.append((resource_key, resource_value))
-
- for (ni_name, protocol), ni_p_storage in network_instance_protocols_storage.items():
- for resource_key_template, resource_key_field_names in NETWORK_INSTANCE_PROTOCOL_CONFIG_STRUCTURE:
- resource_key = resource_key_template.format(ni_name=ni_name, protocol=protocol)
- resource_value = {
- field_name : ni_p_storage[field_name]
- for field_name in resource_key_field_names
- if field_name in ni_p_storage and ni_p_storage[field_name] is not None
- }
- expected_network_instance_config.append((resource_key, resource_value))
-
- for (ni_name, protocol, prefix), ni_p_s_storage in network_instance_protocol_static_storage.items():
- for resource_key_template, resource_key_field_names in NETWORK_INSTANCE_PROTOCOL_STATIC_CONFIG_STRUCTURE:
- resource_key = resource_key_template.format(ni_name=ni_name, protocol=protocol, prefix=prefix)
- resource_value = {
- field_name : ni_p_s_storage[field_name]
- for field_name in resource_key_field_names
- if field_name in ni_p_s_storage and ni_p_s_storage[field_name] is not None
- }
- expected_network_instance_config.append((resource_key, resource_value))
-
- for (ni_name, protocol, address_family), ni_t_storage in network_instance_tables_storage.items():
- for resource_key_template, resource_key_field_names in NETWORK_INSTANCE_TABLE_CONFIG_STRUCTURE:
- resource_key = resource_key_template.format(
- ni_name=ni_name, protocol=protocol, address_family=address_family
- )
- resource_value = {
- field_name : ni_t_storage[field_name]
- for field_name in resource_key_field_names
- if field_name in ni_t_storage and ni_t_storage[field_name] is not None
- }
- expected_network_instance_config.append((resource_key, resource_value))
-
- for (ni_name, vlan_id), ni_v_storage in network_instance_vlans_storage.items():
- for resource_key_template, resource_key_field_names in NETWORK_INSTANCE_VLAN_CONFIG_STRUCTURE:
- resource_key = resource_key_template.format(ni_name=ni_name, vlan_id=vlan_id)
- resource_value = {
- field_name : ni_v_storage[field_name]
- for field_name in resource_key_field_names
- if field_name in ni_v_storage and ni_v_storage[field_name] is not None
- }
- expected_network_instance_config.append((resource_key, resource_value))
-
- return expected_network_instance_config
-
-
-##### REQUEST COMPOSERS ################################################################################################
-
-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 test_get_endpoints(
- driver : GnmiOpenConfigDriver, # pylint: disable=redefined-outer-name
-) -> None:
- resources_to_get = [RESOURCE_ENDPOINTS]
- 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)))
- expected_getconfig = [
- ('/endpoints/endpoint[ethernet1]', {'uuid': 'ethernet1', 'type': '-', 'sample_types': {
- 202: '/openconfig-interfaces:interfaces/interface[name=ethernet1]/state/counters/in-octets',
- 201: '/openconfig-interfaces:interfaces/interface[name=ethernet1]/state/counters/out-octets',
- 102: '/openconfig-interfaces:interfaces/interface[name=ethernet1]/state/counters/in-pkts',
- 101: '/openconfig-interfaces:interfaces/interface[name=ethernet1]/state/counters/out-pkts'
- }}),
- ('/endpoints/endpoint[ethernet10]', {'uuid': 'ethernet10', 'type': '-', 'sample_types': {
- 202: '/openconfig-interfaces:interfaces/interface[name=ethernet10]/state/counters/in-octets',
- 201: '/openconfig-interfaces:interfaces/interface[name=ethernet10]/state/counters/out-octets',
- 102: '/openconfig-interfaces:interfaces/interface[name=ethernet10]/state/counters/in-pkts',
- 101: '/openconfig-interfaces:interfaces/interface[name=ethernet10]/state/counters/out-pkts'
- }})
- ]
- diff_data = deepdiff.DeepDiff(sorted(expected_getconfig), sorted(results_getconfig))
- num_diffs = len(diff_data)
- if num_diffs > 0: LOGGER.error('Differences[{:d}]:\n{:s}'.format(num_diffs, str(diff_data.pretty())))
- assert num_diffs == 0
-
-def test_get_interfaces(
- driver : GnmiOpenConfigDriver, # pylint: disable=redefined-outer-name
- storage : Dict, # pylint: disable=redefined-outer-name
-) -> None:
- resources_to_get = [RESOURCE_INTERFACES]
- 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)))
-
- populate_interfaces_storage(storage, results_getconfig)
- expected_getconfig = get_expected_interface_config(storage)
-
- diff_data = deepdiff.DeepDiff(sorted(expected_getconfig), sorted(results_getconfig))
- num_diffs = len(diff_data)
- if num_diffs > 0: LOGGER.error('Differences[{:d}]:\n{:s}'.format(num_diffs, str(diff_data.pretty())))
- assert num_diffs == 0
-
-def test_get_network_instances(
- driver : GnmiOpenConfigDriver, # pylint: disable=redefined-outer-name
- storage : Dict, # pylint: disable=redefined-outer-name
-) -> None:
- resources_to_get = [RESOURCE_NETWORK_INSTANCES]
- 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)))
-
- populate_network_instances_storage(storage, results_getconfig)
- expected_getconfig = get_expected_network_instance_config(storage)
-
- for resource_key, resource_value in results_getconfig:
- match = re.match(r'^\/network\_instance\[([^\]]+)\]\/vlan\[([^\]]+)\]$', resource_key)
- if match is None: continue
- members = resource_value.get('members')
- if len(members) > 0: resource_value['members'] = sorted(members)
-
- diff_data = deepdiff.DeepDiff(sorted(expected_getconfig), sorted(results_getconfig))
- num_diffs = len(diff_data)
- if num_diffs > 0: LOGGER.error('Differences[{:d}]:\n{:s}'.format(num_diffs, str(diff_data.pretty())))
- assert num_diffs == 0
-
-def test_set_interfaces(
- driver : GnmiOpenConfigDriver, # pylint: disable=redefined-outer-name
- storage : Dict, # pylint: disable=redefined-outer-name
-) -> None:
- resources_to_get = [RESOURCE_INTERFACES]
- 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 = [
- interface('Ethernet1', 0, '192.168.1.1', 24, True),
- interface('Ethernet10', 0, '192.168.10.1', 24, True),
- ]
- 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)))
-
- interfaces = sorted(['Ethernet1', 'Ethernet10'])
- results = set(results_setconfig)
- assert len(results) == len(interfaces)
- for if_name in interfaces:
- assert ('/interface[{:s}]'.format(if_name), True) in results
-
- expected_getconfig = get_expected_interface_config(storage)
- expected_getconfig.extend([
- ('/interface[Ethernet1]/subinterface[0]/ipv4[192.168.1.1]', {
- 'ip': '192.168.1.1', 'origin': 'STATIC', 'prefix': 24
- }),
- ('/interface[Ethernet10]/subinterface[0]/ipv4[192.168.10.1]', {
- 'ip': '192.168.10.1', 'origin': 'STATIC', 'prefix': 24
- })
- ])
-
- permitted_retries = 5
- while permitted_retries > 0:
- resources_to_get = [RESOURCE_INTERFACES]
- 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)))
-
- diff_data = deepdiff.DeepDiff(sorted(expected_getconfig), sorted(results_getconfig))
- num_diffs = len(diff_data)
- if num_diffs == 0: break
- # let the device take some time to reconfigure
- time.sleep(0.5)
- permitted_retries -= 1
-
- if num_diffs > 0: LOGGER.error('Differences[{:d}]:\n{:s}'.format(num_diffs, str(diff_data.pretty())))
- assert num_diffs == 0
-
-def test_set_network_instances(
- driver : GnmiOpenConfigDriver, # pylint: disable=redefined-outer-name
- storage : Dict, # pylint: disable=redefined-outer-name
-) -> None:
- resources_to_get = [RESOURCE_NETWORK_INSTANCES]
- 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-l3-svc', 'L3VRF'),
- ]
- 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)))
-
- network_instances = sorted(['test-l3-svc'])
- results = set(results_setconfig)
- assert len(results) == len(network_instances)
- for ni_name in network_instances:
- assert ('/network_instance[{:s}]'.format(ni_name), True) in results
-
- expected_getconfig = get_expected_network_instance_config(storage)
- expected_getconfig.extend([
- ('/network_instance[test-l3-svc]', {
- 'name': 'test-l3-svc', 'type': 'L3VRF'
- }),
- ('/network_instance[test-l3-svc]/protocol[DIRECTLY_CONNECTED]', {
- 'id': 'DIRECTLY_CONNECTED', 'name': 'DIRECTLY_CONNECTED'
- }),
- ('/network_instance[test-l3-svc]/table[DIRECTLY_CONNECTED,IPV4]', {
- 'protocol': 'DIRECTLY_CONNECTED', 'address_family': 'IPV4'
- }),
- ('/network_instance[test-l3-svc]/table[DIRECTLY_CONNECTED,IPV6]', {
- 'protocol': 'DIRECTLY_CONNECTED', 'address_family': 'IPV6'
- })
- ])
- for resource_key, resource_value in expected_getconfig:
- if resource_key == '/network_instance[default]/vlan[1]':
- resource_value['members'] = list()
- LOGGER.info('expected_getconfig = {:s}'.format(str(sorted(expected_getconfig))))
-
- permitted_retries = 5
- while permitted_retries > 0:
- resources_to_get = [RESOURCE_NETWORK_INSTANCES]
- 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)))
-
- for resource_key, resource_value in results_getconfig:
- match = re.match(r'^\/network\_instance\[([^\]]+)\]\/vlan\[([^\]]+)\]$', resource_key)
- if match is None: continue
- members = resource_value.get('members')
- if len(members) > 0: resource_value['members'] = sorted(members)
-
- diff_data = deepdiff.DeepDiff(sorted(expected_getconfig), sorted(results_getconfig))
- num_diffs = len(diff_data)
- if num_diffs == 0: break
- # let the device take some time to reconfigure
- time.sleep(0.5)
- permitted_retries -= 1
-
- if num_diffs > 0: LOGGER.error('Differences[{:d}]:\n{:s}'.format(num_diffs, str(diff_data.pretty())))
- assert num_diffs == 0
-
-def test_del_interfaces(
- driver : GnmiOpenConfigDriver, # pylint: disable=redefined-outer-name
- storage : Dict, # pylint: disable=redefined-outer-name
-) -> None:
- resources_to_get = [RESOURCE_INTERFACES]
- 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_delete = [
- interface('Ethernet1', 0, '192.168.1.1', 24, True),
- interface('Ethernet10', 0, '192.168.10.1', 24, True),
- ]
- 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)))
-
- interfaces = sorted(['Ethernet1', 'Ethernet10'])
- results = set(results_deleteconfig)
- assert len(results) == len(interfaces)
- for if_name in interfaces:
- assert ('/interface[{:s}]'.format(if_name), True) in results
-
- resources_to_get = [RESOURCE_INTERFACES]
- 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)))
-
- expected_getconfig = get_expected_interface_config(storage)
-
- diff_data = deepdiff.DeepDiff(sorted(expected_getconfig), sorted(results_getconfig))
- num_diffs = len(diff_data)
- if num_diffs > 0: LOGGER.error('Differences[{:d}]:\n{:s}'.format(num_diffs, str(diff_data.pretty())))
- assert num_diffs == 0
-
-def test_del_network_instances(
- driver : GnmiOpenConfigDriver, # pylint: disable=redefined-outer-name
- storage : Dict, # pylint: disable=redefined-outer-name
-) -> None:
- resources_to_get = [RESOURCE_NETWORK_INSTANCES]
- 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_delete = [
- network_instance('test-l3-svc', '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)))
-
- network_instances = sorted(['test-l3-svc'])
- results = set(results_deleteconfig)
- assert len(results) == len(network_instances)
- for ni_name in network_instances:
- assert ('/network_instance[{:s}]'.format(ni_name), True) in results
-
- resources_to_get = [RESOURCE_NETWORK_INSTANCES]
- 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)))
-
- for resource_key, resource_value in results_getconfig:
- match = re.match(r'^\/network\_instance\[([^\]]+)\]\/vlan\[([^\]]+)\]$', resource_key)
- if match is None: continue
- members = resource_value.get('members')
- if len(members) > 0: resource_value['members'] = sorted(members)
-
- expected_getconfig = get_expected_network_instance_config(storage)
-
- diff_data = deepdiff.DeepDiff(sorted(expected_getconfig), sorted(results_getconfig))
- num_diffs = len(diff_data)
- if num_diffs > 0: LOGGER.error('Differences[{:d}]:\n{:s}'.format(num_diffs, str(diff_data.pretty())))
- assert num_diffs == 0
-
-
-#def test_unitary_gnmi_openconfig(
-# driver : GnmiOpenConfigDriver, # pylint: disable=redefined-outer-name
-#) -> None:
-# #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)))