debian-forge/tools/mpp-import-pipeline.py

#!/usr/bin/python3

"""Manifest-Pre-Processor - Pipeline Import

This manifest-pre-processor consumes a manifest on stdin, processes it, and
produces the resulting manifest on stdout.

This tool imports a pipeline from another file and inserts it into a manifest
at the same position the import instruction is located. Sources from the
imported manifest are merged with the existing sources.

Manifest format version "1" and "2" are supported.

The parameters for this pre-processor for format version "1" look like this:

```
...
    "mpp-import-pipeline": {
      "path": "./manifest.json"
    }
...
```

The parameters for this pre-processor for format version "2" look like this:

```
...
    "mpp-import-pipeline": {
      "path": "./manifest.json",
      "id:" "build"
    }
...
```
"""

import argparse
import contextlib
import json
import os
import sys


class State:
    cwd = None                          # CurrentWorkingDirectory for imports

    manifest = None                     # Input/Working Manifest
    manifest_urls = None                # Link to sources URL dict
    manifest_todo = []                  # Array of links to import pipelines


def _manifest_enter(manifest, key, default):
    if key not in manifest:
        manifest[key] = default
    return manifest[key]


def _manifest_parse_v1(state, data):
    manifest = data

    # Resolve "sources"."org.osbuild.files"."urls".
    manifest_sources = _manifest_enter(manifest, "sources", {})
    manifest_files = _manifest_enter(manifest_sources, "org.osbuild.files", {})
    manifest_urls = _manifest_enter(manifest_files, "urls", {})

    # Collect import entries in a TO-DO list.
    manifest_todo = []

    # Find the `mpp-import-pipeline` section. We iterate down the buildtrees
    # until we find one. Since an import overrides a possibly existing pipeline
    # only one import needs to be handled (the others would be overridden). We
    # do support multiple, so this can be easily extended in the future.
    current = manifest
    while current:
        if "mpp-import-pipeline" in current:
            manifest_todo.append(current)
            break

        current = current.get("pipeline", {}).get("build")

    # Remember links of interest.
    state.manifest = manifest
    state.manifest_urls = manifest_urls
    state.manifest_todo = manifest_todo


def _manifest_process_v1(state, todo):
    mpp = _manifest_enter(todo, "mpp-import-pipeline", {})
    mpp_path = mpp["path"]

    # Load the to-be-imported manifest.
    with open(os.path.join(state.cwd, mpp_path), "r") as f:
        imp = json.load(f)

    # Resolve keys from the import.
    imp_sources = _manifest_enter(imp, "sources", {})
    imp_files = _manifest_enter(imp_sources, "org.osbuild.files", {})
    imp_urls = _manifest_enter(imp_files, "urls", {})
    imp_pipeline = _manifest_enter(imp, "pipeline", {})

    # We only support importing manifests with URL sources. Other sources are
    # not supported, yet. This can be extended in the future, but we should
    # maybe rather try to make sources generic (and repeatable?), so we can
    # deal with any future sources here as well.
    assert list(imp_sources.keys()) == ["org.osbuild.files"]

    # We import `sources` from the manifest, as well as a pipeline description
    # from the `pipeline` entry. Make sure nothing else is in the manifest, so
    # we do not accidentally miss new features.
    assert list(imp.keys()).sort() == ["pipeline", "sources"].sort()

    # Now with everything imported and verified, we can merge the pipeline back
    # into the original manifest. We take all URLs and merge them in the pinned
    # url-array, and then we take the pipeline and simply override any original
    # pipeline at the position where the import was declared. Lastly, we delete
    # the mpp-import statement.
    state.manifest_urls.update(imp_urls)
    todo["pipeline"] = imp_pipeline
    del(todo["mpp-import-pipeline"])


def _manifest_import_v1(state, src):
    _manifest_parse_v1(state, src)

    for todo in state.manifest_todo:
        _manifest_process_v1(state, todo)


def _manifest_parse_v2(state, manifest):
    todo = []

    pipelines = manifest.get("pipelines", [])

    for pipeline in pipelines:
        current = pipeline.get("mpp-import-pipeline")
        if current:
            todo.append(pipeline)

    state.manifest = manifest
    state.manifest_todo = todo


def _manifest_process_v2(state, todo):
    manifest = state.manifest
    sources = _manifest_enter(manifest, "sources", {})

    mpp = todo["mpp-import-pipeline"]
    path = mpp["path"]

    with open(os.path.join(state.cwd, path), "r") as f:
        imp = json.load(f)

    # merge the sources
    for source, desc in imp.get("sources", {}).items():
        target = sources.get(source)
        if not target:
            # new source, just copy everything
            sources[source] = desc
            continue

        if desc.get("options"):
            options = _manifest_enter(target, "options", {})
            options.update(desc["options"])

        items = _manifest_enter(target, "items", {})
        items.update(desc.get("items", {}))

    # get the pipeline
    pipelines = imp.get("pipelines", [])

    pid = mpp["id"]

    target = None
    for pipeline in pipelines:
        if pipeline["name"] == pid:
            target = pipeline
            break

    if not target:
        raise ValueError(f"Pipeline '{pid}' not found in {path}")

    todo.update(target)
    del(todo["mpp-import-pipeline"])


def _manifest_import_v2(state, src):
    _manifest_parse_v2(state, src)
    for todo in state.manifest_todo:
        _manifest_process_v2(state, todo)


def _main_args(argv):
    parser = argparse.ArgumentParser(description="Generate Test Manifests")

    parser.add_argument(
        "--cwd",
        metavar="PATH",
        type=os.path.abspath,
        default=None,
        help="Current Working Directory for relative import paths",
    )

    return parser.parse_args(argv[1:])


@contextlib.contextmanager
def _main_state(args):
    state = State()
    state.cwd = args.cwd or "."
    yield state


def _main_process(state):
    src = json.load(sys.stdin)
    version = src.get("version", "1")
    if version == "1":
        _manifest_import_v1(state, src)
    elif version == "2":
        _manifest_import_v2(state, src)
    else:
        return 1

    json.dump(state.manifest, sys.stdout, indent=2)
    sys.stdout.write("\n")
    return 0


def main() -> int:
    args = _main_args(sys.argv)
    with _main_state(args) as state:
        res = _main_process(state)

    return res


if __name__ == "__main__":
    sys.exit(main())