feat: IETF Slices retrieval added

from context.client.ContextClient import ContextClient

from slice.client.SliceClient import SliceClient

from ..tools.HttpStatusCodes import HTTP_CREATED

from ..tools.HttpStatusCodes import HTTP_CREATED, HTTP_OK

from .ietf_slice_handler import IETFSliceHandler

LOGGER = logging.getLogger(__name__)

class NSS_Services(Resource):

    # @HTTP_AUTH.login_required

    def get(self):

        response = jsonify({"message": "All went well!"})

        # TODO Return list of current network-slice-services

        context_client = ContextClient()

        ietf_slices = IETFSliceHandler.get_all_ietf_slices(context_client)

        response = jsonify(ietf_slices)

        response.status_code = HTTP_OK

        return response

    # @HTTP_AUTH.login_required

import json

import logging

import uuid

from typing import Optional

from typing import Dict, List, Optional

from common.Constants import DEFAULT_CONTEXT_NAME

from common.proto.context_pb2 import (

)

from common.tools.context_queries.Slice import get_slice_by_defualt_name

from common.tools.grpc.ConfigRules import update_config_rule_custom

from common.tools.grpc.Tools import grpc_message_to_json

from context.client import ContextClient

from .YangValidator import YangValidator

            return cr

def sort_endpoints(endpoinst_list: list, sdps: list, connection_group: dict) -> list:

def sort_endpoints(endpoinst_list: List, sdps: List, connection_group: Dict) -> List:

    src_sdp_id = connection_group["connectivity-construct"][0]["p2p-sender-sdp"]

    sdp_id_name_mapping = {sdp["id"]: sdp["node-id"] for sdp in sdps}

    if endpoinst_list[0].device_id.device_uuid.uuid == sdp_id_name_mapping[src_sdp_id]:

def replace_ont_endpoint_with_emu_dc(

    endpoint_list: list, context_client: ContextClient

) -> list:

    endpoint_list: List, context_client: ContextClient

) -> List:

    link_list = context_client.ListLinks(Empty())

    links = list(link_list.links)

    devices_list = context_client.ListDevices(Empty())

    return endpoint_list

def validate_ietf_slice_data(request_data: dict) -> None:

def validate_ietf_slice_data(request_data: Dict) -> None:

    yang_validator = YangValidator("ietf-network-slice-service")

    _ = yang_validator.parse_to_dict(request_data)

    yang_validator.destroy()

class IETFSliceHandler:

    @staticmethod

    def get_all_ietf_slices(context_client: ContextClient) -> Dict:

        slices_list = context_client.ListSlices(Empty())

        slices = slices_list.slices

        ietf_slices = {"network-slice-services": {"slice-service": []}}

        for slice in slices:

            candidate_cr = get_custom_config_rule(

                slice.slice_config, CANDIDATE_RESOURCE_KEY

            )

            candidate_ietf_data = json.loads(candidate_cr.custom.resource_value)

            ietf_slices["network-slice-services"]["slice-service"].append(

                candidate_ietf_data["network-slice-services"]["slice-service"][0]

            )

        return ietf_slices

    @staticmethod

    def create_slice_service(

        request_data: dict, context_client: ContextClient