#
# 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.pipeline import detect_host_runner
from .. import test
class TestDescriptions(unittest.TestCase):
def test_canonical(self):
"""Degenerate case. Make sure we always return the same canonical
description when passing empty or null values."""
cases = [
{},
{"assembler": None},
{"stages": []},
{"build": {}},
{"build": None}
]
for pipeline in cases:
with self.subTest(pipeline):
self.assertEqual(osbuild.load(pipeline, {}).description(), {})
def test_stage(self):
name = "org.osbuild.test"
options = {"one": 1}
cases = [
(osbuild.Stage(name, {}, None, None, {}), {"name": name}),
(osbuild.Stage(name, {}, None, None, None), {"name": name}),
(osbuild.Stage(name, {}, None, None, options), {"name": name, "options": options}),
]
for stage, description in cases:
with self.subTest(description):
self.assertEqual(stage.description(), description)
@unittest.skipUnless(test.TestBase.can_bind_mount(), "root-only")
def test_stage_run(self):
stage = osbuild.Stage("org.osbuild.noop", {}, None, None, {})
with tempfile.TemporaryDirectory() as tmpdir:
data = pathlib.Path(tmpdir, "data")
cache = pathlib.Path(tmpdir, "cache")
root = pathlib.Path("/")
runner = detect_host_runner()
monitor = NullMonitor(sys.stderr.fileno())
libdir = os.path.abspath(os.curdir)
for p in [data, cache]:
p.mkdir()
res = stage.run(data, runner, root, cache, monitor, libdir)
self.assertEqual(res.success, True)
self.assertEqual(res.id, stage.id)
def test_assembler(self):
name = "org.osbuild.test"
options = {"one": 1}
cases = [
(osbuild.Assembler(name, None, None, {}), {"name": name}),
(osbuild.Assembler(name, None, None, None), {"name": name}),
(osbuild.Assembler(name, None, None, options), {"name": name, "options": options}),
]
for assembler, description in cases:
with self.subTest(description):
self.assertEqual(assembler.description(), description)
@unittest.skipUnless(test.TestBase.can_bind_mount(), "root-only")
def test_assembler_run(self):
asm = osbuild.Assembler("org.osbuild.noop", None, None, {})
with tempfile.TemporaryDirectory() as tmpdir:
data = pathlib.Path(tmpdir, "data")
cache = pathlib.Path(tmpdir, "cache")
output = pathlib.Path(tmpdir, "output")
root = pathlib.Path("/")
runner = detect_host_runner()
monitor = NullMonitor(sys.stderr.fileno())
libdir = os.path.abspath(os.curdir)
for p in [data, cache, output]:
p.mkdir()
res = asm.run(data, runner, root, monitor, libdir, output)
self.assertEqual(res.success, True)
self.assertEqual(res.id, asm.id)
def test_pipeline(self):
build = osbuild.Pipeline("org.osbuild.test")
build.add_stage("org.osbuild.test", {}, {"one": 1})
pipeline = osbuild.Pipeline("org.osbuild.test", build)
pipeline.add_stage("org.osbuild.test", {}, {"one": 2})
pipeline.set_assembler("org.osbuild.test")
self.assertEqual(pipeline.description(), {
"build": {
"pipeline": {
"stages": [
{
"name": "org.osbuild.test",
"options": {"one": 1}
}
]
},
"runner": "org.osbuild.test"
},
"stages": [
{
"name": "org.osbuild.test",
"options": {"one": 2}
}
],
"assembler": {
"name": "org.osbuild.test"
}
})
def test_moduleinfo(self):
index = osbuild.meta.Index(os.curdir)
modules = []
for klass in ("Stage", "Assembler", "Source"):
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.schema, name)
res = schema.check()
if not res:
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_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)
def test_validation(self):
index = osbuild.meta.Index(os.curdir)
# an empty manifest is OK
res = osbuild.meta.validate({}, index)
self.assertEqual(res.valid, True)
# something totally invalid (by Ondřej Budai)
totally_invalid = {
"osbuild": {
"state": "awesome",
"but": {
"input-validation": 1
}
}
}
res = osbuild.meta.validate(totally_invalid, index)
self.assertEqual(res.valid, False)
# The top-level 'osbuild' is an additional property
self.assertEqual(len(res), 1)
# This is missing the runner
no_runner = {
"pipeline": {
"build": {
"pipeline": {}
}
}
}
res = osbuild.meta.validate(no_runner, index)
self.assertEqual(res.valid, False)
self.assertEqual(len(res), 1) # missing runner
lst = res[".pipeline.build"]
self.assertEqual(len(lst), 1)
# de-dup issues: the manifest checking will report
# the extra element and the recursive build pipeline
# check will also report that same error; make sure
# they get properly de-duplicated
no_runner_extra = {
"pipeline": {
"build": { # missing runner
"pipeline": {
"extra": True, # should not be there
"stages": [{
"name": "org.osbuild.chrony",
"options": {
"timeservers": "string" # should be an array
}
}]
}
}
}
}
res = osbuild.meta.validate(no_runner_extra, index)
self.assertEqual(res.valid, False)
self.assertEqual(len(res), 3)
lst = res[".pipeline.build.pipeline"]
self.assertEqual(len(lst), 1) # should only have one
lst = res[".pipeline.build.pipeline.stages[0].options.timeservers"]
self.assertEqual(len(lst), 1) # should only have one
# stage issues
stage_check = {
"pipeline": {
"stages": [{
"name": "org.osbuild.grub2",
"options": {
"uefi": {
"install": False,
# missing "vendor"
},
# missing rootfs or root_fs_uuid
}
}]
}
}
res = osbuild.meta.validate(stage_check, index)
self.assertEqual(res.valid, False)
self.assertEqual(len(res), 2)
lst = res[".pipeline.stages[0].options"]
self.assertEqual(len(lst), 1) # missing rootfs
lst = res[".pipeline.stages[0].options.uefi"]
self.assertEqual(len(lst), 1) # missing "osname"
assembler_check = {
"pipeline": {
"assembler": {
"name": "org.osbuild.tar",
"options": {
"compression": "foobar"
}
}
}
}
res = osbuild.meta.validate(assembler_check, index)
self.assertEqual(res.valid, False)
self.assertEqual(len(res), 2)
lst = res[".pipeline.assembler.options"]
self.assertEqual(len(lst), 1) # missing "filename"
lst = res[".pipeline.assembler.options.compression"]
self.assertEqual(len(lst), 1) # wrong compression method