rwlaunchpad/plugins/rwlaunchpadtasklet/rift/package/convert.py

# 
#   Copyright 2016 RIFT.IO Inc
#
#   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.
#

import json
import os
import tempfile

import gi
gi.require_version('RwNsdYang', '1.0')
gi.require_version('RwVnfdYang', '1.0')
gi.require_version('RwYang', '1.0')
from gi.repository import (
        RwNsdYang,
        RwVnfdYang,
        NsdYang,
        VnfdYang,
        RwYang,
        )


class UnknownExtensionError(Exception):
    pass


class SerializationError(Exception):
    pass


def decode(desc_data):
    if isinstance(desc_data, bytes):
        desc_data = desc_data.decode()

    return desc_data


class ProtoMessageSerializer(object):
    """(De)Serializer/deserializer fo a specific protobuf message into various formats"""
    libncx_model = None

    def __init__(self, yang_ns, yang_pb_cls):
        """ Create a serializer for a specific protobuf message """
        self._yang_ns = yang_ns
        self._yang_pb_cls = yang_pb_cls

    @classmethod
    def _deserialize_extension_method_map(cls):
        return {
                ".xml": cls._from_xml_file_hdl,
                ".yml": cls._from_yaml_file_hdl,
                ".yaml": cls._from_yaml_file_hdl,
                ".json": cls._from_json_file_hdl,
                }

    @classmethod
    def _serialize_extension_method_map(cls):
        return {
                ".xml": cls.to_xml_string,
                ".yml": cls.to_yaml_string,
                ".yaml": cls.to_yaml_string,
                ".json": cls.to_json_string,
                }

    @classmethod
    def is_supported_file(cls, filename):
        """Returns whether a file has a supported file extension

        Arguments:
            filename - A descriptor file

        Returns:
            True if file extension is supported, False otherwise

        """
        _, extension = os.path.splitext(filename)
        extension_lc = extension.lower()

        return extension_lc in cls._deserialize_extension_method_map()

    @property
    def yang_namespace(self):
        """ The Protobuf's GI namespace class (e.g. RwVnfdYang) """
        return self._yang_ns

    @property
    def yang_class(self):
        """ The Protobuf's GI class (e.g. RwVnfdYang.YangData_Vnfd_VnfdCatalog_Vnfd) """
        return self._yang_pb_cls

    @property
    def model(self):
        cls = self.__class__

        # Cache the libncx model for the serializer class
        if cls.libncx_model is None:
            cls.libncx_model = RwYang.model_create_libncx()
            cls.libncx_model.load_schema_ypbc(self.yang_namespace.get_schema())

        return cls.libncx_model

    def _from_xml_file_hdl(self, file_hdl):
        xml = file_hdl.read()

        return self.yang_class.from_xml_v2(self.model, decode(xml), strict=False)

    def _from_json_file_hdl(self, file_hdl):
        json = file_hdl.read()

        return self.yang_class.from_json(self.model, decode(json), strict=False)

    def _from_yaml_file_hdl(self, file_hdl):
        yaml = file_hdl.read()

        return self.yang_class.from_yaml(self.model, decode(yaml), strict=False)

    def to_json_string(self, pb_msg):
        """ Serialize a protobuf message into JSON

        Arguments:
            pb_msg - A GI-protobuf object of type provided into constructor

        Returns:
            A JSON string representing the protobuf message

        Raises:
            SerializationError - Message could not be serialized
            TypeError - Incorrect protobuf type provided
        """
        if not isinstance(pb_msg, self._yang_pb_cls):
            raise TypeError("Invalid protobuf message type provided")

        try:
            json_str = pb_msg.to_json(self.model)

        except Exception as e:
            raise SerializationError(e)

        return json_str

    def to_yaml_string(self, pb_msg):
        """ Serialize a protobuf message into YAML

        Arguments:
            pb_msg - A GI-protobuf object of type provided into constructor

        Returns:
            A YAML string representing the protobuf message

        Raises:
            SerializationError - Message could not be serialized
            TypeError - Incorrect protobuf type provided
        """
        if not isinstance(pb_msg, self._yang_pb_cls):
            raise TypeError("Invalid protobuf message type provided")

        try:
            yaml_str = pb_msg.to_yaml(self.model)

        except Exception as e:
            raise SerializationError(e)

        return yaml_str

    def to_xml_string(self, pb_msg):
        """ Serialize a protobuf message into XML

        Arguments:
            pb_msg - A GI-protobuf object of type provided into constructor

        Returns:
            A XML string representing the protobuf message

        Raises:
            SerializationError - Message could not be serialized
            TypeError - Incorrect protobuf type provided
        """
        if not isinstance(pb_msg, self._yang_pb_cls):
            raise TypeError("Invalid protobuf message type provided")

        try:
            xml_str = pb_msg.to_xml_v2(self.model)

        except Exception as e:
            raise SerializationError(e)

        return xml_str

    def from_file_hdl(self, file_hdl, extension):
        """ Returns the deserialized protobuf message from file contents

        This function determines the serialization format based on file extension

        Arguments:
            file_hdl - The file hdl to deserialize (set at pos 0)
            extension - Extension of the file format (second item of os.path.splitext())

        Returns:
            A GI-Proto message of type that was provided into the constructor

        Raises:
            UnknownExtensionError - File extension is not of a known serialization format
            SerializationError - File failed to be deserialized into the protobuf message
        """

        extension_lc = extension.lower()
        extension_map = self._deserialize_extension_method_map()

        if extension_lc not in extension_map:
            raise UnknownExtensionError("Cannot detect message format for %s extension" % extension_lc)

        try:
            msg = extension_map[extension_lc](self, file_hdl)
        except Exception as e:
            raise SerializationError(e)

        return msg

    def to_string(self, pb_msg, extension):
        """ Returns the serialized protobuf message for a particular file extension

        This function determines the serialization format based on file extension

        Arguments:
            pb_msg - A GI-protobuf object of type provided into constructor
            extension - Extension of the file format (second item of os.path.splitext())

        Returns:
            A GI-Proto message of type that was provided into the constructor

        Raises:
            UnknownExtensionError - File extension is not of a known serialization format
            SerializationError - File failed to be deserialized into the protobuf message
        """

        extension_lc = extension.lower()
        extension_map = self._serialize_extension_method_map()

        if extension_lc not in extension_map:
            raise UnknownExtensionError("Cannot detect message format for %s extension" % extension_lc)

        try:
            msg = extension_map[extension_lc](self, pb_msg)
        except Exception as e:
            raise SerializationError(e)

        return msg


class VnfdSerializer(ProtoMessageSerializer):
    """ Creates a serializer for the VNFD descriptor"""
    def __init__(self):
        super().__init__(VnfdYang, VnfdYang.YangData_Vnfd_VnfdCatalog_Vnfd)


class NsdSerializer(ProtoMessageSerializer):
    """ Creates a serializer for the NSD descriptor"""
    def __init__(self):
        super().__init__(NsdYang, NsdYang.YangData_Nsd_NsdCatalog_Nsd)


class RwVnfdSerializer(ProtoMessageSerializer):
    """ Creates a serializer for the VNFD descriptor"""
    def __init__(self):
        super().__init__(RwVnfdYang, RwVnfdYang.YangData_Vnfd_VnfdCatalog_Vnfd)


class RwNsdSerializer(ProtoMessageSerializer):
    """ Creates a serializer for the NSD descriptor"""
    def __init__(self):
        super().__init__(RwNsdYang, RwNsdYang.YangData_Nsd_NsdCatalog_Nsd)