81c8374d3e
The `org.osbuild.files` source provides files, but might in the future not be the only one that does. Therefore rename it to match the internal tool that is being used to fetch the files. This is done for most other osbuild modules that target tools. The format v1 loader is adapted to make this change transparent for users of the v1 format, so we are backwards compatible. Change the MPP depsolve preprocessor so that for format v2 based manifest `org.osbuild.curl` source is used. Also rename the corresponding source test. Adapt the format v2 mod test to use the curl source.
286 lines
9.3 KiB
Python
286 lines
9.3 KiB
Python
""" Version 1 of the manifest description
|
|
|
|
This is the first version of the osbuild manifest description,
|
|
that has a "main" pipeline that consists of zero or more stages
|
|
to create a tree and optionally one assembler that assembles
|
|
the created tree into an artefact. The pipeline can have any
|
|
number of nested build pipelines. A sources section is used
|
|
to fetch resources.
|
|
"""
|
|
from typing import Dict, Optional, Tuple
|
|
from osbuild.meta import Index, ValidationResult
|
|
from ..pipeline import Manifest, Pipeline, detect_host_runner
|
|
|
|
|
|
VERSION = "1"
|
|
|
|
|
|
def describe(manifest: Manifest, *, with_id=False) -> Dict:
|
|
"""Create the manifest description for the pipeline"""
|
|
def describe_stage(stage):
|
|
description = {"name": stage.name}
|
|
if stage.options:
|
|
description["options"] = stage.options
|
|
if with_id:
|
|
description["id"] = stage.id
|
|
return description
|
|
|
|
def describe_pipeline(pipeline: Pipeline) -> Dict:
|
|
description = {}
|
|
if pipeline.build:
|
|
build = manifest[pipeline.build]
|
|
description["build"] = {
|
|
"pipeline": describe_pipeline(build),
|
|
"runner": pipeline.runner
|
|
}
|
|
|
|
if pipeline.stages:
|
|
stages = [describe_stage(s) for s in pipeline.stages]
|
|
description["stages"] = stages
|
|
|
|
return description
|
|
|
|
def get_source_name(source):
|
|
name = source.info.name
|
|
if name == "org.osbuild.curl":
|
|
name = "org.osbuild.files"
|
|
return name
|
|
|
|
pipeline = describe_pipeline(manifest["tree"])
|
|
|
|
assembler = manifest.get("assembler")
|
|
if assembler:
|
|
description = describe_stage(assembler.stages[0])
|
|
pipeline["assembler"] = description
|
|
|
|
description = {"pipeline": pipeline}
|
|
|
|
if manifest.sources:
|
|
sources = {
|
|
get_source_name(s): s.options
|
|
for s in manifest.sources
|
|
}
|
|
description["sources"] = sources
|
|
|
|
return description
|
|
|
|
|
|
def load_assembler(description: Dict, index: Index, manifest: Manifest):
|
|
pipeline = manifest["tree"]
|
|
|
|
build, base, runner = pipeline.build, pipeline.id, pipeline.runner
|
|
name, options = description["name"], description.get("options", {})
|
|
|
|
# Add a pipeline with one stage for our assembler
|
|
pipeline = manifest.add_pipeline("assembler", runner, build)
|
|
pipeline.export = True
|
|
|
|
info = index.get_module_info("Assembler", name)
|
|
|
|
stage = pipeline.add_stage(info, options, {})
|
|
info = index.get_module_info("Input", "org.osbuild.tree")
|
|
ip = stage.add_input("tree", info, "org.osbuild.pipeline")
|
|
ip.add_reference(base)
|
|
return pipeline
|
|
|
|
|
|
def load_build(description: Dict, index: Index, manifest: Manifest, n: int):
|
|
pipeline = description.get("pipeline")
|
|
if pipeline:
|
|
build_pipeline = load_pipeline(pipeline, index, manifest, n + 1)
|
|
else:
|
|
build_pipeline = None
|
|
|
|
return build_pipeline, description["runner"]
|
|
|
|
|
|
def load_stage(description: Dict, index: Index, pipeline: Pipeline):
|
|
name = description["name"]
|
|
opts = description.get("options", {})
|
|
info = index.get_module_info("Stage", name)
|
|
|
|
stage = pipeline.add_stage(info, opts)
|
|
|
|
if stage.name == "org.osbuild.rpm":
|
|
info = index.get_module_info("Input", "org.osbuild.files")
|
|
ip = stage.add_input("packages", info, "org.osbuild.source")
|
|
for pkg in stage.options["packages"]:
|
|
options = None
|
|
if isinstance(pkg, dict):
|
|
gpg = pkg.get("check_gpg")
|
|
if gpg:
|
|
options = {"metadata": {"rpm.check_gpg": gpg}}
|
|
pkg = pkg["checksum"]
|
|
ip.add_reference(pkg, options)
|
|
elif stage.name == "org.osbuild.ostree":
|
|
info = index.get_module_info("Input", "org.osbuild.ostree")
|
|
ip = stage.add_input("commits", info, "org.osbuild.source")
|
|
commit, ref = opts["commit"], opts.get("ref")
|
|
options = {"ref": ref} if ref else None
|
|
ip.add_reference(commit, options)
|
|
|
|
|
|
def load_source(name: str, description: Dict, index: Index, manifest: Manifest):
|
|
if name == "org.osbuild.files":
|
|
name = "org.osbuild.curl"
|
|
|
|
info = index.get_module_info("Source", name)
|
|
|
|
if name == "org.osbuild.curl":
|
|
items = description["urls"]
|
|
elif name == "org.osbuild.ostree":
|
|
items = description["commits"]
|
|
else:
|
|
raise ValueError(f"Unknown source type: {name}")
|
|
|
|
# NB: the entries, i.e. `urls`, `commits` are left in the
|
|
# description dict, although the sources are not using
|
|
# it anymore. The reason is that it makes `describe` work
|
|
# without any special casing
|
|
|
|
manifest.add_source(info, items, description)
|
|
|
|
|
|
def load_pipeline(description: Dict, index: Index, manifest: Manifest, n: int = 0) -> Pipeline:
|
|
build = description.get("build")
|
|
if build:
|
|
build_pipeline, runner = load_build(build, index, manifest, n)
|
|
else:
|
|
build_pipeline, runner = None, detect_host_runner()
|
|
|
|
# the "main" pipeline is called `tree`, since it is building the
|
|
# tree that will later be used by the `assembler`. Nested build
|
|
# pipelines will get call "build", and "build-build-...", where
|
|
# the number of repetitions is equal their level of nesting
|
|
if not n:
|
|
name = "tree"
|
|
else:
|
|
name = "-".join(["build"] * n)
|
|
|
|
build_id = build_pipeline and build_pipeline.id
|
|
pipeline = manifest.add_pipeline(name, runner, build_id)
|
|
|
|
for stage in description.get("stages", []):
|
|
load_stage(stage, index, pipeline)
|
|
|
|
return pipeline
|
|
|
|
|
|
def load(description: Dict, index: Index) -> Manifest:
|
|
"""Load a manifest description"""
|
|
|
|
pipeline = description.get("pipeline", {})
|
|
sources = description.get("sources", {})
|
|
|
|
manifest = Manifest()
|
|
|
|
load_pipeline(pipeline, index, manifest)
|
|
|
|
# load the assembler, if any
|
|
assembler = pipeline.get("assembler")
|
|
if assembler:
|
|
load_assembler(assembler, index, manifest)
|
|
|
|
# load the sources
|
|
for name, desc in sources.items():
|
|
load_source(name, desc, index, manifest)
|
|
|
|
for pipeline in manifest.pipelines.values():
|
|
for stage in pipeline.stages:
|
|
stage.sources = sources
|
|
|
|
return manifest
|
|
|
|
|
|
def get_ids(manifest: Manifest) -> Tuple[Optional[str], Optional[str]]:
|
|
pipeline = manifest["tree"]
|
|
assembler = manifest.get("assembler")
|
|
return pipeline.id, assembler and assembler.id
|
|
|
|
|
|
def output(manifest: Manifest, res: Dict) -> Dict:
|
|
"""Convert a result into the v1 format"""
|
|
|
|
def result_for_pipeline(pipeline):
|
|
# The pipeline might not have been built one of its
|
|
# dependencies, i.e. its build pipeline, failed to
|
|
# build. We thus need to be tolerant of a missing
|
|
# result but still need to to recurse
|
|
current = res.get(pipeline.id, {})
|
|
retval = {
|
|
"success": current.get("success", False)
|
|
}
|
|
if pipeline.build:
|
|
build = manifest[pipeline.build]
|
|
retval["build"] = result_for_pipeline(build)
|
|
stages = current.get("stages")
|
|
if stages:
|
|
retval["stages"] = stages
|
|
assembler = current.get("assembler")
|
|
if assembler:
|
|
retval["assembler"] = assembler
|
|
return retval
|
|
|
|
result = result_for_pipeline(manifest["tree"])
|
|
|
|
assembler = manifest.get("assembler")
|
|
if assembler:
|
|
current = res.get(assembler.id)
|
|
if current:
|
|
result["assembler"] = current["stages"][0]
|
|
|
|
return result
|
|
|
|
|
|
def validate(manifest: Dict, index: Index) -> ValidationResult:
|
|
"""Validate a OSBuild manifest
|
|
|
|
This function will validate a OSBuild manifest, including
|
|
all its stages and assembler and build manifests. It will
|
|
try to validate as much as possible and not stop on errors.
|
|
The result is a `ValidationResult` object that can be used
|
|
to check the overall validation status and iterate all the
|
|
individual validation errors.
|
|
"""
|
|
|
|
schema = index.get_schema("Manifest")
|
|
result = schema.validate(manifest)
|
|
|
|
# main pipeline
|
|
pipeline = manifest.get("pipeline", {})
|
|
|
|
# recursively validate the build pipeline as a "normal"
|
|
# pipeline in order to validate its stages and assembler
|
|
# options; for this it is being re-parented in a new plain
|
|
# {"pipeline": ...} dictionary. NB: Any nested structural
|
|
# errors might be detected twice, but de-duplicated by the
|
|
# `ValidationResult.merge` call
|
|
build = pipeline.get("build", {}).get("pipeline")
|
|
if build:
|
|
res = validate({"pipeline": build}, index=index)
|
|
result.merge(res, path=["pipeline", "build"])
|
|
|
|
stages = pipeline.get("stages", [])
|
|
for i, stage in enumerate(stages):
|
|
name = stage["name"]
|
|
schema = index.get_schema("Stage", name)
|
|
res = schema.validate(stage)
|
|
result.merge(res, path=["pipeline", "stages", i])
|
|
|
|
asm = pipeline.get("assembler", {})
|
|
if asm:
|
|
name = asm["name"]
|
|
schema = index.get_schema("Assembler", name)
|
|
res = schema.validate(asm)
|
|
result.merge(res, path=["pipeline", "assembler"])
|
|
|
|
# sources
|
|
sources = manifest.get("sources", {})
|
|
for name, source in sources.items():
|
|
if name == "org.osbuild.files":
|
|
name = "org.osbuild.curl"
|
|
schema = index.get_schema("Source", name)
|
|
res = schema.validate(source)
|
|
result.merge(res, path=["sources", name])
|
|
|
|
return result
|