diff --git a/tools/mpp-import-pipeline.py b/tools/mpp-import-pipeline.py new file mode 100755 index 00000000..ab51d9f7 --- /dev/null +++ b/tools/mpp-import-pipeline.py @@ -0,0 +1,149 @@ +#!/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. + +The parameters for this pre-processor look like this: + +``` +... + "mpp-import-pipeline": { + "path": "./manifest.json" + } +... +``` +""" + +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(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(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 _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) + _manifest_parse(state, src) + + for todo in state.manifest_todo: + _manifest_process(state, todo) + + json.dump(state.manifest, sys.stdout, indent=2) + sys.stdout.write("\n") + + +def main() -> int: + args = _main_args(sys.argv) + with _main_state(args) as state: + _main_process(state) + + return 0 + + +if __name__ == "__main__": + sys.exit(main())