Skip to content
Snippets Groups Projects
Commit 32d82663 authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Device component - GNMI OpenConfig: 

- advanced development of driver
- updated unitary tests and related scripts
parent a3df08b8
No related branches found
No related tags found
2 merge requests!294Release TeraFlowSDN 4.0,!172Resolve "(CTTC) Extend gNMI-OpenConfig SBI driver"
Hide whitespace changes
Inline Side-by-side
Showing
with 1051 additions and 81 deletions
scripts/run_tests_locally-device-gnmi-openconfig.sh
+ 1
− 1
View file @ 32d82663
...@@ -22,4 +22,4 @@ RCFILE=$PROJECTDIR/coverage/.coveragerc ...@@ -22,4 +22,4 @@ RCFILE=$PROJECTDIR/coverage/.coveragerc
# Run unitary tests and analyze coverage of code at same time # 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 # 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 \ 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
src/device/service/drivers/gnmi_openconfig/handlers/InterfaceCounter.py
+ 33
− 33
View file @ 32d82663
...@@ -12,11 +12,10 @@ ...@@ -12,11 +12,10 @@
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
import json, logging import json, libyang, logging
from typing import Any, Dict, List, Tuple from typing import Any, Dict, List, Tuple
import pyangbind.lib.pybindJSON as pybindJSON
from . import openconfig
from ._Handler import _Handler from ._Handler import _Handler
from .YangHandler import YangHandler
LOGGER = logging.getLogger(__name__) LOGGER = logging.getLogger(__name__)
...@@ -25,40 +24,41 @@ class InterfaceCounterHandler(_Handler): ...@@ -25,40 +24,41 @@ class InterfaceCounterHandler(_Handler):
def get_resource_key(self) -> str: return '/interface/counters' def get_resource_key(self) -> str: return '/interface/counters'
def get_path(self) -> str: return '/openconfig-interfaces:interfaces/interface/state/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))) 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 = [] yang_interfaces_path = self.get_path()
for interface_key, oc_interface in oc_interfaces.interface.items(): json_data_valid = yang_handler.parse_to_dict(yang_interfaces_path, json_data, fmt='json')
LOGGER.info('interface_key={:s} oc_interfaces={:s}'.format(
interface_key, pybindJSON.dumps(oc_interface, mode='ietf')
))
interface = {} entries = []
interface['name'] = oc_interface.name for interface in json_data_valid['interfaces']['interface']:
LOGGER.info('interface={:s}'.format(str(interface)))
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
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))) LOGGER.info('interface = {:s}'.format(str(interface)))
if len(interface) == 0: continue entry_interface_key = '/interface[{:s}]'.format(interface_name)
counters.append(('/interface[{:s}]'.format(interface['name']), interface)) entries.append((entry_interface_key, _interface))
return counters return entries
src/device/service/drivers/gnmi_openconfig/handlers/NetworkInstance.py
+ 22
− 12
View file @ 32d82663
...@@ -12,11 +12,10 @@ ...@@ -12,11 +12,10 @@
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
import json, libyang, logging import json, libyang, logging, operator
import operator
from typing import Any, Dict, List, Tuple from typing import Any, Dict, List, Tuple
from ._Handler import _Handler from ._Handler import _Handler
from .Tools import get_bool, get_int, get_str from .Tools import get_str
from .YangHandler import YangHandler from .YangHandler import YangHandler
LOGGER = logging.getLogger(__name__) LOGGER = logging.getLogger(__name__)
...@@ -45,7 +44,7 @@ class NetworkInstanceHandler(_Handler): ...@@ -45,7 +44,7 @@ class NetworkInstanceHandler(_Handler):
def compose( def compose(
self, resource_key : str, resource_value : Dict, yang_handler : YangHandler, delete : bool = False self, resource_key : str, resource_value : Dict, yang_handler : YangHandler, delete : bool = False
) -> Tuple[str, str]: ) -> Tuple[str, str]:
ni_name = get_str(resource_value, 'name') # test-svc ni_name = get_str(resource_value, 'name') # test-svc
if delete: if delete:
PATH_TMPL = '/network-instances/network-instance[name={:s}]' PATH_TMPL = '/network-instances/network-instance[name={:s}]'
...@@ -56,21 +55,32 @@ class NetworkInstanceHandler(_Handler): ...@@ -56,21 +55,32 @@ class NetworkInstanceHandler(_Handler):
ni_type = get_str(resource_value, 'type') # L3VRF / L2VSI / ... ni_type = get_str(resource_value, 'type') # L3VRF / L2VSI / ...
ni_type = MAP_NETWORK_INSTANCE_TYPE.get(ni_type, ni_type) 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) # 'DIRECTLY_CONNECTED' is implicitly added
str_data = json.dumps({ #'protocols': {'protocol': protocols},
'name': ni_name,
'config': {'name': ni_name, 'type': ni_type}, str_data = yang_ni.print_mem('json')
#'protocols': {'protocol': protocols}, 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 return str_path, str_data
def parse( def parse(
self, json_data : Dict, yang_handler : YangHandler self, json_data : Dict, yang_handler : YangHandler
) -> List[Tuple[str, Dict[str, Any]]]: ) -> 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: # Arista Parsing Fixes:
# - Default instance comes with mpls/signaling-protocols/rsvp-te/global/hellos/state/hello-interval set to 0 # - Default instance comes with mpls/signaling-protocols/rsvp-te/global/hellos/state/hello-interval set to 0
......
src/device/service/drivers/gnmi_openconfig/handlers/NetworkInstanceInterface.py
+ 40
− 12
View file @ 32d82663
...@@ -12,21 +12,31 @@ ...@@ -12,21 +12,31 @@
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
import json, logging import json, libyang, logging
from typing import Any, Dict, List, Tuple from typing import Any, Dict, List, Tuple
from ._Handler import _Handler from ._Handler import _Handler
from .Tools import get_int, get_str
from .YangHandler import YangHandler
LOGGER = logging.getLogger(__name__) LOGGER = logging.getLogger(__name__)
IS_CEOS = True
class NetworkInstanceInterfaceHandler(_Handler): class NetworkInstanceInterfaceHandler(_Handler):
def get_resource_key(self) -> str: return '/network_instance/interface' 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]: def compose(
ni_name = str(resource_value['name' ]) # test-svc self, resource_key : str, resource_value : Dict, yang_handler : YangHandler, delete : bool = False
if_name = str(resource_value['if_name' ]) # ethernet-1/1 ) -> Tuple[str, str]:
sif_index = int(resource_value['sif_index']) # 0 ni_name = get_str(resource_value, 'name' ) # test-svc
if_id = '{:s}.{:d}'.format(if_name, sif_index) 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: if delete:
PATH_TMPL = '/network-instances/network-instance[name={:s}]/interfaces/interface[id={:s}]' PATH_TMPL = '/network-instances/network-instance[name={:s}]/interfaces/interface[id={:s}]'
...@@ -35,12 +45,30 @@ class NetworkInstanceInterfaceHandler(_Handler): ...@@ -35,12 +45,30 @@ class NetworkInstanceInterfaceHandler(_Handler):
return str_path, str_data return str_path, str_data
str_path = '/network-instances/network-instance[name={:s}]/interfaces/interface[id={:s}]'.format(ni_name, if_id) str_path = '/network-instances/network-instance[name={:s}]/interfaces/interface[id={:s}]'.format(ni_name, if_id)
str_data = json.dumps({ #str_data = json.dumps({
'id': if_id, # 'id': if_id,
'config': {'id': if_id, 'interface': if_name, 'subinterface': sif_index}, # '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 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 = [] response = []
return response return response
src/device/service/drivers/gnmi_openconfig/handlers/NetworkInstanceStaticRoute.py
+ 56
− 23
View file @ 32d82663
...@@ -12,21 +12,25 @@ ...@@ -12,21 +12,25 @@
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
import json, logging import json, libyang, logging
from typing import Any, Dict, List, Tuple from typing import Any, Dict, List, Tuple
from ._Handler import _Handler from ._Handler import _Handler
from .Tools import get_int, get_str
from .YangHandler import YangHandler
LOGGER = logging.getLogger(__name__) LOGGER = logging.getLogger(__name__)
class NetworkInstanceStaticRouteHandler(_Handler): class NetworkInstanceStaticRouteHandler(_Handler):
def get_resource_key(self) -> str: return '/network_instance/static_route' 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]: def compose(
ni_name = str(resource_value['name' ]) # test-svc self, resource_key : str, resource_value : Dict, yang_handler : YangHandler, delete : bool = False
prefix = str(resource_value['prefix' ]) # '172.0.1.0/24' ) -> 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' name = 'static'
if delete: if delete:
PATH_TMPL = '/network-instances/network-instance[name={:s}]/protocols' PATH_TMPL = '/network-instances/network-instance[name={:s}]/protocols'
...@@ -35,27 +39,56 @@ class NetworkInstanceStaticRouteHandler(_Handler): ...@@ -35,27 +39,56 @@ class NetworkInstanceStaticRouteHandler(_Handler):
str_data = json.dumps({}) str_data = json.dumps({})
return str_path, str_data return str_path, str_data
next_hop = str(resource_value['next_hop' ]) # '172.0.0.1' next_hop = get_str(resource_value, 'next_hop' ) # '172.0.0.1'
next_hop_index = int(resource_value.get('next_hop_index', 0)) # 0 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}]' PATH_TMPL = '/network-instances/network-instance[name={:s}]/protocols/protocol[identifier={:s}][name={:s}]'
str_path = PATH_TMPL.format(ni_name, identifier, name) str_path = PATH_TMPL.format(ni_name, identifier, name)
str_data = json.dumps({ #str_data = json.dumps({
'identifier': identifier, 'name': name, # 'identifier': identifier, 'name': name,
'config': {'identifier': identifier, 'name': name, 'enabled': True}, # 'config': {'identifier': identifier, 'name': name, 'enabled': True},
'static_routes': {'static': [{ # 'static_routes': {'static': [{
'prefix': prefix, # 'prefix': prefix,
'config': {'prefix': prefix}, # 'config': {'prefix': prefix},
'next_hops': { # 'next_hops': {
'next-hop': [{ # 'next-hop': [{
'index': next_hop_index, # 'index': next_hop_index,
'config': {'index': next_hop_index, 'next_hop': next_hop} # '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 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 = [] response = []
return response return response
src/device/tests/gnmi_openconfig/__init__.py 0 → 100644
+ 14
− 0
View file @ 32d82663
# 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.
src/device/tests/gnmi_openconfig/request_composers.py 0 → 100644
+ 44
− 0
View file @ 32d82663
# 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
src/device/tests/test_unitary_gnmi_openconfig.py → src/device/tests/gnmi_openconfig/storage.py
+ 285
− 0
View file @ 32d82663
...@@ -12,45 +12,15 @@ ...@@ -12,45 +12,15 @@
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
import deepdiff, logging, os, pytest, re, time import pytest, re
from typing import Dict, List, Tuple 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') @pytest.fixture(scope='session')
def storage() -> Dict: def storage() -> Dict:
yield dict() yield dict()
##### STORAGE POPULATORS ############################################################################################### ##### POPULATE INTERFACE STORAGE #######################################################################################
def populate_interfaces_storage( def populate_interfaces_storage(
storage : Dict, # pylint: disable=redefined-outer-name storage : Dict, # pylint: disable=redefined-outer-name
...@@ -105,6 +75,9 @@ def populate_interfaces_storage( ...@@ -105,6 +75,9 @@ def populate_interfaces_storage(
ipv4_address_storage['prefix'] = resource_value.get('prefix') ipv4_address_storage['prefix'] = resource_value.get('prefix')
continue continue
##### POPULATE NETWORK INSTANCE STORAGE ################################################################################
def populate_network_instances_storage( def populate_network_instances_storage(
storage : Dict, # pylint: disable=redefined-outer-name storage : Dict, # pylint: disable=redefined-outer-name
resources : List[Tuple[str, Dict]], resources : List[Tuple[str, Dict]],
...@@ -165,7 +138,7 @@ def populate_network_instances_storage( ...@@ -165,7 +138,7 @@ def populate_network_instances_storage(
continue continue
##### EXPECTED CONFIG COMPOSERS ######################################################################################## ##### GET EXPECTED INTERFACE CONFIG ####################################################################################
INTERFACE_CONFIG_STRUCTURE : List[Tuple[str, List[str]]] = [ INTERFACE_CONFIG_STRUCTURE : List[Tuple[str, List[str]]] = [
('/interface[{if_name:s}]', [ ('/interface[{if_name:s}]', [
...@@ -224,6 +197,9 @@ def get_expected_interface_config( ...@@ -224,6 +197,9 @@ def get_expected_interface_config(
return expected_interface_config return expected_interface_config
##### GET EXPECTED NETWORK INSTANCE CONFIG #############################################################################
NETWORK_INSTANCE_CONFIG_STRUCTURE : List[Tuple[str, List[str]]] = [ NETWORK_INSTANCE_CONFIG_STRUCTURE : List[Tuple[str, List[str]]] = [
('/network_instance[{ni_name:s}]', ['name', 'type']), ('/network_instance[{ni_name:s}]', ['name', 'type']),
] ]
...@@ -307,331 +283,3 @@ def get_expected_network_instance_config( ...@@ -307,331 +283,3 @@ def get_expected_network_instance_config(
expected_network_instance_config.append((resource_key, resource_value)) expected_network_instance_config.append((resource_key, resource_value))
return expected_network_instance_config 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)))
src/device/tests/gnmi_openconfig/test_unitary_gnmi_openconfig.py 0 → 100644
+ 556
− 0
View file @ 32d82663
# 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
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment