stages: add a new mkswap stage

A stage very similar to the existing mkfs ones (actually, I started from
the xfs one). It creates a swap area on a given device.
This commit is contained in:
Ondřej Budai 2024-08-28 13:23:49 +02:00 committed by Michael Vogt
parent 83fcc8a0b1
commit 2f84ba96c9
3 changed files with 170 additions and 0 deletions

25
stages/org.osbuild.mkswap Executable file
View file

@ -0,0 +1,25 @@
#!/usr/bin/python3
import subprocess
import sys
import osbuild.api
def main(devices, options):
device = devices["device"]["path"]
uuid = options["uuid"]
label = options.get("label")
opts = []
if label:
opts = ["--label", label]
subprocess.run(["mkswap", "--uuid", uuid] + opts + [device],
encoding='utf8', check=True)
if __name__ == '__main__':
args = osbuild.api.arguments()
ret = main(args["devices"], args["options"])
sys.exit(ret)

View file

@ -0,0 +1,40 @@
{
"summary": "Construct a swap area via mkswap(8)",
"description": [
"Construct a swap area with the given options at the device",
"specified via `device`.",
"Buildhost commands used: `mkswap`."
],
"schema_2": {
"devices": {
"type": "object",
"additionalProperties": true,
"required": [
"device"
],
"properties": {
"device": {
"type": "object",
"additionalProperties": true
}
}
},
"options": {
"additionalProperties": false,
"required": [
"uuid"
],
"properties": {
"uuid": {
"description": "UUID for the file system",
"type": "string"
},
"label": {
"description": "Label for the file system",
"type": "string",
"maxLength": 12
}
}
}
}
}

105
stages/test/test_mkswap.py Normal file
View file

@ -0,0 +1,105 @@
#!/usr/bin/python3
import os.path
import subprocess
import uuid
from unittest import mock
import pytest
from osbuild import testutil
from osbuild.testutil import has_executable
STAGE_NAME = "org.osbuild.mkswap"
# Prepare dataset containing good and bad API call parameters
@pytest.mark.parametrize("test_data, expected_err", [
# Bad API parameters
({}, "'uuid' is a required property"),
({"uuid": 123}, "123 is not of type 'string'"),
({"uuid": "text", "label": 1234}, "1234 is not of type 'string'"),
({"uuid": "text", "label": "12345678901234567"}, "is too long"),
# Good API parameters
# mkswap requires valid UUID, but API will accept any string right now.
# Bellow will fail when API will be more restricted in the future.
({"uuid": "text"}, ""),
({"uuid": "text", "label": "123456789012"}, ""),
])
# This test validates only API calls using correct and incorrect queries
def test_schema_validation_mkswap(stage_schema, test_data, expected_err):
test_input = {
"type": STAGE_NAME,
"devices": {
"device": {
"path": "some-path",
},
},
"options": {
}
}
test_input["options"].update(test_data)
res = stage_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
testutil.assert_jsonschema_error_contains(res, expected_err, expected_num_errs=1)
# This test creates dummy swap using predefined parameters and verifies
# that end result has those parameters set
@pytest.mark.skipif(not has_executable("mkswap") or not has_executable("swaplabel"), reason="need mkswap and swaplabel")
def test_mkswap_integration(tmp_path, stage_module):
fake_disk_path = tmp_path / "fake.img"
with fake_disk_path.open("w") as fp:
fp.truncate(300 * 1024 * 1024)
devices = {
"device": {
"path": fake_disk_path,
},
}
fake_uuid = str(uuid.uuid4())
fake_label = "osbuild"
options = {
"uuid": fake_uuid,
"label": fake_label,
}
stage_module.main(devices, options)
assert os.path.exists(fake_disk_path)
output = subprocess.check_output([
"swaplabel", fake_disk_path], encoding="utf-8")
assert f'UUID: {fake_uuid}' in output, \
f'expected UUID not found in: {output}'
assert f'LABEL: {fake_label}' in output, \
f'expected label not found in: {output}'
@pytest.mark.parametrize("test_input, expected", [
({}, []),
({"label": "osbuild"}, ["--label", "osbuild"]),
])
@mock.patch("subprocess.run")
def test_mkswap_cmdline(mock_run, stage_module, test_input, expected):
fake_disk_path = "/dev/xxd1"
devices = {
"device": {
"path": fake_disk_path,
},
}
fake_uuid = str(uuid.uuid4())
options = {
"uuid": fake_uuid,
}
options.update(test_input)
stage_module.main(devices, options)
expected = [
"mkswap",
"--uuid",
fake_uuid,
] + expected + [
fake_disk_path,
]
mock_run.assert_called_once_with(expected, encoding="utf8", check=True)