Merge branch 'develop' of ssh://gifrerenom_labs.etsi.org/tfs/controller into...

#!/bin/bash

# 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.

PROJECTDIR=`pwd`

cd $PROJECTDIR/src

RCFILE=$PROJECTDIR/coverage/.coveragerc

python3 -m pytest --log-level=DEBUG --log-cli-level=DEBUG --verbose \

    analytics/backend/tests/test_backend.py

#!/bin/bash

# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)

#

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

# See the License for the specific language governing permissions and

# limitations under the License.

import logging

from flask import request

from flask.json import jsonify

from flask_restful import Resource

from common.proto.context_pb2 import Empty

from context.client.ContextClient import ContextClient

from ..tools.Authentication import HTTP_AUTH

from ..tools.HttpStatusCodes import HTTP_OK, HTTP_SERVERERROR

from .YangHandler import YangHandler

PROJECTDIR=`pwd`

LOGGER = logging.getLogger(__name__)

cd $PROJECTDIR/src

# RCFILE=$PROJECTDIR/coverage/.coveragerc

# coverage run --rcfile=$RCFILE --append -m pytest --log-level=INFO --verbose \

#     kpi_manager/tests/test_unitary.py

class HardwareMultipleDevices(Resource):

    @HTTP_AUTH.login_required

    def get(self):

RCFILE=$PROJECTDIR/coverage/.coveragerc

python3 -m pytest --log-level=DEBUG --log-cli-level=debug --verbose \

    telemetry/tests/test_telemetryDB.py

        LOGGER.debug('Request: {:s}'.format(str(request)))

        try:

            context_client = ContextClient()

            list_devices = context_client.ListDevices(Empty())

            LOGGER.info('Request: {:s}'.format(str(list_devices)))

            hardware_list_reply = []

            yang_handler = YangHandler()

            for device in list_devices.devices:

                hardware_reply = yang_handler.compose(device)

                hardware_list_reply.append(hardware_reply)

            yang_handler.destroy()

            response = jsonify(hardware_list_reply)

            response.status_code = HTTP_OK

        except Exception as e: # pylint: disable=broad-except

            MSG = 'Something went wrong Retrieving Hardware of Devices({:s})'

            response = jsonify({'error': str(e)})

            response.status_code = HTTP_SERVERERROR

        return response

# See the License for the specific language governing permissions and

# limitations under the License.

import libyang, os

from common.proto.context_pb2 import Device

from typing import Dict, Optional

import datetime

import json

import logging

import libyang

import os

import re

import datetime

LOGGER = logging.getLogger(__name__)

YANG_DIR = os.path.join(os.path.dirname(__file__), 'yang')

YANG_MODULES = [

    'iana-hardware',

    'ietf-hardware'

    'ietf-hardware',

    'ietf-network-hardware-inventory'

]

class YangHandler:

            self._yang_context.load_module(yang_module_name).feature_enable_all()

    def parse_to_dict(self, message : Dict) -> Dict:

        yang_module = self._yang_context.get_module('ietf-hardware')

        yang_module = self._yang_context.get_module('ietf-network-hardware-inventory')

        dnode : Optional[libyang.DNode] = yang_module.parse_data_dict(

            message, validate_present=True, validate=True, strict=True

        )

        dnode.free()

        return message

    @staticmethod

    def convert_to_iso_date(date_str: str) -> Optional[str]:

        date_str = date_str.strip('"')

        # Define the regex pattern for ISO 8601 date format

        pattern = r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d+)?(Z|[\+\-]\d{2}:\d{2})"

        # Check if the input date string matches the pattern

        if re.match(pattern, date_str):

            return date_str  # Already in ISO format

            return date_str 

        else:

            try:

                # Parse the input date string as a datetime object

                datetime_obj = datetime.datetime.strptime(date_str, "%Y-%m-%d")

                # Convert to ISO format

                iso_date = datetime_obj.isoformat() + "Z"

                return iso_date

            except ValueError:

                return None  # Invalid date format

                return None  

    def compose(self, device : Device) -> Dict:

        # compose device iterating through the components

        hardware = self._yang_context.create_data_path('/ietf-network-hardware-inventory:network-hardware-inventory')

        network_elements = hardware.create_path('network-elements')

        hardware = self._yang_context.create_data_path('/ietf-hardware:hardware')

        network_element = network_elements.create_path('network-element[uuid="{:s}"]'.format(device.device_id.device_uuid.uuid))

        network_element.create_path('uuid', device.device_id.device_uuid.uuid)

        network_element.create_path('name', device.name)

        components = network_element.create_path('components')

        physical_index = 1

        for component in device.components:

            attributes = component.attributes

            component_new = hardware.create_path('component[name="{:s}"]'.format(component.name))

            component_new = components.create_path('component[uuid="{:s}"]'.format(component.component_uuid.uuid))

            component_new.create_path('name', component.name)

            #Cambiar las clases especiales, su formato  y añadir isfru 

            component_type = component.type

            if component_type == "TRANSCEIVER" :

                component_type = "module"

            if component_type == "FRU" :

                component_type = "slack"

                component_new.create_path('is-fru', True)

            else :

                component_new.create_path('is-fru', False)

            component_type = component_type.replace("_", "-").lower()

            component_type = 'iana-hardware:' + component_type

            component_new.create_path('class', component_type)

            #Añadir resto de atributos en IETF

            physical_index += 1

            component_new.create_path('physical-index', physical_index)

            component_new.create_path('description', attributes["description"])

            component_new.create_path('parent', component.parent)

            component_new.create_path('description', attributes["description"].replace('/"',""))

            if "CHASSIS" not in component.type:

                parent_component_references = component_new.create_path('parent-component-references')

                parent = parent_component_references.create_path('component-reference[index="{:d}"]'.format(physical_index))

                for component_parent in device.components:

                    if component.parent == component_parent.name : 

                      parent.create_path('uuid', component_parent.component_uuid.uuid)

                      break

            if attributes["mfg-date"] != "":

                mfg_date = self.convert_to_iso_date(attributes["mfg-date"])

                LOGGER.info('component[name="{:s}"]'.format(attributes["mfg-date"]))

                component_new.create_path('mfg-date', mfg_date)

            component_new.create_path('hardware-rev', attributes["hardware-rev"])

            component_new.create_path('firmware-rev', attributes["firmware-version"])

            component_new.create_path('serial-num', attributes["serial-num"])

            component_new.create_path('mfg-name', attributes["mfg-name"])

            if attributes["removable"]:

                removable = attributes["removable"].lower()

                if 'true' in removable:

                    component_new.create_path('is-fru', True)

                elif 'false' in removable:

                    component_new.create_path('is-fru', False)

            if attributes["id"]:

                component_new.create_path('parent-rel-pos', attributes["id"])

                try:

                    if  "CHASSIS" in component.type :  

                        component_new.create_path('parent-rel-pos', 0)

                    else:                        

                        parent_rel_pos = int(attributes["id"].replace("\"", ""))

                        component_new.create_path('parent-rel-pos', parent_rel_pos)

                except ValueError:

                    LOGGER.info('ERROR:{:s} '.format(component.name ))

                    continue

            component_new.create_path('uri', component.name)

            component_new.create_path('uuid', component.component_uuid.uuid)

            contains_child = []

            for component2 in device.components:

                if component.name == component2.parent : 

                 contains_child.append(component2.name)

            component_new.create_path('contains-child', contains_child)

            for child in device.components:

                if component.name == child.parent : 

                    component_new.create_path('contained-child', child.component_uuid.uuid)

        return json.loads(hardware.print_mem('json'))

# limitations under the License.

from nbi.service.rest_server.nbi_plugins.ietf_hardware.Hardware import Hardware

from nbi.service.rest_server.nbi_plugins.ietf_hardware.HardwareMultipleDevices import HardwareMultipleDevices

from nbi.service.rest_server.RestServer import RestServer

URL_PREFIX = "/restconf/data/device=<path:device_uuid>/ietf-hardware:hardware"

URL_PREFIX_DEVICE   = "/restconf/data/device=<path:device_uuid>/ietf-network-hardware-inventory:network-hardware-inventory"

URL_PREFIX_HARDWARE = "/restconf/data/ietf-network-hardware-inventory:network-hardware-inventory"

def register_ietf_hardware(rest_server: RestServer):

    rest_server.add_resource(Hardware, URL_PREFIX)

 No newline at end of file

    rest_server.add_resource(Hardware, URL_PREFIX_DEVICE)

    rest_server.add_resource(HardwareMultipleDevices, URL_PREFIX_HARDWARE)