def load()

in servicecatalog_puppet/manifest_utils.py [0:0]
91 lines of code
21 McCabe index (conditional complexity)

def load(f, puppet_account_id):
    manifest_name = f.name
    manifest = {
        "schema": "puppet-2019-04-01",
        "parameters": {},
        "accounts": [],
        constants.LAUNCHES: {},
        constants.STACKS: {},
        constants.SPOKE_LOCAL_PORTFOLIOS: {},
        constants.ASSERTIONS: {},
        constants.CODE_BUILD_RUNS: {},
        constants.LAMBDA_INVOCATIONS: {},
        constants.APPS: {},
        constants.WORKSPACES: {},
        constants.CFCT: {},
        constants.SERVICE_CONTROL_POLICIES: {},
        constants.SIMULATE_POLICIES: {},
    }
    intrinsic_functions_map = get_intrinsic_functions_map(
        manifest_name, puppet_account_id
    )

    contents = f.read()
    contents = interpolate_intrinsic_functions(contents, intrinsic_functions_map)
    manifest.update(yaml_utils.load(contents))
    d = os.path.dirname(os.path.abspath(f.name))

    extendable = constants.ALL_SECTION_NAMES + ["parameters"]
    for t in extendable:
        t_path = f"{d}{os.path.sep}{t}"
        if os.path.exists(t_path):
            for f in os.listdir(t_path):
                source = f"{t_path}{os.path.sep}{f}"
                with open(source, "r") as file:
                    contents = file.read()
                    contents = interpolate_intrinsic_functions(
                        contents, intrinsic_functions_map
                    )
                    new = yaml_utils.load(contents)
                    for n, v in new.items():
                        if manifest[t].get(n):
                            raise Exception(f"{source} declares a duplicate {t}: {n}")
                    manifest[t].update(new)

    if os.path.exists(f"{d}{os.path.sep}manifests"):
        for f in os.listdir(f"{d}{os.path.sep}manifests"):
            with open(f"{d}{os.path.sep}manifests{os.path.sep}{f}", "r") as file:
                contents = file.read()
                contents = interpolate_intrinsic_functions(
                    contents, intrinsic_functions_map
                )
                ext = yaml_utils.load(contents)
                for t in extendable:
                    manifest[t].update(ext.get(t, {}))

    if os.path.exists(f"{d}{os.path.sep}capabilities"):
        for f in os.listdir(f"{d}{os.path.sep}capabilities"):
            with open(f"{d}{os.path.sep}capabilities{os.path.sep}{f}", "r") as file:
                contents = file.read()
                contents = interpolate_intrinsic_functions(
                    contents, intrinsic_functions_map
                )
                ext = yaml_utils.load(contents)
                always_merger.merge(manifest, ext)

    for config_file in [
        manifest_name.replace(".yaml", ".properties"),
        manifest_name.replace(".yaml", f"-{puppet_account_id}.properties"),
    ]:
        parser = configparser.ConfigParser(
            interpolation=configparser.BasicInterpolation()
        )
        parser.optionxform = str

        if os.path.exists(config_file):
            logger.info(f"reading {config_file}")
            parser.read(config_file)

            for section_name, section_values in parser.items():
                if section_name == "DEFAULT":
                    continue
                for item_name, item_value in section_values.items():
                    name, property_name = item_name.split(".")
                    if property_name != "version":
                        raise Exception(
                            "You can only specify a version in the properties file"
                        )
                    if manifest.get(section_name, {}).get(name):
                        manifest[section_name][name][property_name] = item_value
                    else:
                        logger.warning(
                            f"Could not find manifest[{section_name}][{name}]"
                        )

    for section in constants.ALL_SPOKE_EXECUTABLE_SECTION_NAMES:
        for name, details in manifest.get(section, {}).items():
            if details.get("execution") is None:
                details["execution"] = constants.EXECUTION_MODE_DEFAULT
    return manifest