particle-os/debian-forge
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

format/v2: generalize stage module validation

Browse source 
Generalize the code that validates the stage `inputs`, so it can
be used for future extensions of the stage with new sub-modules.
This commit is contained in:
Christian Kellner 2021-06-07 10:53:48 +00:00 • committed by Tom Gundersen
parent 1ed85dc790
commit 26b15a062d
1 changed files with 12 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

20
osbuild/formats/v2.py
View file

@ -3,7 +3,7 @@
Second, and current, version of the manifest description Second, and current, version of the manifest description
""" """
from typing import Dict from typing import Dict
from osbuild.meta import Index, ValidationResult from osbuild.meta import Index, ModuleInfo, ValidationResult
from ..inputs import Input from ..inputs import Input
from ..pipeline import Manifest, Pipeline, Stage, detect_host_runner from ..pipeline import Manifest, Pipeline, Stage, detect_host_runner
from ..sources import Source from ..sources import Source
@ -301,21 +301,25 @@ def validate(manifest: Dict, index: Index) -> ValidationResult:
schema = index.get_schema("Manifest", version="2") schema = index.get_schema("Manifest", version="2")
result = schema.validate(manifest) result = schema.validate(manifest)
def validate_input(ip, path): def validate_module(mod, klass, path):
name = ip["type"] name = mod["type"]
schema = index.get_schema("Input", name, version="2") schema = index.get_schema(klass, name, version="2")
res = schema.validate(ip) res = schema.validate(mod)
result.merge(res, path=path) result.merge(res, path=path)
def validate_stage_modules(klass, stage, path):
group = ModuleInfo.MODULES[klass]
items = stage.get(group, {})
for name, mod in items.items():
validate_module(mod, klass, path + [group, name])
def validate_stage(stage, path): def validate_stage(stage, path):
name = stage["type"] name = stage["type"]
schema = index.get_schema("Stage", name, version="2") schema = index.get_schema("Stage", name, version="2")
res = schema.validate(stage) res = schema.validate(stage)
result.merge(res, path=path) result.merge(res, path=path)
inputs = stage.get("inputs", {}) validate_stage_modules("Input", stage, path)
for name, ip in inputs.items():
validate_input(ip, path + ["inputs", name])
def validate_pipeline(pipeline, path): def validate_pipeline(pipeline, path):
stages = pipeline.get("stages", []) stages = pipeline.get("stages", [])