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

test: convert to new osbuild executor

Browse source 
Convert the pipeline tests to the new osbuild executor. This will remove
the last users of the "output_id" and "tree_id" identifiers from
osbuild.
This commit is contained in:
David Rheinsberg 2020-05-20 11:15:13 +02:00
parent 845148993c
commit c84f5280c1
3 changed files with 193 additions and 79 deletions
Download patch file Download diff file Expand all files Collapse all files

89
test/test_assemblers.py
View file

@ -6,36 +6,52 @@ import json
import subprocess
import tempfile
import time
import unittest
from . import osbuildtest
from . import test
class TestAssemblers(osbuildtest.TestCase):
class TestAssemblers(unittest.TestCase, test.TestBase):
@classmethod
def setUpClass(cls):
super().setUpClass()
subprocess.run(["modprobe", "nbd"], check=False)
def run_assembler(self, name, options):
with open("test/pipelines/f30-base.json") as f:
base = json.load(f)
base["pipeline"] = dict(base["pipeline"], assembler={"name": name, "options": options})
return self.run_osbuild("-", input_data=json.dumps(base))
def setUp(self):
self.osbuild = test.OSBuild(self)
@contextlib.contextmanager
def run_assembler(self, name, options, output_path):
with self.osbuild as osb:
with open("test/pipelines/f30-base.json") as f:
manifest = json.load(f)
manifest["pipeline"] = dict(
manifest["pipeline"],
assembler={"name": name, "options": options}
)
data = json.dumps(manifest)
treeid = osb.treeid_from_manifest(data)
assert treeid
osb.compile(data, checkpoints=[treeid])
with osb.map_object(treeid) as tree, \
osb.map_output(output_path) as output:
yield tree, output
def assertImageFile(self, filename, fmt, expected_size=None):
info = json.loads(subprocess.check_output(["qemu-img", "info", "--output", "json", filename]))
self.assertEqual(info["format"], fmt)
self.assertEqual(info["virtual-size"], expected_size)
def assertFilesystem(self, device, uuid, fstype, tree_id):
def assertFilesystem(self, device, uuid, fstype, tree):
output = subprocess.check_output(["blkid", "--output", "export", device], encoding="utf-8")
blkid = dict(line.split("=") for line in output.strip().split("\n"))
self.assertEqual(blkid["UUID"], uuid)
self.assertEqual(blkid["TYPE"], fstype)
with mount(device) as target_tree:
tree = f"{self.store}/refs/{tree_id}"
diff = json.loads(subprocess.check_output(["./tree-diff", tree, target_tree]))
diff = self.tree_diff(tree, target_tree)
self.assertEqual(diff["added_files"], ["/lost+found"])
self.assertEqual(diff["deleted_files"], [])
self.assertEqual(diff["differences"], {})
@ -64,17 +80,18 @@ class TestAssemblers(osbuildtest.TestCase):
self.assertEqual(bootable.count(True), 1)
self.assertEqual(bootable.index(True) + 1, boot_partition)
@unittest.skipUnless(test.TestBase.have_tree_diff(), "tree-diff missing")
def test_rawfs(self):
options = {
"filename": "image.raw",
"root_fs_uuid": "016a1cda-5182-4ab3-bf97-426b00b74eb0",
"size": 2 * 1024 * 1024 * 1024
}
tree_id, output_id = self.run_assembler("org.osbuild.rawfs", options)
image = f"{self.store}/refs/{output_id}/image.raw"
self.assertImageFile(image, "raw", options["size"])
self.assertFilesystem(image, options["root_fs_uuid"], "ext4", tree_id)
with self.run_assembler("org.osbuild.rawfs", options, "image.raw") as (tree, image):
self.assertImageFile(image, "raw", options["size"])
self.assertFilesystem(image, options["root_fs_uuid"], "ext4", tree)
@unittest.skipUnless(test.TestBase.have_tree_diff(), "tree-diff missing")
def test_qemu(self):
for fmt in ["raw", "raw.xz", "qcow2", "vmdk", "vdi"]:
with self.subTest(fmt=fmt):
@ -86,20 +103,25 @@ class TestAssemblers(osbuildtest.TestCase):
"root_fs_uuid": "aff010e9-df95-4f81-be6b-e22317251033",
"size": 2 * 1024 * 1024 * 1024
}
tree_id, output_id = self.run_assembler("org.osbuild.qemu", options)
image = f"{self.store}/refs/{output_id}/image.{fmt}"
if fmt == "raw.xz":
subprocess.run(["unxz", "--keep", "--force", image], check=True)
image = image[:-3]
fmt = "raw"
self.assertImageFile(image, fmt, options["size"])
with nbd_connect(image) as device:
self.assertPartitionTable(device, "dos", options["ptuuid"], 1, boot_partition=1)
self.assertGRUB2(device,
"26e3327c6b5ac9b5e21d8b86f19ff7cb4d12fb2d0406713f936997d9d89de3ee",
"18031c9465e3f9ccb9aeb9c8e59dec6b82e91376e2628c8100b5461af62ad67c",
1024 * 1024)
self.assertFilesystem(device + "p1", options["root_fs_uuid"], "ext4", tree_id)
with self.run_assembler("org.osbuild.qemu",
options,
f"image.{fmt}") as (tree, image):
if fmt == "raw.xz":
subprocess.run(["unxz", "--keep", "--force", image], check=True)
image = image[:-3]
fmt = "raw"
self.assertImageFile(image, fmt, options["size"])
with nbd_connect(image) as device:
self.assertPartitionTable(device,
"dos",
options["ptuuid"],
1,
boot_partition=1)
self.assertGRUB2(device,
"26e3327c6b5ac9b5e21d8b86f19ff7cb4d12fb2d0406713f936997d9d89de3ee",
"18031c9465e3f9ccb9aeb9c8e59dec6b82e91376e2628c8100b5461af62ad67c",
1024 * 1024)
self.assertFilesystem(device + "p1", options["root_fs_uuid"], "ext4", tree)
def test_tar(self):
cases = [
@ -110,11 +132,12 @@ class TestAssemblers(osbuildtest.TestCase):
options = {"filename": filename}
if compression:
options["compression"] = compression
_, output_id = self.run_assembler("org.osbuild.tar", options)
image = f"{self.store}/refs/{output_id}/{filename}"
output = subprocess.check_output(["file", "--mime-type", image], encoding="utf-8")
_, mimetype = output.strip().split(": ") # "filename: mimetype"
self.assertIn(mimetype, expected_mimetypes)
with self.run_assembler("org.osbuild.tar",
options,
filename) as (_, image):
output = subprocess.check_output(["file", "--mime-type", image], encoding="utf-8")
_, mimetype = output.strip().split(": ") # "filename: mimetype"
self.assertIn(mimetype, expected_mimetypes)
@contextlib.contextmanager

54
test/test_boot.py
View file

@ -2,35 +2,43 @@
import os
import subprocess
import tempfile
import unittest
from . import osbuildtest
from . import test
class TestBoot(osbuildtest.TestCase):
class TestBoot(unittest.TestCase):
def setUp(self):
self.osbuild = test.OSBuild(self)
def test_boot(self):
_, output_id = self.run_osbuild("test/pipelines/f30-boot.json")
#
# Build an image and test-boot it.
#
with tempfile.TemporaryDirectory() as d:
output_file = os.path.join(d, "output")
with self.osbuild as osb:
osb.compile_file("test/pipelines/f30-boot.json")
with osb.map_output("f30-boot.qcow2") as qcow2, \
tempfile.TemporaryDirectory() as d:
subprocess.run(["qemu-system-x86_64",
"-snapshot",
"-m", "1024",
"-M", "accel=kvm:hvf:tcg",
output_file = os.path.join(d, "output")
subprocess.run(["qemu-system-x86_64",
"-snapshot",
"-m", "1024",
"-M", "accel=kvm:hvf:tcg",
# be silent
"-nographic",
"-monitor", "none",
"-serial", "none",
# be silent
"-nographic",
"-monitor", "none",
"-serial", "none",
# create /dev/vport0p1
"-chardev", f"file,path={output_file},id=stdio",
"-device", "virtio-serial",
"-device", "virtserialport,chardev=stdio",
# create /dev/vport0p1
"-chardev", f"file,path={output_file},id=stdio",
"-device", "virtio-serial",
"-device", "virtserialport,chardev=stdio",
f"{self.get_path_to_store(output_id)}/f30-boot.qcow2"],
encoding="utf-8",
check=True)
with open(output_file, "r") as f:
self.assertEqual(f.read().strip(), "running")
qcow2],
encoding="utf-8",
check=True)
with open(output_file, "r") as f:
self.assertEqual(f.read().strip(), "running")

129
test/test_stages.py
View file

@ -1,48 +1,131 @@
import difflib
import json
import os
import pprint
import tempfile
import unittest
from test import osbuildtest
from . import test
class TestDescriptions(osbuildtest.TestCase):
class TestDescriptions(unittest.TestCase, test.TestBase):
def assertTreeDiffsEqual(self, tree_diff1, tree_diff2):
"""
Asserts two tree diffs for equality.
Before assertion, the two trees are sorted, therefore order of files
doesn't matter.
There's a special rule for asserting differences where we don't
know the exact before/after value. This is useful for example if
the content of file is dependant on current datetime. You can use this
feature by putting null value in difference you don't care about.
Example:
"/etc/shadow": {content: ["sha256:xxx", null]}
In this case the after content of /etc/shadow doesn't matter.
The only thing that matters is the before content and that
the content modification happened.
"""
def _sorted_tree(tree):
sorted_tree = json.loads(json.dumps(tree, sort_keys=True))
sorted_tree["added_files"] = sorted(sorted_tree["added_files"])
sorted_tree["deleted_files"] = sorted(sorted_tree["deleted_files"])
return sorted_tree
tree_diff1 = _sorted_tree(tree_diff1)
tree_diff2 = _sorted_tree(tree_diff2)
def raise_assertion(msg):
diff = '\n'.join(
difflib.ndiff(
pprint.pformat(tree_diff1).splitlines(),
pprint.pformat(tree_diff2).splitlines(),
)
)
raise AssertionError(f"{msg}\n\n{diff}")
self.assertEqual(tree_diff1['added_files'], tree_diff2['added_files'])
self.assertEqual(tree_diff1['deleted_files'], tree_diff2['deleted_files'])
if len(tree_diff1['differences']) != len(tree_diff2['differences']):
raise_assertion('length of differences different')
for (file1, differences1), (file2, differences2) in \
zip(tree_diff1['differences'].items(), tree_diff2['differences'].items()):
if file1 != file2:
raise_assertion(f"filename different: {file1}, {file2}")
if len(differences1) != len(differences2):
raise_assertion("length of file differences different")
for (difference1_kind, difference1_values), (difference2_kind, difference2_values) in \
zip(differences1.items(), differences2.items()):
if difference1_kind != difference2_kind:
raise_assertion(f"different difference kinds: {difference1_kind}, {difference2_kind}")
if difference1_values[0] is not None \
and difference2_values[0] is not None \
and difference1_values[0] != difference2_values[0]:
raise_assertion(f"before values are different: {difference1_values[0]}, {difference2_values[0]}")
if difference1_values[1] is not None \
and difference2_values[1] is not None \
and difference1_values[1] != difference2_values[1]:
raise_assertion(f"after values are different: {difference1_values[1]}, {difference2_values[1]}")
def setUp(self):
self.osbuild = test.OSBuild(self)
def run_stage_test(self, test_dir: str):
pipeline1 = f"{test_dir}/a.json"
pipeline2 = f"{test_dir}/b.json"
tree_id1, _ = self.run_osbuild(pipeline1)
tree_id2, _ = self.run_osbuild(pipeline2)
with self.osbuild as osb:
with tempfile.TemporaryDirectory() as empty:
if tree_id1:
tree1 = self.get_path_to_store(tree_id1)
else:
tree1 = empty
def run(path):
checkpoints = []
context = None
if tree_id2:
tree2 = self.get_path_to_store(tree_id2)
else:
tree2 = empty
with open(path, "r") as f:
data = f.read()
actual_diff = self.run_tree_diff(tree1, tree2)
tree = osb.treeid_from_manifest(data)
if tree:
checkpoints += [tree]
context = osb.map_object(tree)
with open(f"{test_dir}/diff.json") as f:
expected_diff = json.load(f)
osb.compile(data, checkpoints=checkpoints)
return context
self.assertTreeDiffsEqual(expected_diff, actual_diff)
ctx_a = run(f"{test_dir}/a.json")
ctx_b = run(f"{test_dir}/b.json")
ctx_a = ctx_a or tempfile.TemporaryDirectory()
ctx_b = ctx_b or tempfile.TemporaryDirectory()
with ctx_a as tree1, ctx_b as tree2:
actual_diff = self.tree_diff(tree1, tree2)
with open(f"{test_dir}/diff.json") as f:
expected_diff = json.load(f)
self.assertTreeDiffsEqual(expected_diff, actual_diff)
def generate_test_case(test):
def generate_test_case(path):
@unittest.skipUnless(test.TestBase.have_tree_diff(), "tree-diff missing")
def test_case(self):
self.run_stage_test(test)
self.run_stage_test(path)
return test_case
def init_tests():
test_dir = 'test/stages_tests'
for test in os.listdir(test_dir):
setattr(TestDescriptions, f"test_{test}", generate_test_case(f"{test_dir}/{test}"))
for name in os.listdir(test_dir):
setattr(TestDescriptions, f"test_{name}", generate_test_case(f"{test_dir}/{name}"))
init_tests()