Commit bbd72aac authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Minor code corrections

parent 20095d7f

src/service/service/task_scheduler/TaskExecutor.py

+1 −2
Original line number Diff line number Diff line
@@ -14,10 +14,9 @@ 


import json, logging

from enum import Enum

from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, Union

from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union

from common.DeviceTypes import DeviceTypeEnum

from common.method_wrappers.ServiceExceptions import NotFoundException

from typing import List

from common.proto.qkd_app_pb2 import QKDAppStatusEnum

from common.proto.context_pb2 import (

    Connection, ConnectionId, Device, DeviceDriverEnum, DeviceId, Service, ServiceId,

src/service/tests/test_l3nm_gnmi_static_rule_gen/MockServiceHandler.py

+18 −3
Original line number Diff line number Diff line
@@ -14,13 +14,15 @@ 


import json, logging

from typing import Any, Dict, List, Optional, Tuple, Union

from common.proto.context_pb2 import ConfigRule, DeviceId, Service

from common.proto.context_pb2 import ConfigRule, ConnectionId, DeviceId, Service

from common.tools.object_factory.Connection import json_connection_id

from common.tools.object_factory.Device import json_device_id

from common.type_checkers.Checkers import chk_type

from service.service.service_handler_api._ServiceHandler import _ServiceHandler

from service.service.service_handler_api.SettingsHandler import SettingsHandler

from service.service.service_handler_api.Tools import get_device_endpoint_uuids, get_endpoint_matching

from .MockTaskExecutor import MockTaskExecutor

from service.service.tools.EndpointIdFormatters import endpointids_to_raw

from service.service.service_handlers.l3nm_gnmi_openconfig.ConfigRuleComposer import ConfigRuleComposer

from service.service.service_handlers.l3nm_gnmi_openconfig.StaticRouteGenerator import StaticRouteGenerator
@@ -48,17 +50,20 @@ class MockServiceHandler(_ServiceHandler): 


            device_obj = self.__task_executor.get_device(DeviceId(**json_device_id(device_uuid)))

            device_settings = self.__settings_handler.get_device_settings(device_obj)

            self.__config_rule_composer.set_device_alias(device_obj.name, device_uuid)

            _device = self.__config_rule_composer.get_device(device_obj.name)

            _device.configure(device_obj, device_settings)



            endpoint_obj = get_endpoint_matching(device_obj, endpoint_uuid)

            endpoint_settings = self.__settings_handler.get_endpoint_settings(device_obj, endpoint_obj)

            _device.set_endpoint_alias(endpoint_obj.name, endpoint_uuid)

            _endpoint = _device.get_endpoint(endpoint_obj.name)

            _endpoint.configure(endpoint_obj, endpoint_settings)



            self.__endpoint_map[(device_uuid, endpoint_uuid)] = (device_obj.name, endpoint_obj.name)



        self.__static_route_generator.compose(endpoints)

        LOGGER.debug('config_rule_composer = {:s}'.format(json.dumps(self.__config_rule_composer.dump())))



    def _do_configurations(

        self, config_rules_per_device : Dict[str, List[Dict]], endpoints : List[Tuple[str, str, Optional[str]]],
@@ -94,7 +99,12 @@ class MockServiceHandler(_ServiceHandler): 
    ) -> List[Union[bool, Exception]]:

        chk_type('endpoints', endpoints, list)

        if len(endpoints) == 0: return []

        self._compose_config_rules(endpoints)

        #service_uuid = self.__service.service_id.service_uuid.uuid

        connection = self.__task_executor.get_connection(ConnectionId(**json_connection_id(connection_uuid)))

        connection_endpoint_ids = endpointids_to_raw(connection.path_hops_endpoint_ids)

        self._compose_config_rules(connection_endpoint_ids)

        #network_instance_name = service_uuid.split('-')[0]

        #config_rules_per_device = self.__config_rule_composer.get_config_rules(network_instance_name, delete=False)

        config_rules_per_device = self.__config_rule_composer.get_config_rules(delete=False)

        LOGGER.debug('config_rules_per_device={:s}'.format(str(config_rules_per_device)))

        results = self._do_configurations(config_rules_per_device, endpoints)
@@ -106,7 +116,12 @@ class MockServiceHandler(_ServiceHandler): 
    ) -> List[Union[bool, Exception]]:

        chk_type('endpoints', endpoints, list)

        if len(endpoints) == 0: return []

        self._compose_config_rules(endpoints)

        #service_uuid = self.__service.service_id.service_uuid.uuid

        connection = self.__task_executor.get_connection(ConnectionId(**json_connection_id(connection_uuid)))

        connection_endpoint_ids = endpointids_to_raw(connection.path_hops_endpoint_ids)

        self._compose_config_rules(connection_endpoint_ids)

        #network_instance_name = service_uuid.split('-')[0]

        #config_rules_per_device = self.__config_rule_composer.get_config_rules(network_instance_name, delete=True)

        config_rules_per_device = self.__config_rule_composer.get_config_rules(delete=True)

        LOGGER.debug('config_rules_per_device={:s}'.format(str(config_rules_per_device)))

        results = self._do_configurations(config_rules_per_device, endpoints, delete=True)

src/service/tests/test_l3nm_gnmi_static_rule_gen/MockTaskExecutor.py

+31 −3
Original line number Diff line number Diff line
@@ -14,10 +14,12 @@ 


import logging

from enum import Enum

from typing import Dict, Optional, Union

from typing import Any, Dict, Optional, Union

from common.method_wrappers.ServiceExceptions import NotFoundException

from common.proto.context_pb2 import Connection, Device, DeviceId, Service

from service.service.tools.ObjectKeys import get_device_key

from common.proto.context_pb2 import (

    Connection, ConnectionId, Device, DeviceId, Service

)

from service.service.tools.ObjectKeys import get_connection_key, get_device_key



LOGGER = logging.getLogger(__name__)
@@ -46,6 +48,32 @@ class MockTaskExecutor: 
        object_key = '{:s}:{:s}'.format(object_type.value, object_key)

        self._grpc_objects_cache.pop(object_key, None)



    def _store_editable_grpc_object(

        self, object_type : CacheableObjectType, object_key : str, grpc_class, grpc_ro_object

    ) -> Any:

        grpc_rw_object = grpc_class()

        grpc_rw_object.CopyFrom(grpc_ro_object)

        self._store_grpc_object(object_type, object_key, grpc_rw_object)

        return grpc_rw_object



    # ----- Connection-related methods ---------------------------------------------------------------------------------



    def get_connection(self, connection_id : ConnectionId) -> Connection:

        connection_key = get_connection_key(connection_id)

        connection = self._load_grpc_object(CacheableObjectType.CONNECTION, connection_key)

        if connection is None: raise NotFoundException('Connection', connection_key)

        return connection



    def set_connection(self, connection : Connection) -> None:

        connection_key = get_connection_key(connection.connection_id)

        self._store_grpc_object(CacheableObjectType.CONNECTION, connection_key, connection)



    def delete_connection(self, connection_id : ConnectionId) -> None:

        connection_key = get_connection_key(connection_id)

        self._delete_grpc_object(CacheableObjectType.CONNECTION, connection_key)



    # ----- Device-related methods -------------------------------------------------------------------------------------



    def get_device(self, device_id : DeviceId) -> Device:

        device_key = get_device_key(device_id)

        device = self._load_grpc_object(CacheableObjectType.DEVICE, device_key)

src/service/tests/test_l3nm_gnmi_static_rule_gen/test_unitary_sns4sns.py

0 → 100644
+108 −0
Original line number Diff line number Diff line 
# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)

#

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

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

# You may obtain a copy of the License at

#

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

#

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

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

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

# See the License for the specific language governing permissions and

# limitations under the License.



# Run with:

# $ PYTHONPATH=./src python -m service.tests.test_l3nm_gnmi_static_rule_gen.test_unitary_sns4sns



import logging

from typing import List, Optional, Tuple

from common.DeviceTypes import DeviceTypeEnum

from common.proto.context_pb2 import Connection, Device, DeviceOperationalStatusEnum, Service

from common.tools.object_factory.ConfigRule import json_config_rule_set

from common.tools.object_factory.Connection import json_connection

from common.tools.object_factory.Device import json_device, json_device_id

from common.tools.object_factory.EndPoint import json_endpoint, json_endpoint_id

from common.tools.object_factory.Service import json_service_l3nm_planned

from .MockServiceHandler import MockServiceHandler

from .MockTaskExecutor import CacheableObjectType, MockTaskExecutor



logging.basicConfig(level=logging.DEBUG)

LOGGER = logging.getLogger(__name__)



SERVICE = Service(**json_service_l3nm_planned(

    'svc-core-edge-uuid',

    endpoint_ids=[

        json_endpoint_id(json_device_id('core-net'), 'eth1'),

        json_endpoint_id(json_device_id('edge-net'), 'eth1'),

    ],

    config_rules=[

        json_config_rule_set('/device[core-net]/endpoint[eth1]/settings', {

            'address_ip': '10.10.10.0', 'address_prefix': 24, 'index': 0

        }),

        json_config_rule_set('/device[r1]/endpoint[eth10]/settings', {

            'address_ip': '10.10.10.229', 'address_prefix': 24, 'index': 0

        }),

        json_config_rule_set('/device[r2]/endpoint[eth10]/settings', {

            'address_ip': '10.158.72.229', 'address_prefix': 24, 'index': 0

        }),

        json_config_rule_set('/device[edge-net]/endpoint[eth1]/settings', {

            'address_ip': '10.158.72.0', 'address_prefix': 24, 'index': 0

        }),

    ]

))



CONNECTION_ENDPOINTS : List[Tuple[str, str, Optional[str]]] = [

    #('core-net', 'int',   None),

    ('core-net', 'eth1',  None),

    ('r1',       'eth10', None), ('r1',       'eth2',  None),

    ('r2',       'eth1',  None), ('r2',       'eth10', None),

    ('edge-net', 'eth1',  None),

    #('edge-net', 'int',   None),

]



def test_l3nm_gnmi_static_rule_gen() -> None:

    dev_op_st_enabled = DeviceOperationalStatusEnum.DEVICEOPERATIONALSTATUS_ENABLED



    mock_task_executor = MockTaskExecutor()

    mock_task_executor._store_grpc_object(CacheableObjectType.DEVICE, 'core-net', Device(**json_device(

        'core-net', DeviceTypeEnum.EMULATED_DATACENTER.value, dev_op_st_enabled, name='core-net',

        endpoints=[

            json_endpoint(json_device_id('core-net'), 'int',  'packet', name='int' ),

            json_endpoint(json_device_id('core-net'), 'eth1', 'packet', name='eth1'),

        ]

    )))

    mock_task_executor._store_grpc_object(CacheableObjectType.DEVICE, 'edge-net', Device(**json_device(

        'edge-net', DeviceTypeEnum.EMULATED_DATACENTER.value, dev_op_st_enabled, name='edge-net',

        endpoints=[

            json_endpoint(json_device_id('edge-net'), 'int',  'packet', name='int' ),

            json_endpoint(json_device_id('edge-net'), 'eth1', 'packet', name='eth1'),

        ]

    )))

    mock_task_executor._store_grpc_object(CacheableObjectType.DEVICE, 'r1', Device(**json_device(

        'r1', DeviceTypeEnum.EMULATED_PACKET_ROUTER.value, dev_op_st_enabled, name='r1',

        endpoints=[

            json_endpoint(json_device_id('r1'), 'eth2',  'packet', name='eth2' ),

            json_endpoint(json_device_id('r1'), 'eth10', 'packet', name='eth10'),

        ]

    )))

    mock_task_executor._store_grpc_object(CacheableObjectType.DEVICE, 'r2', Device(**json_device(

        'r2', DeviceTypeEnum.EMULATED_PACKET_ROUTER.value, dev_op_st_enabled, name='r2',

        endpoints=[

            json_endpoint(json_device_id('r1'), 'eth1',  'packet', name='eth1' ),

            json_endpoint(json_device_id('r1'), 'eth10', 'packet', name='eth10'),

        ]

    )))

    mock_task_executor._store_grpc_object(CacheableObjectType.CONNECTION, 'conn', Connection(**json_connection(

        'conn', path_hops_endpoint_ids=[

            json_endpoint_id(json_device_id(device_uuid), endpoint_uuid=endpoint_uuid)

            for device_uuid, endpoint_uuid, _ in CONNECTION_ENDPOINTS

        ]

    )))



    mock_service_handler = MockServiceHandler(SERVICE, mock_task_executor)

    mock_service_handler.SetEndpoint(CONNECTION_ENDPOINTS, connection_uuid='conn')

    mock_service_handler.DeleteEndpoint(CONNECTION_ENDPOINTS, connection_uuid='conn')



if __name__ == '__main__':

    test_l3nm_gnmi_static_rule_gen()