AltoParser.py

"""
Library for ALTO project robot system test framework.
Author: linxiao9292@outlook.com
"""

import json
import re

content_key_set = {"meta", "resources"}
resource_key_set = {"uri", "media-type", "accepts", "capabilities", "uses"}
cost_type_key_set = {"cost-mode", "cost-metric", "description"}
media_type_set = {
    "application/alto-directory+json",
    "application/alto-networkmap+json",
    "application/alto-networkmapfilter+json",
    "application/alto-costmap+json",
    "application/alto-costmapfilter+json",
    "application/alto-endpointprop+json",
    "application/alto-endpointpropparams+json",
    "application/alto-endpointcost+json",
    "application/alto-endpointcostparams+json",
    "application/alto-error+json",
}


def get_basic_info(response):
    """Get basic information of the simple IRD.

    Args:
        :param response: response from restconf/operational/alto-simple-ird:information
            contains context-id and base-url for alto-simple-ird.
    Returns:
        :returns tuple: context-id - Identifier of different implementations of one service.
            The formation of it is xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
            For example, we have three implementations of IRD service, and we could use
            different context-id to identify them.

                        base-url - ALTO northbound URL of simple IRD.
    """
    resp = json.loads(response)
    return resp["information"]["context-id"], resp["information"]["base-url"]


def check_ird_configuration_entry(response, ird_resource_id, context_id, resource_id):
    """Check whether resources we added are in the resource pool.

    Args:
        :param response: response from restconf/operational/alto-resourcepool:context/context-id
            context-id is from get_basic_info(response).

        :param ird_resource_id: ID of the IRD.

        :param context_id: See above.

        :param resource_id: ID of the resource.
            context-id and resource-id together could determine one implementation of one service.
    Returns:
        :return bool: False if we do not get the resource we added before
    """
    resp = json.loads(response)
    resources = resp["context"][0]["resource"]
    for resource in resources:
        if resource["resource-id"] == ird_resource_id:
            context_tags = resource["context-tag"]
            for tag in context_tags:
                if "dependency" in tag:
                    for one_dependency in tag["dependency"]:
                        _context_id = re.findall(
                            r"\d{8}-\d{4}-\d{4}-\d{4}-\d{12}", one_dependency
                        )[0]
                        if _context_id == context_id:
                            long_resource_id = re.findall(
                                r"resource-id='[a-zA-Z\-]*'", one_dependency
                            )[0]
                            short_resource_id = re.findall("'.*'", long_resource_id)[0]
                            _resource_id = short_resource_id.replace("'", "")
                            if _resource_id == resource_id:
                                return True
    return False


def verify_ird(response):
    """Semantic check of IRD response, more information in RFC 7285 9.2.

    Args:
        :param response: response from ALTO northbound URL of IRD.
    Returns:
        :return: bool: False if there are some semantic errors.
            One semantic error is that we only define routing-cost in cost-type, but we find that one capability of a
            resource is bandwidth.
    """
    if response.headers["content-type"] != "application/alto-directory+json":
        return False
    try:
        resp = json.loads(response.content)
    except ValueError:
        return False
    if "meta" not in resp:
        return False
    meta = resp["meta"]
    if "cost-types" in meta:
        cost_types = meta["cost-types"]
        for cost_type in cost_types:
            if set(cost_type).issubset(cost_type_key_set):
                if "cost-mode" in cost_type and "cost-metric" in cost_type:
                    continue
                else:
                    return False
            else:
                return False

    resources = resp["resources"]
    for resource in resources.keys():
        if set(resources[resource].keys()).issubset(resource_key_set):
            if (
                "uri" not in resources[resource]
                or "media-type" not in resources[resource]
            ):
                return False
            else:
                _resource = resources[resource]
                media_type = _resource["media-type"]
                if media_type not in media_type_set:
                    return False
                if "capabilities" in _resource:
                    capabilities = _resource["capabilities"]
                    if "cost-type-names" in capabilities:
                        cost_type_names = capabilities["cost-type-names"]
                        for cost_type_name in cost_type_names:
                            if cost_type_name not in cost_types:
                                return False
        else:
            return False
    return True