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

Service component: 

- Fixed broken imports
- Minor code polishing
parent c83cc810

src/service/service/service_handlers/oc/OCServiceHandler.py

+1 −2
Original line number Diff line number Diff line
@@ -25,8 +25,7 @@ from service.service.service_handler_api.SettingsHandler import SettingsHandler 
from service.service.task_scheduler.TaskExecutor import TaskExecutor

from .ConfigRules import setup_config_rules, teardown_config_rules

from .OCTools import (

    endpoints_to_flows,convert_endpoints_to_flows,

    convert_or_endpoints_to_flows

    endpoints_to_flows, convert_endpoints_to_flows, convert_or_endpoints_to_flows

)



LOGGER = logging.getLogger(__name__)

src/service/service/task_scheduler/tasks/Task_OpticalConnectionConfigure.py

+8 −13
Original line number Diff line number Diff line 
# Copyright 2022-2024 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)

# Copyright 2022-2025 ETSI 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.
@@ -12,11 +12,9 @@ 
# See the License for the specific language governing permissions and

# limitations under the License.



from typing import TYPE_CHECKING, Dict, Tuple

from common.DeviceTypes import DeviceTypeEnum

import logging

from common.method_wrappers.ServiceExceptions import OperationFailedException

from common.proto.context_pb2 import ConnectionId, Device , ServiceTypeEnum



from common.proto.context_pb2 import ConnectionId, ServiceTypeEnum

from common.tools.grpc.Tools import grpc_message_to_json_string

from service.service.service_handler_api.Tools import check_errors_setendpoint

from service.service.task_scheduler.TaskExecutor import TaskExecutor
@@ -24,15 +22,12 @@ from service.service.tools.EndpointIdFormatters import endpointids_to_raw 
from service.service.tools.ObjectKeys import get_connection_key

from service.service.service_handlers.oc.OCServiceHandler import OCServiceHandler





from ._Task import _Task

import logging



if TYPE_CHECKING:

    from service.service.service_handler_api._ServiceHandler import _ServiceHandler



KEY_TEMPLATE       = 'optical_Connection ({connection_id:s}):configure'

supported_handlers = (OCServiceHandler)

KEY_TEMPLATE       = 'optical_connection({connection_id:s}):configure'

SUPPORTED_HANDLERS = (OCServiceHandler,)





class Task_OpticalConnectionConfigure(_Task):

    def __init__(self, task_executor : TaskExecutor, connection_id : ConnectionId) -> None:
@@ -63,7 +58,7 @@ class Task_OpticalConnectionConfigure(_Task): 
                logging.info(f"type_servicehandler {handler} and {service_handler}")

                if type(handler) != type(service_handler) : 

                    if  service.service_type == ServiceTypeEnum.SERVICETYPE_OPTICAL_CONNECTIVITY:

                       if isinstance(handler, supported_handlers) and isinstance(service_handler, supported_handlers):

                       if isinstance(handler, SUPPORTED_HANDLERS) and isinstance(service_handler, SUPPORTED_HANDLERS):

                          break



                    raise Exception("Devices are not compatible ")

src/service/service/task_scheduler/tasks/Task_OpticalConnectionDeconfigure.py

+10 −6
Original line number Diff line number Diff line
@@ -19,10 +19,13 @@ from service.service.service_handler_api.Tools import check_errors_deleteendpoin 
from service.service.task_scheduler.TaskExecutor import TaskExecutor

from service.service.tools.EndpointIdFormatters import endpointids_to_raw

from service.service.tools.ObjectKeys import get_connection_key

from service.service.service_handlers.oc.OCServiceHandler import OCServiceHandler



from ._Task import _Task

from service.service.service_handlers.openroadm.OpenROADMServiceHandler import OpenROADMServiceHandler





KEY_TEMPLATE       = 'optical_connection({connection_id:s}):deconfigure'

SUPPORTED_HANDLERS = (OCServiceHandler,)



class Task_OpticalConnectionDeconfigure(_Task):

    def __init__(
@@ -52,12 +55,13 @@ class Task_OpticalConnectionDeconfigure(_Task): 


        service_handlers = self._task_executor.get_service_handlers(connection, service, **service_handler_settings)

        for _, (service_handler, connection_devices) in service_handlers.items():

            

            endpointids_to_delete = endpointids_to_raw(connection.path_hops_endpoint_ids)

            connection_uuid = connection.connection_id.connection_uuid.uuid

            if self._has_media_channel:

                if isinstance(service_handler,OpenROADMServiceHandler):is_media_channel=True

                else : is_media_channel = service_handler.check_media_channel(connection_uuid)

                if isinstance(service_handler, SUPPORTED_HANDLERS):

                    is_media_channel = True

                else:

                    is_media_channel = service_handler.check_media_channel(connection_uuid)

                if is_media_channel:

                    results_deleteendpoint = service_handler.DeleteEndpoint(endpointids_to_delete, connection_uuid=connection_uuid)

                    errors = check_errors_deleteendpoint(endpointids_to_delete, results_deleteendpoint)