749912c75a
New function that take a list of pipelines and return the list of pipelines that need to be build, i.e. the pipelines and all their dependencies that are not already present in the store. Add corresponding test.
298 lines
9.9 KiB
Python
298 lines
9.9 KiB
Python
#
|
|
# Basic tests for a collection of osbuild modules.
|
|
#
|
|
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
import osbuild
|
|
import osbuild.meta
|
|
from osbuild.monitor import NullMonitor
|
|
from osbuild.objectstore import ObjectStore
|
|
from osbuild.pipeline import Manifest, detect_host_runner
|
|
from .. import test
|
|
|
|
|
|
def names(*lst):
|
|
return [x.name for x in lst]
|
|
|
|
|
|
class MockStore:
|
|
def __init__(self) -> None:
|
|
self.have = set()
|
|
|
|
def contains(self, pipeline_id):
|
|
return pipeline_id in self.have
|
|
|
|
|
|
class TestDescriptions(unittest.TestCase):
|
|
|
|
@unittest.skipUnless(test.TestBase.can_bind_mount(), "root-only")
|
|
def test_stage_run(self):
|
|
index = osbuild.meta.Index(os.curdir)
|
|
info = index.get_module_info("Stage", "org.osbuild.noop")
|
|
stage = osbuild.Stage(info, {}, None, None, {})
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
|
|
data = pathlib.Path(tmpdir, "data")
|
|
storedir = pathlib.Path(tmpdir, "store")
|
|
root = pathlib.Path("/")
|
|
runner = detect_host_runner()
|
|
monitor = NullMonitor(sys.stderr.fileno())
|
|
libdir = os.path.abspath(os.curdir)
|
|
store = ObjectStore(storedir)
|
|
data.mkdir()
|
|
|
|
res = stage.run(data, runner, root, store, monitor, libdir)
|
|
|
|
self.assertEqual(res.success, True)
|
|
self.assertEqual(res.id, stage.id)
|
|
|
|
def test_manifest(self):
|
|
index = osbuild.meta.Index(os.curdir)
|
|
|
|
info = index.get_module_info("Stage", "org.osbuild.noop")
|
|
|
|
manifest = Manifest()
|
|
|
|
# each pipeline gets a noop stage with different
|
|
# options so that their ids are different
|
|
build = manifest.add_pipeline("build", None, None)
|
|
build.add_stage(info, {"option": 1})
|
|
|
|
tree = manifest.add_pipeline("tree",
|
|
"org.osbuild.linux",
|
|
build.id)
|
|
tree.add_stage(info, {"option": 2})
|
|
|
|
assembler = manifest.add_pipeline("assembler",
|
|
"org.osbuild.inux",
|
|
build.id)
|
|
assembler.add_stage(info, {"option": 3})
|
|
|
|
self.assertEqual(len(manifest.pipelines), 3)
|
|
|
|
self.assertIn("build", manifest.pipelines)
|
|
self.assertIn("tree", manifest.pipelines)
|
|
self.assertIn("assembler", manifest.pipelines)
|
|
|
|
self.assertIn("build", manifest)
|
|
self.assertIn("tree", manifest)
|
|
self.assertIn("assembler", manifest)
|
|
|
|
# make sure the order is correct
|
|
lst = ["build", "tree", "assembler"]
|
|
for a, b in zip(manifest.pipelines, lst):
|
|
self.assertEqual(a, b)
|
|
|
|
for a, b in zip(manifest, [build, tree, assembler]):
|
|
self.assertEqual(a.name, b.name)
|
|
|
|
# check we get exceptions on unknown names
|
|
with self.assertRaises(KeyError):
|
|
_ = manifest.pipelines["foo"]
|
|
|
|
with self.assertRaises(KeyError):
|
|
_ = manifest["foo"]
|
|
|
|
# check helper functions
|
|
# access by name
|
|
check = manifest["build"]
|
|
self.assertEqual(build.name, check.name)
|
|
|
|
check = manifest["tree"]
|
|
self.assertEqual(tree.name, check.name)
|
|
|
|
check = manifest["assembler"]
|
|
self.assertEqual(assembler.name, check.name)
|
|
|
|
# `None` return for unknown items
|
|
check = manifest.get("foo")
|
|
self.assertIsNone(check)
|
|
|
|
# id based access
|
|
for i in [build, tree, assembler]:
|
|
check = manifest[i.id]
|
|
self.assertEqual(i.name, check.name)
|
|
|
|
# pylint: disable=too-many-statements
|
|
def test_on_demand(self):
|
|
index = osbuild.meta.Index(os.curdir)
|
|
|
|
manifest = Manifest()
|
|
noop = index.get_module_info("Stage", "org.osbuild.noop")
|
|
noip = index.get_module_info("Input", "org.osbuild.noop")
|
|
|
|
# the shared build pipeline
|
|
build = manifest.add_pipeline("build", None, None)
|
|
build.add_stage(noop, {"option": 1})
|
|
|
|
# a pipeline simulating some intermediate artefact
|
|
# that other pipeline need as dependency
|
|
dep = manifest.add_pipeline("dep",
|
|
"org.osbuild.linux",
|
|
build.id)
|
|
|
|
dep.add_stage(noop, {"option": 2})
|
|
|
|
# a pipeline that is not linked to the "main"
|
|
# assembler artefact and thus should normally
|
|
# not be built unless explicitly requested
|
|
# has an input that depends on `dep`
|
|
ul = manifest.add_pipeline("unlinked",
|
|
"org.osbuild.linux",
|
|
build.id)
|
|
|
|
stage = ul.add_stage(noop, {"option": 3})
|
|
ip = stage.add_input("dep", noip, "org.osbuild.pipeline")
|
|
ip.add_reference(dep.id)
|
|
|
|
# the main os root file system
|
|
rootfs = manifest.add_pipeline("rootfs",
|
|
"org.osbuild.inux",
|
|
build.id)
|
|
stage = rootfs.add_stage(noop, {"option": 4})
|
|
|
|
# the main raw image artefact, depending on "dep" and
|
|
# "rootfs"
|
|
image = manifest.add_pipeline("image",
|
|
"org.osbuild.inux",
|
|
build.id)
|
|
|
|
stage = image.add_stage(noop, {"option": 5})
|
|
ip = stage.add_input("dep", noip, "org.osbuild.pipeline")
|
|
ip.add_reference(dep.id)
|
|
|
|
stage = image.add_stage(noop, {"option": 6})
|
|
|
|
# a stage using the rootfs as input (named 'image')
|
|
ip = stage.add_input("image", noip, "org.osbuild.pipeline")
|
|
ip.add_reference(rootfs.id)
|
|
|
|
# some compression of the image, like a qcow2
|
|
qcow2 = manifest.add_pipeline("qcow2",
|
|
"org.osbuild.inux",
|
|
build.id)
|
|
stage = qcow2.add_stage(noop, {"option": 7})
|
|
ip = stage.add_input("image", noip, "org.osbuild.pipeline")
|
|
ip.add_reference(image.id)
|
|
|
|
fmt = index.get_format_info("osbuild.formats.v2").module
|
|
self.assertIsNotNone(fmt)
|
|
print(json.dumps(fmt.describe(manifest), indent=2))
|
|
|
|
# The pipeline graph in the manifest with dependencies:
|
|
# ├─╼ build
|
|
# ├─╼ dep
|
|
# │ └ build
|
|
# ├─╼ unlinked
|
|
# │ ├ build
|
|
# │ └ dep
|
|
# ├─╼ rootfs
|
|
# │ └ build
|
|
# ├─╼ image
|
|
# │ ├ build
|
|
# │ ├ dep
|
|
# │ └ rootfs
|
|
# └─╼ qcow2
|
|
# ├ build
|
|
# └ image
|
|
|
|
store = MockStore()
|
|
|
|
# check an empty input leads to an empty list
|
|
res = manifest.depsolve(store, [])
|
|
assert res == []
|
|
|
|
# the build pipeline should resolve to just itself
|
|
res = manifest.depsolve(store, names(build))
|
|
assert res == names(build)
|
|
|
|
# if we build the 'unlinked' pipeline, we get it
|
|
# and its dependencies, dep and build
|
|
res = manifest.depsolve(store, names(ul))
|
|
assert res == names(build, dep, ul)
|
|
|
|
# building image with nothing in the store should
|
|
# result in all pipelines but 'unlinked'
|
|
res = manifest.depsolve(store, names(image))
|
|
assert res == names(build, rootfs, dep, image)
|
|
|
|
# if we have the 'dep' dependency in the store,
|
|
# we should be not be building that
|
|
store.have.add(dep.id)
|
|
res = manifest.depsolve(store, names(image))
|
|
assert res == names(build, rootfs, image)
|
|
|
|
# if we only have the build pipeline in the
|
|
# store we should not build that
|
|
store.have.clear()
|
|
store.have.add(build.id)
|
|
res = manifest.depsolve(store, names(image))
|
|
assert res == names(rootfs, dep, image)
|
|
|
|
# if we have the final artefact in the store,
|
|
# nothing should be built at all
|
|
store.have.clear()
|
|
store.have.add(image.id)
|
|
res = manifest.depsolve(store, names(image))
|
|
assert res == []
|
|
|
|
# we have a checkpoint of the stage in the image
|
|
# pipeline with the `dep` dependency, so that
|
|
# it effectively only depends on `rootfs`
|
|
store.have.clear()
|
|
store.have.add(image.stages[0].id)
|
|
res = manifest.depsolve(store, names(image))
|
|
assert res == names(build, rootfs, image)
|
|
|
|
def check_moduleinfo(self, version):
|
|
index = osbuild.meta.Index(os.curdir)
|
|
|
|
modules = []
|
|
for klass in ("Assembler", "Input", "Source", "Stage"):
|
|
mods = index.list_modules_for_class(klass)
|
|
modules += [(klass, module) for module in mods]
|
|
|
|
self.assertTrue(modules)
|
|
|
|
for module in modules:
|
|
klass, name = module
|
|
try:
|
|
info = osbuild.meta.ModuleInfo.load(os.curdir, klass, name)
|
|
schema = osbuild.meta.Schema(info.get_schema(version), name)
|
|
res = schema.check()
|
|
if not res:
|
|
err = "SCHEMA: " + json.dumps(schema.data, indent=2) + "\n"
|
|
err += "\n ".join(str(e) for e in res)
|
|
self.fail(str(res) + "\n " + err)
|
|
except json.decoder.JSONDecodeError as e:
|
|
msg = f"{klass} '{name}' has invalid STAGE_OPTS\n\t" + str(e)
|
|
self.fail(msg)
|
|
|
|
def test_moduleinfo(self):
|
|
for version in ["1", "2"]:
|
|
with self.subTest(version=version):
|
|
self.check_moduleinfo(version)
|
|
|
|
def test_schema(self):
|
|
schema = osbuild.meta.Schema(None)
|
|
self.assertFalse(schema)
|
|
|
|
schema = osbuild.meta.Schema({"type": "bool"}) # should be 'boolean'
|
|
self.assertFalse(schema.check().valid)
|
|
self.assertFalse(schema)
|
|
|
|
schema = osbuild.meta.Schema({"type": "array", "minItems": 3})
|
|
self.assertTrue(schema.check().valid)
|
|
self.assertTrue(schema)
|
|
|
|
res = schema.validate([1, 2])
|
|
self.assertFalse(res)
|
|
res = schema.validate([1, 2, 3])
|
|
self.assertTrue(res)
|