Commit 3a2f62eb authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Device component - GNMI OpenConfig: 

- Updated unitary tests
parent 03f748da

src/device/tests/gnmi_openconfig/storage/StorageNetworkInstance.py

+7 −4
Original line number Diff line number Diff line
@@ -45,16 +45,14 @@ class NetworkInstances: 


class Interfaces:

    STRUCT : List[Tuple[str, List[str]]] = [

        ('/network_instance[{:s}]/interface[{:s}]', ['ni_name', 'if_name']),

        ('/network_instance[{:s}]/interface[{:s}]', []),

    ]



    def __init__(self) -> None:

        self._items : Dict[Tuple[str, str], Dict] = dict()



    def add(self, ni_name : str, if_name : str) -> None:

        item = self._items.setdefault((ni_name, if_name), dict())

        item['ni_name'] = ni_name

        item['if_name'] = if_name

        self._items.setdefault((ni_name, if_name), dict())



    def remove(self, ni_name : str, if_name : str) -> None:

        self._items.pop((ni_name, if_name), None)
@@ -86,6 +84,11 @@ class StaticRoutes: 
        ('/network_instance[{:s}]/protocol[{:s}]/static_routes[{:s}]', ['prefix', 'next_hops']),

    ]



    #('/network_instance[{:s}]/static_route[{:s}]'.format(ni_name, static_route['prefix']), {

    #    'name': ni_name, 'prefix': static_route['prefix'], 'next_hop': static_route['gateway'],

    #    'next_hop_index': 0, 'metric': static_route['metric']

    #})



    def __init__(self) -> None:

        self._items : Dict[Tuple[str, str, str], Dict] = dict()

src/device/tests/gnmi_openconfig/test_unitary_gnmi_openconfig.py

+71 −68
Original line number Diff line number Diff line
@@ -16,21 +16,17 @@ import os 
os.environ['DEVICE_EMULATED_ONLY'] = 'YES'



# pylint: disable=wrong-import-position

import itertools, logging, pytest, time

import logging, pytest, time

from typing import Dict

from device.service.driver_api._Driver import (

    RESOURCE_ENDPOINTS, RESOURCE_INTERFACES, RESOURCE_NETWORK_INSTANCES, RESOURCE_ROUTING_POLICIES, RESOURCE_SERVICES

)

from device.service.drivers.gnmi_openconfig.GnmiOpenConfigDriver import GnmiOpenConfigDriver

from .tools.check_config import check_config_endpoints, check_config_interfaces, check_config_network_instances

from .tools.check_updates import check_updates

from .tools.expected_config_composers import (

    compose_expected_config__interface, compose_expected_config__network_instance

from .storage.Storage import Storage

from .tools.check_config import (

    check_config_endpoints, check_config_interfaces, check_config_network_instances

)

from .tools.check_updates import check_updates

from .tools.request_composers import (

    interface, network_instance, network_instance_interface, network_instance_static_route

)

from .storage.Storage import Storage



logging.basicConfig(level=logging.DEBUG)

LOGGER = logging.getLogger(__name__)
@@ -200,7 +196,7 @@ def test_add_interfaces_to_network_instance( 
            subif_index = ni_if['sif_index']

            resources_to_set.append(network_instance_interface(ni_name, if_name, subif_index))

            ni_if_names.append((ni_name, '{:s}.{:d}'.format(if_name, subif_index)))

            storage.network_instances.interfaces.add(ni_name, if_name, subif_index)

            storage.network_instances.interfaces.add(ni_name, if_name) # TODO: add subif_index



    LOGGER.info('resources_to_set = {:s}'.format(str(resources_to_set)))

    results_setconfig = driver.SetConfig(resources_to_set)
@@ -218,22 +214,26 @@ def test_set_network_instance_static_routes( 
    check_config_interfaces(driver, storage)

    check_config_network_instances(driver, storage)



    # TODO: update structure

    resources_to_set = list()

    ni_sr_prefixes = list()

    for ni in NETWORK_INSTANCES:

        ni_name = ni['name']

        for ni_sr in ni.get('static_routes', list()):

            ni_sr_prefix  = ni_sr['prefix' ]

            ni_sr_gateway = ni_sr['gateway']

            ni_sr_metric  = ni_sr['metric' ]

            resources_to_set.append(

                network_instance_static_route(ni_name, ni_sr_prefix, ni_sr_gateway, metric=ni_sr_metric)

            )

            ni_sr_prefixes.append((ni_name, ni_sr_prefix))

            storage.network_instances.protocol_static.add(ni_name, 'STATIC', ni_sr_prefix, {

                'prefix': ni_sr_prefix, 

            })



    resources_to_set = list(itertools.chain(*[

        [

            network_instance_static_route(ni['name'], ni_sr['prefix'], ni_sr['gateway'], metric=ni_sr['metric'])

            for ni_sr in ni.get('static_routes', list())

        ]

        for ni in NETWORK_INSTANCES

    ]))

    LOGGER.info('resources_to_set = {:s}'.format(str(resources_to_set)))

    results_setconfig = driver.SetConfig(resources_to_set)

    LOGGER.info('results_setconfig = {:s}'.format(str(results_setconfig)))

    check_updates(results_setconfig, '/network_instance[{:s}]/static_route[{:s}]', list(itertools.chain(*[

        [(ni['name'], ni_sr['prefix']) for ni_sr in ni.get('static_routes', list())]

        for ni in NETWORK_INSTANCES

    ])))

    check_updates(results_setconfig, '/network_instance[{:s}]/static_route[{:s}]', ni_sr_prefixes)



    check_config_interfaces(driver, storage, max_retries=5)

    check_config_network_instances(driver, storage, max_retries=5)
@@ -246,22 +246,24 @@ def test_del_network_instance_static_routes( 
    check_config_interfaces(driver, storage)

    check_config_network_instances(driver, storage)



    # TODO: update structure

    resources_to_delete = list()

    ni_sr_prefixes = list()

    for ni in NETWORK_INSTANCES:

        ni_name = ni['name']

        for ni_sr in ni.get('static_routes', list()):

            ni_sr_prefix  = ni_sr['prefix' ]

            ni_sr_gateway = ni_sr['gateway']

            ni_sr_metric  = ni_sr['metric' ]

            resources_to_delete.append(

                network_instance_static_route(ni_name, ni_sr_prefix, ni_sr_gateway, metric=ni_sr_metric)

            )

            ni_sr_prefixes.append((ni_name, ni_sr_prefix))

            storage.network_instances.protocol_static.remove(ni_name, 'STATIC', ni_sr_prefix)



    resources_to_delete = list(itertools.chain(*[

        [

            network_instance_static_route(ni['name'], ni_sr['prefix'], ni_sr['gateway'], metric=ni_sr['metric'])

            for ni_sr in ni.get('static_routes', list())

        ]

        for ni in NETWORK_INSTANCES

    ]))

    LOGGER.info('resources_to_delete = {:s}'.format(str(resources_to_delete)))

    results_deleteconfig = driver.DeleteConfig(resources_to_delete)

    LOGGER.info('results_deleteconfig = {:s}'.format(str(results_deleteconfig)))

    check_updates(results_deleteconfig, '/network_instance[{:s}]/static_route[{:s}]', list(itertools.chain(*[

        [(ni['name'], ni_sr['prefix']) for ni_sr in ni.get('static_routes', list())]

        for ni in NETWORK_INSTANCES

    ])))

    check_updates(results_deleteconfig, '/network_instance[{:s}]/static_route[{:s}]', ni_sr_prefixes)



    check_config_interfaces(driver, storage, max_retries=5)

    check_config_network_instances(driver, storage, max_retries=5)
@@ -274,25 +276,21 @@ def test_del_interfaces_from_network_instance( 
    check_config_interfaces(driver, storage)

    check_config_network_instances(driver, storage)



    # TODO: update structure

    resources_to_delete = list()

    ni_if_names = list()

    for ni in NETWORK_INSTANCES:

        ni_name = ni['name']

        for ni_if in ni.get('interfaces', list()):

            if_name     = ni_if['if_name']

            subif_index = ni_if['sif_index']

            resources_to_delete.append(network_instance_interface(ni_name, if_name, subif_index))

            ni_if_names.append((ni_name, '{:s}.{:d}'.format(if_name, subif_index)))

            storage.network_instances.interfaces.remove(ni_name, if_name) # TODO: add subif_index



    resources_to_delete = list(itertools.chain(*[

        [

            network_instance_interface(ni['name'], ni_if['if_name'], ni_if['subif_index'])

            for ni_if in ni.get('interfaces', list())

        ]

        for ni in NETWORK_INSTANCES

    ]))

    LOGGER.info('resources_to_delete = {:s}'.format(str(resources_to_delete)))

    results_deleteconfig = driver.DeleteConfig(resources_to_delete)

    LOGGER.info('results_deleteconfig = {:s}'.format(str(results_deleteconfig)))

    check_updates(results_deleteconfig, '/network_instance[{:s}]/interface[{:s}]', list(itertools.chain(*[

        [

            (ni['name'], '{:s}.{:d}'.format(ni_if['if_name'], ni_if['subif_index']))

            for ni_if in ni.get('interfaces', list())

        ]

        for ni in NETWORK_INSTANCES

    ])))

    check_updates(results_deleteconfig, '/network_instance[{:s}]/interface[{:s}]', ni_if_names)

    

    check_config_interfaces(driver, storage, max_retries=5)

    check_config_network_instances(driver, storage, max_retries=5)
@@ -305,22 +303,24 @@ def test_del_interfaces( 
    check_config_interfaces(driver, storage)

    check_config_network_instances(driver, storage)



    # TODO: update structure

    resources_to_delete = list()

    if_names = list()

    for ni in NETWORK_INSTANCES:

        ni_name = ni['name']

        for ni_if in ni.get('interfaces', list()):

            if_name      = ni_if['if_name']

            subif_index  = ni_if['sif_index']

            ipv4_address = ni_if['ipv4_addr']

            ipv4_prefix  = ni_if['ipv4_prefix']

            enabled      = ni_if['enabled']

            resources_to_delete.append(interface(if_name, subif_index, ipv4_address, ipv4_prefix, enabled))

            if_names.append(ni_name)

            storage.interfaces.ipv4_addresses.remove(if_name, subif_index, ipv4_address)



    resources_to_delete = list(itertools.chain(*[

        [

            interface(ni_if['if_name'], ni_if['sif_index'], ni_if['ipv4_addr'], ni_if['ipv4_prefix'], ni_if['enabled'])

            for ni_if in ni.get('interfaces', list())

        ]

        for ni in NETWORK_INSTANCES

    ]))

    LOGGER.info('resources_to_delete = {:s}'.format(str(resources_to_delete)))

    results_deleteconfig = driver.DeleteConfig(resources_to_delete)

    LOGGER.info('results_deleteconfig = {:s}'.format(str(results_deleteconfig)))

    check_updates(results_deleteconfig, '/interface[{:s}]', list(itertools.chain(*[

        [ni_if['name'] for ni_if in ni.get('interfaces', list())]

        for ni in NETWORK_INSTANCES

    ])))

    check_updates(results_deleteconfig, '/interface[{:s}]', if_names)



    check_config_interfaces(driver, storage, max_retries=5)

    check_config_network_instances(driver, storage, max_retries=5)
@@ -333,16 +333,19 @@ def test_del_network_instances( 
    check_config_interfaces(driver, storage)

    check_config_network_instances(driver, storage)



    # TODO: update structure

    resources_to_delete = list()

    ni_names = list()

    for ni in NETWORK_INSTANCES:

        ni_name = ni['name']

        ni_type = ni['type']

        resources_to_delete.append(network_instance(ni_name, ni_type))

        ni_names.append(ni_name)

        storage.network_instances.network_instances.remove(ni_name)



    resources_to_delete = [

        network_instance(ni['name'], ni['type'])

        for ni in NETWORK_INSTANCES

    ]

    LOGGER.info('resources_to_delete = {:s}'.format(str(resources_to_delete)))

    results_deleteconfig = driver.DeleteConfig(resources_to_delete)

    LOGGER.info('results_deleteconfig = {:s}'.format(str(results_deleteconfig)))

    check_updates(results_deleteconfig, '/network_instance[{:s}]', [ni['name'] for ni in NETWORK_INSTANCES])

    check_updates(results_deleteconfig, '/network_instance[{:s}]', ni_names)



    check_config_interfaces(driver, storage, max_retries=5)

    check_config_network_instances(driver, storage, max_retries=5)

src/device/tests/gnmi_openconfig/tools/check_config.py

+0 −1
Original line number Diff line number Diff line
@@ -75,7 +75,6 @@ def check_config_network_instances( 
    driver : GnmiOpenConfigDriver, storage : Storage,

    max_retries : int = 1, retry_delay : float = 0.5

) -> List[Dict]:

    expected_config = 

    return check_expected_config(

        driver, [RESOURCE_NETWORK_INSTANCES], storage.network_instances.get_expected_config(),

        adapt_network_instance, max_retries=max_retries, retry_delay=retry_delay

src/device/tests/gnmi_openconfig/tools/expected_config_composers.py

deleted100644 → 0
+0 −58
Original line number Diff line number Diff line 
# Copyright 2022-2023 ETSI TeraFlowSDN - TFS OSG (https://tfs.etsi.org/)

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

#      http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.



from typing import Dict, List





def compose_expected_config__network_instance(

    network_instances : List[Dict], include_interfaces : bool = False, include_static_routes : bool = False

) -> List[Dict]:

    expected_config = list()

    for network_instance in network_instances:

        ni_name = network_instance['name']

        ni_type = network_instance['type']



        expected_config.extend([

            ('/network_instance[{:s}]'.format(ni_name), {

                'name': ni_name, 'type': ni_type

            }),

            ('/network_instance[{:s}]/protocol[DIRECTLY_CONNECTED]'.format(ni_name), {

                'id': 'DIRECTLY_CONNECTED', 'name': 'DIRECTLY_CONNECTED'

            }),

            ('/network_instance[{:s}]/table[DIRECTLY_CONNECTED,IPV4]'.format(ni_name), {

                'protocol': 'DIRECTLY_CONNECTED', 'address_family': 'IPV4'

            }),

            ('/network_instance[{:s}]/table[DIRECTLY_CONNECTED,IPV6]'.format(ni_name), {

                'protocol': 'DIRECTLY_CONNECTED', 'address_family': 'IPV6'

            })

        ])



        if include_interfaces:

            expected_config.extend([

                ('/network_instance[{:s}]/interface[{:s}]'.format(ni_name, interface['name']), {



                })

                for interface in network_instance.get('interfaces', list())

            ])



        if include_static_routes:

            expected_config.extend([

                ('/network_instance[{:s}]/static_route[{:s}]'.format(ni_name, static_route['prefix']), {

                    'name': ni_name, 'prefix': static_route['prefix'], 'next_hop': static_route['gateway'],

                    'next_hop_index': 0, 'metric': static_route['metric']

                })

                for static_route in network_instance.get('static_routes', list())

            ])



    return expected_config