stages(erofs): add tests and fix small bug in options handling

stages/org.osbuild.erofs

@@ -84,6 +84,7 @@ def main(inputs, output_dir, options):
             method += f",{compression['level']}"
         cmd += ["-z", method]
 
+    options = options.get("options")
     if options:
         cmd += ["-E", ",".join(options)]
 

stages/test/test_erofs.py

@@ -0,0 +1,122 @@
+#!/usr/bin/python3
+
+import os.path
+import subprocess
+from unittest import mock
+
+import pytest
+
+import osbuild.meta
+from osbuild.testutil.imports import import_module_from_path
+
+
+def make_fake_input_tree(tmpdir, fake_content: dict) -> str:
+    basedir = os.path.join(tmpdir, "tree")
+    for path, content in fake_content.items():
+        dirp, name = os.path.split(os.path.join(basedir, path.lstrip("/")))
+        os.makedirs(dirp, exist_ok=True)
+        with open(os.path.join(dirp, name), "w", encoding="utf-8") as fp:
+            fp.write(content)
+    return basedir
+
+
+def have_erofs() -> bool:
+    try:
+        r = subprocess.run(["mkfs.erofs", "--help"], stderr=subprocess.DEVNULL)
+    except FileNotFoundError:
+        return False
+    return r.returncode == 0
+
+
+@pytest.mark.skipif(not have_erofs(), reason="no mkfs.erofs")
+def test_erofs_integration(tmp_path):
+    fake_input_tree = make_fake_input_tree(tmp_path, {
+        "/file-in-root.txt": "other content",
+        "/subdir/file-in-subdir.txt": "subdir content",
+    })
+    inputs = {
+        "tree": {
+            "path": fake_input_tree,
+        }
+    }
+
+    stage_path = os.path.join(os.path.dirname(__file__), "../org.osbuild.erofs")
+    stage = import_module_from_path("erofs_stage", stage_path)
+    filename = "some-file.img"
+    options = {
+        "filename": filename,
+    }
+
+    stage.main(inputs, tmp_path, options)
+
+    img_path = os.path.join(tmp_path, "some-file.img")
+    assert os.path.exists(img_path)
+    # todo: run dump.erofs
+
+
+@mock.patch("subprocess.run")
+@pytest.mark.parametrize("test_options,expected", [
+    ({}, []),
+    ({"compression": {"method": "lz4"}}, ["-z", "lz4"]),
+    ({"compression": {"method": "lz4", "level": 9}}, ["-z", "lz4,9"]),
+    ({"options": ["dedupe"]}, ["-E", "dedupe"]),
+    ({"options": ["all-fragments", "force-inline-compact"]}, ["-E", "all-fragments,force-inline-compact"]),
+])
+def test_erofs(mock_run, tmp_path, test_options, expected):
+    fake_input_tree = make_fake_input_tree(tmp_path, {
+        "/some-dir/some-file.txt": "content",
+    })
+    inputs = {
+        "tree": {
+            "path": fake_input_tree,
+        }
+    }
+
+    stage_path = os.path.join(os.path.dirname(__file__), "../org.osbuild.erofs")
+    stage = import_module_from_path("erofs_stage", stage_path)
+    filename = "some-file.img"
+    options = {
+        "filename": filename,
+    }
+    options.update(test_options)
+
+    stage.main(inputs, tmp_path, options)
+
+    expected = [
+        "mkfs.erofs",
+        f"{os.path.join(tmp_path, filename)}",
+        f"{fake_input_tree}",
+    ] + expected
+    mock_run.assert_called_with(expected, check=True)
+
+
+@pytest.mark.parametrize("test_data,expected_err", [
+    # bad
+    ({"extra": "option"}, "'extra' was unexpected"),
+    ({"compression": {"method": "invalid"}}, "'invalid' is not one of ["),
+    ({"compression": {"method": "lz4", "level": "string"}}, "'string' is not of type "),
+    # good
+    ({"compression": {"method": "lz4"}}, ""),
+])
+def test_schema_validation_erofs(test_data, expected_err):
+    name = "org.osbuild.erofs"
+    root = os.path.join(os.path.dirname(__file__), "../..")
+    mod_info = osbuild.meta.ModuleInfo.load(root, "Stage", name)
+    schema = osbuild.meta.Schema(mod_info.get_schema(version="2"), name)
+
+    test_input = {
+        "type": "org.osbuild.erofs",
+        "options": {
+            "filename": "some-filename.img",
+        }
+    }
+    test_input["options"].update(test_data)
+    res = schema.validate(test_input)
+
+    if expected_err == "":
+        assert res.valid is True, f"err: {[e.as_dict() for e in res.errors]}"
+    else:
+        assert res.valid is False
+        assert len(res.errors) == 1, [e.as_dict() for e in res.errors]
+        err_msgs = [e.as_dict()["message"] for e in res.errors]
+        assert expected_err in err_msgs[0]