Profile checkers

check_profile_to_providers.py

import json
import sys


def find_rp(provider_name, providers):
    for entry in providers:
        if entry["namespace"].lower() == provider_name.lower():
            return entry

def find_resource_type(provider_object, resource_type_name):
    for entry in provider_object["resourceTypes"]:
        if entry["resourceType"].lower() == resource_type_name.lower():
            return entry


def doit(profile):
    with open("providers_list.json", 'r') as fd:
        providers = json.load(fd)

    with open(profile, 'r') as fd:
        profile = json.load(fd)

    for provider_name, api_versions in profile["resource-manager"].items():
        provider_object = find_rp(provider_name, providers)

        if not provider_object:
            print("Did NOT found RP: "+provider_name)
            continue
        #print("Found provider entry: "+provider_name)

        for api_version, rt_list in api_versions.items():
            for resource_type_name in rt_list:
                rt_object = find_resource_type(provider_object, resource_type_name)

                if not rt_object:
                    print("Did NOT found RP {} RT {}".format(provider_name, resource_type_name))
                    continue
                #print("Found resource type: "+resource_type_name)

                # Filter api_version now

                if api_version.lower() in rt_object["apiVersions"]:
                    print("Found RP {} RT {} and ApiVersion {} in providers list".format(
                        provider_name,
                        resource_type_name,
                        api_version
                    ))
                else:
                    print("Did NOT found RP {} RT {} and ApiVersion {} in providers list. Found API version are: {}".format(
                        provider_name,
                        resource_type_name,
                        api_version,
                        rt_object["apiVersions"]
                    ))




if __name__ == "__main__":
    doit(sys.argv[1])

index_swagger_files.py

#! py -3.7

import json
import logging
from pathlib import Path
import re

_LOGGER = logging.getLogger(__name__)


def extract_from_path_string(path_string):

    providers = path_string.rfind("providers/")
    sub_path = path_string[providers + len("providers/"):]

    split_path = sub_path.split("/")
    rp = split_path.pop(0)
    if not rp.lower().startswith("microsoft."):
        raise ValueError("RP must start with Microsoft. something")

    rt_parts = []
    for part_path in split_path:
        if "{" in part_path:
            continue
        rt_parts.append(part_path)
    rt = "/".join(rt_parts)

    return rp, rt

def extract_rt(swagger_path):
    _LOGGER.info(f"Working on file {swagger_path}")
    with open(swagger_path, "r", encoding='utf-8-sig') as fd:
        json_data = json.load(fd)
        found_values = set()
        for path in json_data["paths"].keys():
            _LOGGER.info(f"\tWorking on path '{path}'")
            try:
                provider, resource_type = extract_from_path_string(path)
                found_values.add((provider, resource_type))
            except Exception:
                _LOGGER.warning(f"\tUnable to extract anything")
            else:
                _LOGGER.info(f"\tRP={provider}\tRT={resource_type}")

        return {
            json_data['info']['version']: list(found_values)
        }

def walk_rest_api_repo(base_dir=Path(".")):

    json_result = {}
    for path in base_dir.rglob("specification/*/resource-manager/**/*.json"):
        if "examples" in str(path):
            continue

        file_result = extract_rt(path)
        json_result[path.as_posix()] = file_result

    with open("rt_result.json", "w") as fd:
        json.dump(json_result, fd)

def revert_index(index_path=Path("rt_result.json")):
    """Build:
    Provider -> RT -> api-version -> file
    """
    with open(index_path, "r") as fd:
        json_raw = json.load(fd)

    result = {}
    for path in json_raw:
        for api_version in json_raw[path]:
            for rp, rt in json_raw[path][api_version]:
                result.setdefault(rp.lower(), {}).setdefault(rt.lower(), {})[api_version] = path

    with open("inverted_index.json", "w") as fd_write:
        json.dump(result, fd_write)


if __name__ == "__main__":
    logging.basicConfig(
        level=logging.INFO,
        #filename='rtextract.log',
        filemode='w'
    )
    walk_rest_api_repo()
    revert_index()

solve_profile.py

import json
from pathlib import Path
import sys


def solve_profile(profile_path):
    with open(profile_path, "r") as fd:
        data = json.load(fd)

    with open("inverted_index.json", "r") as fd:
        inverted_json = json.load(fd)

    profile_name = data['info']['name']
    print(f"Solving {profile_name}")
    for rp, api_versions in data["resource-manager"].items():
        for api_version, rt_list in api_versions.items():
            for rt_item in rt_list:
                print(f"Solve RP: {rp}, ApiVersion: {api_version}, RT: {rt_item} to:")
                try:
                    filepath = inverted_json[rp.lower()][rt_item.lower()][api_version.lower()]
                    print(f"\t{filepath}")
                except KeyError:
                    print("\tDidn't find it")

if __name__ == "__main__":
    solve_profile(sys.argv[1])