diff --git a/.gitignore b/.gitignore index eeb8a6ec4087..adef82b6cc82 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,7 @@ **/__pycache__ + +# coverage.py data file +/.coverage + +# HTML coverage report (default output dir for run-tests.sh --html) +/htmlcov/ diff --git a/README.md b/README.md index 75229421763c..afbb3aa6867f 100644 --- a/README.md +++ b/README.md @@ -5,6 +5,17 @@ Python CLI using Hatch. It either consumes a BitBake-exported environment file or folder (generated via `bitbake -c rootfs_wicenv `) or will invoke BitBake directly (if available on the PATH). +## Contents + +- [Quick start](#quick-start) + - [Using wicenv (one environment file)](#using-wicenv-one-environment-file) + - [Using wicenv (environment folder)](#using-wicenv-environment-folder) + - [With bitbake](#with-bitbake) +- [Project layout](#project-layout) +- [Testing](#testing) +- [Contributing](#contributing) +- [Licensing](#licensing) + ## Quick start ### Using wicenv (one environment file) @@ -46,6 +57,23 @@ environment file or folder (generated via `bitbake -c rootfs_wicenv - `src/wic/*`: core engine, plugins, and helpers. - `src/bb/*`: various bitbake helpers that were brought along and used in other parts of wic. - `src/oe/*`: various oe-core helpers that were brought along and used in other parts of wic. +- `tests/*`: the standalone test suite and its documentation (see Testing below). + +## Testing + +wic ships a standalone test suite under `tests/` that runs from a plain +checkout with nothing but `pytest` -- no bitbake and no OpenEmbedded +build required. Install the test extras and run it: + +``` +pip install -e ".[tests]" +tests/run-tests.sh +``` + +Add branch coverage of the wic source with `tests/run-tests.sh +--coverage`, or lint with `tests/run-tests.sh --lint`. See +[tests/docs/README.md](tests/docs/README.md) for the suite layout, how +to add a test, and the test-driven conventions it follows. ## Contributing diff --git a/pyproject.toml b/pyproject.toml index fdc1ce0f5ece..2898091abc15 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,6 +17,14 @@ classifiers = [ "Programming Language :: Python :: 3", ] +[project.optional-dependencies] +tests = [ + "pytest >= 7.0", + "coverage >= 7.0", + "pytest-cov >= 4.0", + "ruff >= 0.5", +] + [project.urls] Homepage = "https://git.yoctoproject.org/wic" Repository = "https://git.yoctoproject.org/wic" @@ -36,3 +44,25 @@ build-backend = "hatchling.build" [tool.hatch.version] path = "src/wic/cli.py" + +[tool.pytest.ini_options] +# Keep a test's scratch directory only when it fails; a passing test's +# tmp_path is removed automatically so repeated runs do not accumulate +# leftover files under the pytest base temp directory. +tmp_path_retention_policy = "failed" +tmp_path_retention_count = 1 + +[tool.ruff] +# For now only the test suite is actually linted (run-tests.sh --lint +# passes the tests/ path); the wic source under src/ is not yet +# ruff-clean and is left out until its findings are fixed (see +# tests/docs/linting.md). The per-file-ignores below are the only +# intentional, test-specific exceptions. + +[tool.ruff.lint.per-file-ignores] +# Test modules prepend the in-tree src/ directory to sys.path before +# importing wic, so that the suite runs against a checkout that has not +# been installed. That bootstrap necessarily runs before the wic imports, +# which trips E402 (module-level import not at top of file). The ordering +# is deliberate; ignore E402 for the test tree only. +"tests/**" = ["E402"] diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 000000000000..1c368ebf8595 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,38 @@ +# Session banner for the wic test suite. +# +# Before collection, print the environment the tests exercise: +# - the host +# - the Python and pytest versions +# - the wic under test (its version and the module path of its import) + +import platform + +import pytest + + +def _wic_under_test(): + """Return (version, module_path) for the wic being tested.""" + try: + import wic.cli as wic_cli + except Exception as exc: # pragma: no cover - reported in the banner + return ("(import failed: %s)" % exc, "(unimported)") + version = getattr(wic_cli, "__version__", "(unknown)") + module_path = getattr(wic_cli, "__file__", "(unknown)") + return (version, module_path) + + +def _format_banner(): + wic_version, wic_path = _wic_under_test() + lines = [ + "wic test suite", + " host: %s %s %s" % ( + platform.node(), platform.system(), platform.machine()), + " python: %s pytest: %s" % ( + platform.python_version(), pytest.__version__), + " wic: %s (%s)" % (wic_version, wic_path), + ] + return "\n".join(lines) + + +def pytest_report_header(config): + return _format_banner() diff --git a/tests/docs/README.md b/tests/docs/README.md new file mode 100644 index 000000000000..98a8d9d8adf7 --- /dev/null +++ b/tests/docs/README.md @@ -0,0 +1,77 @@ +# wic test suite + +A standalone test suite for the wic source tree. It runs from a plain +checkout with nothing but `pytest` -- no bitbake, no OpenEmbedded +build, and no target image -- so wic's logic can be exercised and kept +stable as the code changes. + +## Contents + +- [Layout](#layout) +- [Running](#running) + - [Options](#options) + - [Passing arguments through to pytest](#passing-arguments-through-to-pytest) +- [Documentation](#documentation) + +## Layout + +``` +tests/ + conftest.py session banner describing the run + run-tests.sh wrapper for running the suite + unit/ unit tests that import wic modules directly + docs/ this documentation +``` + +The current suite is unit-only: every test under `tests/unit/` imports +a wic module in-process and asserts on its behaviour. None of it needs +host tools or a build environment. + +See [Running](#running) for how to invoke `run-tests.sh`. + +## Running + +Install wic with its test extras, then run the suite: + +```bash +pip install -e ".[tests]" +tests/run-tests.sh +``` + +`run-tests.sh` works from anywhere in the checkout. + +### Options + +``` +--coverage run the suite and report branch coverage of src/wic +--html [DIR] also write an HTML coverage report (default: htmlcov/) +--lint run ruff over tests/ instead of the suite +-h, --help print usage and exit +``` + +`--lint` must be used on its own (no other options or pytest arguments). + +`--help` is the authoritative, always-current list; the script may +grow more options than are shown here. + +### Passing arguments through to pytest + +Anything after `--` (or any unrecognised argument) is handed straight +to pytest, so the whole pytest command line is available: + +```bash +tests/run-tests.sh -- -k filemap -v # one area, verbose +tests/run-tests.sh -- tests/unit/test_partition.py +``` + +A healthy run ends in `passed` and `xfailed` with zero `failed` and +zero unexpected `xpassed`. See `philosophy.md` for what `xfailed` +means here. + +## Documentation + +| File | Content | +|------|---------| +| `philosophy.md` | How assertions are written, and what xfail means | +| `authoring.md` | How to add a unit test to the suite | +| `linting.md` | How ruff is used on the test suite | diff --git a/tests/docs/authoring.md b/tests/docs/authoring.md new file mode 100644 index 000000000000..ea1a1c59a048 --- /dev/null +++ b/tests/docs/authoring.md @@ -0,0 +1,97 @@ +# Authoring a unit test + +Unit tests live under `tests/unit/`, one module per area of wic, named +`test_.py`. They import the wic module under test directly and +run in-process, so they need no host tools and no build environment. + +## Contents + +- [Where a test goes](#where-a-test-goes) +- [Shape of a test](#shape-of-a-test) +- [When wic does not do the right thing yet](#when-wic-does-not-do-the-right-thing-yet) +- [Running what you wrote](#running-what-you-wrote) + +## Where a test goes + +Group tests by the wic module they exercise. Existing modules: + +- `test_filemap.py` -- the sparse-file/FIEMAP block mapping +- `test_ksparser_types.py` -- kickstart option type parsing +- `test_partition.py` -- partition geometry and sizing +- `test_oe_path.py` -- the path helpers under `wic/oe/` +- `test_cli_types.py` -- argparse argument types +- `test_update_fstab.py` -- fstab rewriting +- `test_misc.py` -- the misc helpers +- `test_bb_utils.py` -- the small `wic/bb` utilities + +Add a new `test_.py` when you start covering a module that has no +file yet; otherwise extend the existing one. + +## Shape of a test + +```python +import pytest + +from wic.ksparser import sizetype + + +class TestSizetype: + # sizetype(default, size_in_bytes=False) returns a parser f(arg); + # with default "M", a bare number is read as mebibytes-in-KiB. + def test_plain_value_uses_default_unit(self): + parse = sizetype("M") + assert parse("100") == 100 * 1024 + + @pytest.mark.parametrize("arg,expected", [ + ("1K", 1), + ("1G", 1 * 1024 * 1024), + ("0", 0), + ]) + def test_suffixes(self, arg, expected): + assert sizetype("M")(arg) == expected +``` + +Prefer `@pytest.mark.parametrize` to spell each input/output pair as +its own case rather than looping inside one test, so a single bad +value shows up as one named failure. + +Use `pytest`'s `tmp_path` fixture for any scratch files; a passing +test's scratch directory is removed automatically (see the retention +policy in `pyproject.toml`), and a failing one is kept for inspection. + +## When wic does not do the right thing yet + +Assert the correct behaviour, then mark the test `xfail` with a short +reason describing the defect. Do not assert the wrong behaviour. See +`philosophy.md`. + +xfail can be marked two ways. Use the decorator when the whole test is +a pending behaviour: + +```python +@pytest.mark.xfail(reason="sizetype accepts a negative size silently") +def test_rejects_negative(self): + with pytest.raises(ValueError): + sizetype("M")("-1") +``` + +Use the inline call when only one branch is pending, or when the +reason depends on what happened at runtime: + +```python +def test_extra_space_zero_is_honoured(self): + part = parse_one_part("--extra-space 0") + if part.extra_space != 0: + pytest.xfail("--extra-space 0 is overridden by the default") + assert part.extra_space == 0 +``` + +## Running what you wrote + +```bash +tests/run-tests.sh -- tests/unit/test_.py -v +tests/run-tests.sh --coverage -- tests/unit/test_.py +``` + +A new test should leave the suite at zero `failed` and zero unexpected +`xpassed`. diff --git a/tests/docs/linting.md b/tests/docs/linting.md new file mode 100644 index 000000000000..562a6e5c2197 --- /dev/null +++ b/tests/docs/linting.md @@ -0,0 +1,50 @@ +# Linting + +## Contents + +- [tests/ must be clean](#tests-must-be-clean) +- [src/ is not linted yet](#src-is-not-linted-yet) +- [The one intentional exception in tests/](#the-one-intentional-exception-in-tests) + +The test suite is linted with [ruff](https://docs.astral.sh/ruff/). Run +it with: + +```bash +tests/run-tests.sh --lint # ruff over tests/ +``` + +ruff is configured in `pyproject.toml` (`[tool.ruff]`). + +## tests/ must be clean + +Our own test code is held to a clean bar: `tests/run-tests.sh --lint` +reports nothing. If you add a test that trips a rule, fix the test. The +one standing exception is documented below. + +## src/ is not linted yet + +ruff is currently pointed at `tests/` only. The wic source under `src/` +is **not** yet ruff-clean, so it is deliberately left out: enabling it +now would make every `--lint` run noisy with findings whose fixes have +not landed. Once the source is cleaned up, `src/` will be brought under +ruff as well. + +## The one intentional exception in tests/ + +Test modules prepend the in-tree `src/` directory to `sys.path` before +importing `wic`, so the suite runs against a plain checkout that has not +been installed: + +```python +_SRC = Path(__file__).resolve().parent.parent.parent / "src" +if str(_SRC) not in sys.path: + sys.path.insert(0, str(_SRC)) + +from wic.oe.path import realpath +``` + +That bootstrap necessarily runs before the `wic` imports, which trips +`E402` (module-level import not at top of file). The ordering is +required, so `E402` is ignored for the test tree via +`per-file-ignores` in `pyproject.toml`. This is the only rule relaxed +for `tests/`. diff --git a/tests/docs/philosophy.md b/tests/docs/philosophy.md new file mode 100644 index 000000000000..bdeb0f6541a9 --- /dev/null +++ b/tests/docs/philosophy.md @@ -0,0 +1,84 @@ +# Philosophy + +## Contents + +- [Assert what the code should do, not what it does](#assert-what-the-code-should-do-not-what-it-does) +- [xfail is a normal, healthy state](#xfail-is-a-normal-healthy-state) +- [Probe the boundaries](#probe-the-boundaries) +- [Keep the assertion strong](#keep-the-assertion-strong) + +## Assert what the code should do, not what it does + +Every assertion states the behaviour wic is *expected* to provide, +never the behaviour it currently happens to have. A test that locks in +a wrong result is worse than no test: it gives false confidence and it +fights the eventual fix by turning red exactly when someone repairs the +code. + +When wic does not yet meet the asserted behaviour, the test is written +ahead of the fix and marked `xfail` with a short reason. This is a +test-driven style: the specification lands first, captured in the test +itself, and the `xfail` marker is removed when the matching change makes +the test pass. + +Wrong -- bakes in the bug: + +```python +# value lost its comma; asserting the broken output +assert result["file"] == "s3://bucket/a" +``` + +Right -- asserts the correct output, marks it pending: + +```python +assert result["file"] == "s3://bucket/a,b.img" +pytest.xfail("sourceparams value with a comma is silently truncated") +``` + +## xfail is a normal, healthy state + +In this suite an `xfail` means "a behaviour we want, written down and +waiting for the implementation." It is not a flaky test and not a +disabled test. A run is healthy when it ends in `passed` +plus `xfailed` with zero `failed`. An unexpected `xpassed` is a signal +too: the behaviour now works, so the test should become a plain +passing assertion (drop the `xfail`). + +## Probe the boundaries + +For a parameter, probe the boundary and just past it rather than only +the comfortable middle of its range. The classes worth covering: + +- numbers: empty, zero, negative, the maximum and one beyond it, + non-power-of-two and non-multiple-of-1024/1000 values, primes, + fractional values where an integer is assumed, `0` versus `0.0` + versus `"0"`, off-by-one at a block or sector boundary, and very + large magnitudes near `2**31` and `2**63` +- strings: the empty string, whitespace only, embedded spaces, tabs, + newlines and CRLF endings, non-ASCII characters, shell + metacharacters in a value that becomes part of a command line, + leading-dash values that look like options, over-delimited or + malformed forms (stray commas, doubled or missing separators), and + very long values +- paths: traversal (`../`), doubled slashes, a trailing slash or its + absence, `.` and `..` components, broken or looping symlinks, and a + non-existent path +- types and shape: each wrong type the parameter might receive (a + string where an integer is expected, a list where a string is + expected, `None`, and so on), the wrong argument count, and + duplicate keys or entries +- state: a second call that must not see stale cached data, a + `cache=False` path that must evict, calling an operation twice, and + finalising a half-constructed object + +Each case asserts a specific outcome -- an exact value, or a specific +exception -- rather than merely "it did not crash." A test that only +checks for the absence of an exception will pass against badly wrong +output. + +## Keep the assertion strong + +If a test is hard to make pass, the fix is in the code or in +understanding the correct behaviour -- never in weakening the +assertion to something vague enough to pass. Loosening an assertion to +get green is how a suite quietly stops catching regressions. diff --git a/tests/run-tests.sh b/tests/run-tests.sh new file mode 100755 index 000000000000..f9e916531603 --- /dev/null +++ b/tests/run-tests.sh @@ -0,0 +1,135 @@ +#!/usr/bin/env bash +# +# Run the wic test suite. +# +# Thin wrapper around pytest that runs from anywhere in the checkout. +# With --coverage it also measures branch coverage of the wic source; +# with --lint it runs ruff over the tests instead of pytest. +# +# Usage: +# tests/run-tests.sh [--coverage] [--html [DIR]] [-- pytest args] +# tests/run-tests.sh --lint +# +# Options: +# --coverage also measure branch coverage of src/wic and +# print a terminal report with the missing lines +# --html [DIR] also write an HTML coverage report (default dir: +# htmlcov/); implies --coverage +# --lint run ruff over tests/ and exit; must be used on +# its own (no other options or pytest args) +# -h, --help show this help and exit +# +# Anything after `--`, or any unrecognised argument, is passed straight +# through to pytest (e.g. a path, -k EXPR, -v). With no such argument +# the whole suite under tests/ is run. +# +# Examples: +# tests/run-tests.sh # whole suite +# tests/run-tests.sh --coverage # + terminal coverage report +# tests/run-tests.sh --html # + HTML report in htmlcov/ +# tests/run-tests.sh --html /tmp/cov # + HTML report in /tmp/cov +# tests/run-tests.sh --lint # ruff over tests/ +# tests/run-tests.sh -- -k filemap -v # pass args through to pytest +# tests/run-tests.sh -- tests/unit # a single tier or file +# +# Needs the test extras: pip install -e ".[tests]" + +set -euo pipefail + +REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +PY="${PYTHON:-python3}" + +coverage=0 +lint=0 +html=0 +html_dir="htmlcov" +pytest_args=() + +while [ $# -gt 0 ]; do + case "$1" in + --coverage) + coverage=1 + ;; + --html) + coverage=1 + html=1 + # Optional directory argument: consume the next token only if + # it is not another option or the pytest pass-through marker. + case "${2:-}" in + ""|-*|--) ;; + *) html_dir="$2"; shift ;; + esac + ;; + --lint) + lint=1 + ;; + -h|--help) + sed -n '2,/^set -euo/p' "${BASH_SOURCE[0]}" | sed 's/^# \{0,1\}//; /^set -euo/d' + exit 0 + ;; + --) + shift + pytest_args+=("$@") + break + ;; + *) + pytest_args+=("$1") + ;; + esac + shift +done + +cd "$REPO_ROOT" + +# --lint runs ruff and exits, so it must be used on its own. Reject any +# other option or pytest argument loudly instead of silently ignoring it. +if [ "$lint" -eq 1 ] && \ + { [ "$coverage" -eq 1 ] || [ "$html" -eq 1 ] || [ ${#pytest_args[@]} -gt 0 ]; }; then + echo "error: --lint must be used on its own." >&2 + echo " run --lint to lint, or drop it to run the suite." >&2 + exit 2 +fi + +if [ "$lint" -eq 1 ]; then + if ! "$PY" -c "import ruff" >/dev/null 2>&1 && ! command -v ruff >/dev/null 2>&1; then + echo "error: --lint requested but ruff is not installed." >&2 + echo " run: $PY -m pip install -e \".[tests]\"" >&2 + exit 1 + fi + # Lint the test suite only; src/ is not yet ruff-clean and is left + # out until its findings are fixed (see tests/docs/linting.md). + exec "$PY" -m ruff check tests +fi + +# Make sure the interpreter that will run pytest can actually import wic. +# The suites pass even without an install (each adds src/ to sys.path), +# but the session banner and any install-dependent behaviour would be +# wrong, so fail loudly instead of running against the wrong interpreter. +if ! "$PY" -c "import wic" >/dev/null 2>&1; then + echo "error: '$PY' cannot import wic." >&2 + echo " install wic with its test extras:" >&2 + echo " $PY -m pip install -e \".[tests]\"" >&2 + exit 1 +fi + +# Default target: the whole suite. Overridden if the caller passed a +# path or -k/-m selector to pytest. +if [ ${#pytest_args[@]} -eq 0 ]; then + pytest_args=("tests") +fi + +if [ "$coverage" -eq 1 ]; then + if ! "$PY" -c "import pytest_cov" >/dev/null 2>&1; then + echo "error: coverage requested but pytest-cov is not installed." >&2 + echo " run: $PY -m pip install -e \".[tests]\"" >&2 + exit 1 + fi + cov_args=(--cov=wic --cov-branch --cov-report=term-missing) + if [ "$html" -eq 1 ]; then + cov_args+=(--cov-report="html:${html_dir}") + echo "HTML coverage report: ${html_dir}/index.html" >&2 + fi + exec "$PY" -m pytest "${pytest_args[@]}" "${cov_args[@]}" +fi + +exec "$PY" -m pytest "${pytest_args[@]}" diff --git a/tests/unit/test_bb_utils.py b/tests/unit/test_bb_utils.py new file mode 100644 index 000000000000..f419437ae44f --- /dev/null +++ b/tests/unit/test_bb_utils.py @@ -0,0 +1,133 @@ +r""" +Unit tests for wic.bb.utils -- a small standalone subset of BitBake's +bb.utils. The module was previously wholly untested. mkdirhier() is a +mkdir -p wrapper with two guards: an unexpanded-${} check and an OSError +handler that re-checks EEXIST. +""" +import os +import sys +import tempfile +import unittest.mock as mock +from pathlib import Path + +import pytest + +_SRC = Path(__file__).resolve().parent.parent.parent / "src" +if str(_SRC) not in sys.path: + sys.path.insert(0, str(_SRC)) + +from wic.bb.utils import mkdirhier + + +# --------------------------------------------------------------------------- +# Happy path +# --------------------------------------------------------------------------- + +class TestMkdirhierHappyPath: + def test_creates_nested_directory(self): + d = tempfile.mkdtemp() + target = os.path.join(d, "a", "b", "c") + mkdirhier(target) + assert os.path.isdir(target) + + def test_idempotent_on_existing_directory(self): + d = tempfile.mkdtemp() + target = os.path.join(d, "x") + mkdirhier(target) + # Second call must not raise (exist_ok=True). + mkdirhier(target) + assert os.path.isdir(target) + + def test_single_level(self): + d = tempfile.mkdtemp() + target = os.path.join(d, "single") + mkdirhier(target) + assert os.path.isdir(target) + + def test_existing_root_directory(self): + # An already-existing directory (the tmpdir itself) is a no-op. + d = tempfile.mkdtemp() + mkdirhier(d) + assert os.path.isdir(d) + + +# --------------------------------------------------------------------------- +# Unexpanded bitbake-variable guard +# --------------------------------------------------------------------------- + +class TestMkdirhierUnexpandedVar: + def test_unexpanded_var_rejected(self): + with pytest.raises(Exception) as ei: + mkdirhier("/tmp/${WORKDIR}/x") + assert "unexpanded bitbake variable" in str(ei.value) + + def test_unexpanded_var_not_created(self): + d = tempfile.mkdtemp() + bad = os.path.join(d, "${VAR}", "sub") + with pytest.raises(Exception): + mkdirhier(bad) + # Nothing should have been created. + assert not os.path.exists(os.path.join(d, "${VAR}")) + + def test_brace_without_dollar_is_allowed(self): + # Only the literal '${' marker triggers the guard; a bare '{' is a + # legal (if unusual) directory name. + d = tempfile.mkdtemp() + target = os.path.join(d, "plain{brace") + mkdirhier(target) + assert os.path.isdir(target) + + +# --------------------------------------------------------------------------- +# OSError handler references undefined `errno` +# --------------------------------------------------------------------------- + +class TestMkdirhierErrorPath: + def test_oserror_under_a_file_component(self): + # Create a regular file, then try to mkdir a subdir *under* it. On + # Linux os.makedirs raises NotADirectoryError (an OSError subclass), + # which enters the except branch -> the undefined `errno` reference + # raises NameError instead of surfacing the real OSError. + d = tempfile.mkdtemp() + f = os.path.join(d, "afile") + Path(f).write_text("x") + target = os.path.join(f, "sub") + try: + mkdirhier(target) + except NameError: + pytest.xfail("mkdirhier OSError handler references undefined " + "`errno`; raises NameError instead of the real OSError") + except OSError: + return # desired behaviour once errno is imported + pytest.fail("expected an OSError for mkdir under a file component") + + def test_generic_oserror_propagates_without_nameerror(self): + # Force os.makedirs to raise an arbitrary OSError and assert the + # handler does not turn it into a NameError. EACCES != EEXIST, so the + # handler takes the `raise e` path -- but only after evaluating + # errno.EEXIST, which is the bug. + err = OSError() + err.errno = 13 # EACCES + with mock.patch("wic.bb.utils.os.makedirs", side_effect=err): + try: + mkdirhier("/whatever/path") + except NameError: + pytest.xfail("errno not imported; OSError handler raises " + "NameError") + except OSError: + return # desired: the original OSError propagates + pytest.fail("expected the OSError to propagate") + + def test_eexist_oserror_is_swallowed_once_errno_available(self): + # If makedirs raises EEXIST on an existing dir, the handler should + # swallow it. Today it cannot even evaluate the condition (NameError). + d = tempfile.mkdtemp() + err = OSError() + err.errno = 17 # EEXIST + with mock.patch("wic.bb.utils.os.makedirs", side_effect=err): + try: + mkdirhier(d) # d exists and is a dir -> should be swallowed + except NameError: + pytest.xfail("errno not imported; cannot evaluate the " + "EEXIST swallow condition") + # No exception -> correct behaviour. diff --git a/tests/unit/test_bootfiles.py b/tests/unit/test_bootfiles.py new file mode 100644 index 000000000000..510f9f892ad0 --- /dev/null +++ b/tests/unit/test_bootfiles.py @@ -0,0 +1,188 @@ +r""" +Unit tests for wic/oe/bootfiles.py. + +get_boot_files(deploy_dir, boot_files) turns a boot-files spec string +into a list of (src, dst) tuples, expanding '*' globs against +deploy_dir and honouring "src;dst" rename entries. It is pure logic +(a regex tokeniser plus a glob), so it can be exercised directly with a +temp directory and never needs bitbake or host tools. + +The spec grammar comes from the tokeniser + re.findall(r'[\w;\-\./\*]+', boot_files) +which keeps only word characters and the set [ ; - . / * ]. Anything +else (spaces, shell metacharacters, non-ASCII) acts as a separator, +which is the root of several of the bugs probed below. + +Where wic does not yet behave correctly, the test is marked xfail with +a short reason describing the bug. +""" + +import sys +from pathlib import Path + +import pytest + +_SRC = Path(__file__).resolve().parent.parent.parent / "src" +if str(_SRC) not in sys.path: + sys.path.insert(0, str(_SRC)) + +from wic.oe.bootfiles import get_boot_files + + +@pytest.fixture +def deploy(tmp_path): + """A deploy dir with a few predictable artifacts.""" + for name in ("zImage", "u-boot.bin", "a.dtb", "b.dtb"): + (tmp_path / name).write_text(name) + return tmp_path + + +# --------------------------------------------------------------------------- +# Happy path: plain entries, renames, globs +# --------------------------------------------------------------------------- + +class TestGetBootFilesHappyPath: + + def test_none_returns_none(self, deploy): + assert get_boot_files(str(deploy), None) is None + + def test_single_plain_entry(self, deploy): + assert get_boot_files(str(deploy), "zImage") == [("zImage", "zImage")] + + def test_multiple_plain_entries(self, deploy): + result = get_boot_files(str(deploy), "zImage u-boot.bin") + assert result == [("zImage", "zImage"), ("u-boot.bin", "u-boot.bin")] + + def test_rename_entry(self, deploy): + assert get_boot_files(str(deploy), "zImage;kernel") == [("zImage", "kernel")] + + def test_glob_keeps_basename(self, deploy): + result = sorted(get_boot_files(str(deploy), "*.dtb")) + assert result == [("a.dtb", "a.dtb"), ("b.dtb", "b.dtb")] + + def test_glob_into_target_directory(self, deploy): + result = sorted(get_boot_files(str(deploy), "*.dtb;boot/")) + assert result == [("a.dtb", "boot/a.dtb"), ("b.dtb", "boot/b.dtb")] + + def test_glob_no_match_yields_nothing(self, deploy): + assert get_boot_files(str(deploy), "nope*.xyz") == [] + + +# --------------------------------------------------------------------------- +# Degenerate spec strings +# --------------------------------------------------------------------------- + +class TestGetBootFilesDegenerate: + + def test_empty_string_yields_empty_list(self, deploy): + assert get_boot_files(str(deploy), "") == [] + + def test_whitespace_only_yields_empty_list(self, deploy): + assert get_boot_files(str(deploy), " \t ") == [] + + def test_leading_semicolon_is_malformed(self, deploy): + with pytest.raises(ValueError): + get_boot_files(str(deploy), ";dst") + + def test_trailing_semicolon_is_malformed(self, deploy): + with pytest.raises(ValueError): + get_boot_files(str(deploy), "src;") + + +# --------------------------------------------------------------------------- +# Multiple-semicolon entry: must raise a clean, described error +# --------------------------------------------------------------------------- + +class TestGetBootFilesMultipleSemicolons: + """An entry with more than one ';' is malformed. + + The code splits on ';' and unpacks into exactly two names. With two + semicolons the split yields three parts and the unpack blows up with + a bare "too many values to unpack" instead of the intended, + described "Malformed boot file entry" ValueError. + """ + + def test_two_semicolons_raises_described_valueerror(self, deploy): + try: + get_boot_files(str(deploy), "a;b;c") + except ValueError as exc: + if "too many values to unpack" in str(exc): + pytest.xfail( + "multi-semicolon entry 'a;b;c' raises an undescribed " + "'too many values to unpack' instead of a 'Malformed " + "boot file entry' message naming the bad entry") + assert "a;b;c" in str(exc) + else: + pytest.fail("multi-semicolon entry should be rejected") + + +# --------------------------------------------------------------------------- +# Filenames the tokeniser cannot represent (spaces, metachars, non-ASCII) +# --------------------------------------------------------------------------- + +class TestGetBootFilesUnrepresentableNames: + r"""The tokeniser keeps only [\w;\-./*]; everything else is a + separator. Any legitimate filename containing a space, a shell + metacharacter, or a non-ASCII character is therefore silently torn + into pieces (or dropped), corrupting the install list rather than + either handling or rejecting the name. + """ + + def test_space_in_filename_not_silently_split(self, tmp_path): + (tmp_path / "my file.dtb").write_text("x") + result = get_boot_files(str(tmp_path), "my file.dtb") + if result == [("my", "my"), ("file.dtb", "file.dtb")]: + pytest.xfail( + "a space in a boot filename is treated as a separator; " + "'my file.dtb' is split into bogus 'my' and 'file.dtb' " + "entries instead of being kept or rejected") + assert result == [("my file.dtb", "my file.dtb")] + + def test_non_ascii_filename_not_dropped(self, tmp_path): + (tmp_path / "kernel-\u00e9.dtb").write_text("x") + result = get_boot_files(str(tmp_path), "kernel-\u00e9.dtb") + if ("kernel-\u00e9.dtb", "kernel-\u00e9.dtb") not in result: + pytest.xfail( + "a non-ASCII character in a boot filename is treated as a " + "separator; the filename is fractured/dropped instead of " + "preserved or rejected") + assert result == [("kernel-\u00e9.dtb", "kernel-\u00e9.dtb")] + + def test_shell_metacharacters_are_not_smuggled_through(self, deploy): + """Metacharacters must not silently become plausible tokens. + + Each kept token is later interpolated into command strings, so + 'a$(reboot).dtb' decaying into real entries is an injection + hazard; the spec should be rejected rather than quietly cleaned. + """ + result = get_boot_files(str(deploy), "a$(reboot).dtb") + if result == [("a", "a"), ("reboot", "reboot"), (".dtb", ".dtb")]: + pytest.xfail( + "shell metacharacters in a boot-files spec are silently " + "stripped into plausible tokens ('a', 'reboot', '.dtb') " + "instead of the spec being rejected as malformed") + assert result == [] + + +# --------------------------------------------------------------------------- +# Path components in entries +# --------------------------------------------------------------------------- + +class TestGetBootFilesPaths: + + def test_subdir_source_is_preserved(self, tmp_path): + sub = tmp_path / "sub" + sub.mkdir() + (sub / "fw.bin").write_text("x") + assert get_boot_files(str(tmp_path), "sub/fw.bin") == [("sub/fw.bin", "sub/fw.bin")] + + def test_glob_in_subdir_relpath(self, tmp_path): + sub = tmp_path / "dtb" + sub.mkdir() + (sub / "x.dtb").write_text("x") + assert get_boot_files(str(tmp_path), "dtb/*.dtb") == [("dtb/x.dtb", "x.dtb")] + + def test_parent_traversal_entry_kept_verbatim(self, deploy): + # Documents current behaviour: traversal is not specifically + # rejected; the tokeniser keeps '.' and '/', so '../x' survives. + assert get_boot_files(str(deploy), "../x") == [("../x", "../x")] diff --git a/tests/unit/test_cli_types.py b/tests/unit/test_cli_types.py new file mode 100644 index 000000000000..59f98ebc7a46 --- /dev/null +++ b/tests/unit/test_cli_types.py @@ -0,0 +1,288 @@ +""" +Unit tests for wic/cli.py type helpers and argparse actions. + +expandtype(), imgtype(), and RootfsArgAction are probed with boundary +inputs, duplicate keys, multiple colons/equals, empty strings, and None. +""" + +import sys +import argparse +import pytest +from pathlib import Path + +_SRC = Path(__file__).resolve().parent.parent.parent / "src" +if str(_SRC) not in sys.path: + sys.path.insert(0, str(_SRC)) + +from wic.cli import expandtype, RootfsArgAction + + +# --------------------------------------------------------------------------- +# expandtype() — converts "1:100M,2:200M" to {1: bytes, 2: bytes} +# --------------------------------------------------------------------------- + +class TestExpandtype: + + # --- valid inputs ------------------------------------------------------- + + def test_auto_returns_empty_dict(self): + assert expandtype("auto") == {} + + def test_single_M_rule(self): + assert expandtype("1:100M") == {1: 100 * 1024 * 1024} + + def test_single_K_rule(self): + assert expandtype("1:512K") == {1: 512 * 1024} + + def test_single_G_rule(self): + assert expandtype("1:1G") == {1: 1 * 1024 * 1024 * 1024} + + def test_multiple_rules(self): + result = expandtype("1:100M,2:200M") + assert result == {1: 100 * 1024 * 1024, 2: 200 * 1024 * 1024} + + def test_three_rules(self): + result = expandtype("1:1G,2:512M,3:256K") + assert len(result) == 3 + assert result[3] == 256 * 1024 + + def test_case_insensitive_suffix(self): + # The suffix matching uses .upper() — lowercase suffixes work + result_upper = expandtype("1:100M") + result_lower = expandtype("1:100m") + assert result_upper == result_lower + + def test_no_suffix_bare_integer(self): + # A bare integer with no suffix — size.isdigit() True, multiplier=1 + result = expandtype("1:1024") + assert result == {1: 1024} + + def test_zero_size(self): + result = expandtype("1:0") + assert result == {1: 0} + + def test_large_partition_number(self): + result = expandtype("128:100M") + assert 128 in result + + # --- duplicate partition ------------------------------------------ + + @pytest.mark.xfail(reason="expandtype silently overwrites duplicate partition " + "rules; the second '1:...' entry wins without any warning") + def test_duplicate_partition_raises(self): + with pytest.raises(argparse.ArgumentTypeError): + expandtype("1:100M,1:200M") + + def test_duplicate_partition_last_wins(self): + # Document the actual (buggy) behaviour: second entry silently wins + result = expandtype("1:100M,1:200M") + assert result[1] == 200 * 1024 * 1024 + + # --- invalid format errors ---------------------------------------------- + + def test_no_colon_raises(self): + with pytest.raises(argparse.ArgumentTypeError): + expandtype("1M") + + def test_non_digit_partition_raises(self): + with pytest.raises(argparse.ArgumentTypeError): + expandtype("abc:100M") + + def test_non_digit_size_raises(self): + with pytest.raises(argparse.ArgumentTypeError): + expandtype("1:abc") + + def test_negative_size_raises(self): + # "-100M"[:-1] == "-100"; "-100".isdigit() is False → rejected + with pytest.raises(argparse.ArgumentTypeError): + expandtype("1:-100M") + + def test_float_size_raises(self): + with pytest.raises(argparse.ArgumentTypeError): + expandtype("1:1.5M") + + def test_float_partition_raises(self): + with pytest.raises(argparse.ArgumentTypeError): + expandtype("1.5:100M") + + def test_empty_string_raises(self): + # "" split by ',' gives [""] which has no ':' → ValueError → ArgumentTypeError + with pytest.raises(argparse.ArgumentTypeError): + expandtype("") + + def test_none_raises(self): + with pytest.raises((argparse.ArgumentTypeError, TypeError, AttributeError)): + expandtype(None) + + def test_comma_only_raises(self): + with pytest.raises(argparse.ArgumentTypeError): + expandtype(",") + + def test_colon_only_raises(self): + # ":" → part="", size="" → part.isdigit() is False → ArgumentTypeError + with pytest.raises(argparse.ArgumentTypeError): + expandtype(":") + + def test_very_large_partition_number(self): + # Very large int is fine; just a dict key + result = expandtype("9999:1M") + assert 9999 in result + + def test_very_long_string_raises(self): + with pytest.raises((argparse.ArgumentTypeError, MemoryError)): + expandtype("1:" + "M" * 100000) + + +# --------------------------------------------------------------------------- +# imgtype() — parses "image[:partition[/path]]" into a named tuple +# +# imgtype calls os.path.isfile(image) so we need a real file to avoid +# hitting the "not a regular file" error before we reach the split logic. +# We test the split-error path (multiple colons) via a real temp file. +# --------------------------------------------------------------------------- + +class TestImgtype: + + @pytest.fixture + def img_file(self, tmp_path): + f = tmp_path / "test.direct" + f.write_bytes(b"\x00" * 512) + return str(f) + + def test_image_only(self, img_file): + from wic.cli import imgtype + result = imgtype(img_file) + assert result.image == img_file + assert result.part is None + assert result.path is None + + def test_image_with_partition(self, img_file): + from wic.cli import imgtype + result = imgtype("%s:1" % img_file) + assert result.part == "1" + assert result.path == "/" + + def test_image_with_partition_and_path(self, img_file): + from wic.cli import imgtype + result = imgtype("%s:1/etc" % img_file) + assert result.part == "1" + assert result.path == "etc" + + def test_nonexistent_image_raises(self, tmp_path): + from wic.cli import imgtype + with pytest.raises(argparse.ArgumentTypeError): + imgtype(str(tmp_path / "no_such.direct")) + + @pytest.mark.xfail(reason="imgtype raises bare ValueError (too many values " + "to unpack) when the spec contains more than one ':'") + def test_multiple_colons_raises_argument_type_error(self, img_file): + from wic.cli import imgtype + with pytest.raises(argparse.ArgumentTypeError): + imgtype("%s:1:extra" % img_file) + + def test_multiple_colons_raises_something(self, img_file): + from wic.cli import imgtype + with pytest.raises((argparse.ArgumentTypeError, ValueError)): + imgtype("%s:1:extra" % img_file) + + def test_empty_string_raises(self): + from wic.cli import imgtype + with pytest.raises((argparse.ArgumentTypeError, ValueError, TypeError)): + imgtype("") + + def test_none_raises(self): + from wic.cli import imgtype + with pytest.raises((argparse.ArgumentTypeError, TypeError, AttributeError)): + imgtype(None) + + +# --------------------------------------------------------------------------- +# RootfsArgAction — argparse action that builds a rootfs_dir dict +# --------------------------------------------------------------------------- + +class TestRootfsArgAction: + + def _make_namespace(self): + return argparse.Namespace() + + def _invoke(self, value, namespace=None): + if namespace is None: + namespace = self._make_namespace() + parser = argparse.ArgumentParser() + action = RootfsArgAction( + option_strings=["--rootfs-dir"], + dest="rootfs_dir", + ) + action(parser, namespace, value) + return namespace + + # --- valid inputs ------------------------------------------------------- + + def test_plain_path_uses_ROOTFS_DIR_key(self, tmp_path): + ns = self._invoke(str(tmp_path)) + assert ns.rootfs_dir["ROOTFS_DIR"] == str(tmp_path) + + def test_key_equals_path(self, tmp_path): + ns = self._invoke("myroot=%s" % tmp_path) + assert ns.rootfs_dir["myroot"] == str(tmp_path) + + def test_second_call_adds_to_dict(self, tmp_path): + ns = self._make_namespace() + self._invoke(str(tmp_path), ns) + self._invoke("second=%s" % tmp_path, ns) + assert "ROOTFS_DIR" in ns.rootfs_dir + assert "second" in ns.rootfs_dir + + def test_empty_key_via_leading_equals(self): + # "=value" → key="", rootfs_dir="value" + ns = self._invoke("=somepath") + assert "" in ns.rootfs_dir + assert ns.rootfs_dir[""] == "somepath" + + def test_empty_value_via_trailing_equals(self): + ns = self._invoke("mykey=") + assert ns.rootfs_dir["mykey"] == "" + + def test_empty_string_uses_ROOTFS_DIR_key(self): + # '=' not in "" → plain path → key='ROOTFS_DIR', value='' + ns = self._invoke("") + assert ns.rootfs_dir["ROOTFS_DIR"] == "" + + # --- duplicate key: just overwrites (no bug filed, but document it) ----- + + def test_duplicate_key_overwritten(self, tmp_path): + ns = self._make_namespace() + self._invoke("myroot=first", ns) + self._invoke("myroot=second", ns) + assert ns.rootfs_dir["myroot"] == "second" + + # --- known bug — multiple '=' characters -------------------------- + + @pytest.mark.xfail(reason="RootfsArgAction raises bare ValueError when value " + "contains more than one '='; split('=') returns >2 elements") + def test_multiple_equals_raises_argument_type_error(self): + with pytest.raises((argparse.ArgumentTypeError, argparse.ArgumentError)): + self._invoke("key=path=with=extras") + + def test_multiple_equals_raises_something(self): + with pytest.raises((ValueError, argparse.ArgumentTypeError, + argparse.ArgumentError)): + self._invoke("key=path=with=extras") + + # --- None input --------------------------------------------------------- + + def test_none_value_raises(self): + with pytest.raises((TypeError, AttributeError)): + self._invoke(None) + + # --- very long strings -------------------------------------------------- + + def test_very_long_plain_path(self): + long_path = "x" * 10000 + ns = self._invoke(long_path) + assert ns.rootfs_dir["ROOTFS_DIR"] == long_path + + def test_very_long_key(self): + long_key = "k" * 10000 + ns = self._invoke("%s=value" % long_key) + assert long_key in ns.rootfs_dir diff --git a/tests/unit/test_engine.py b/tests/unit/test_engine.py new file mode 100644 index 000000000000..9605017fbc91 --- /dev/null +++ b/tests/unit/test_engine.py @@ -0,0 +1,282 @@ +r""" +Unit tests for wic/engine.py. + +These cover the pure-logic parts that do not need a real image or +bitbake: + - verify_build_env: the BUILDDIR check + - canned-image discovery: build_canned_image_list, find_canned_image, + list_canned_images, list_canned_image_help, find_canned + - debugfs_version_check: version parsing, caching, and the + too-old-version rejection + - Disk.get_partitions: parsing of `parted -m ... print` output, and + the sector-size != 512 fstype guess + +`parted`/`debugfs` are never actually run: exec_cmd and shutil.which +are stubbed so the parsing and decision logic is what gets exercised. +Where wic does not yet behave correctly, the test is marked xfail with +a short reason. +""" + +import sys +from pathlib import Path + +import pytest + +_SRC = Path(__file__).resolve().parent.parent.parent / "src" +if str(_SRC) not in sys.path: + sys.path.insert(0, str(_SRC)) + +import wic.engine as engine +from wic import WicError +from wic.engine import ( + verify_build_env, build_canned_image_list, find_canned_image, + find_canned, debugfs_version_check, Disk, +) + + +# --------------------------------------------------------------------------- +# verify_build_env +# --------------------------------------------------------------------------- + +class TestVerifyBuildEnv: + + def test_raises_without_builddir(self, monkeypatch): + monkeypatch.delenv("BUILDDIR", raising=False) + with pytest.raises(WicError): + verify_build_env() + + def test_true_with_builddir(self, monkeypatch): + monkeypatch.setenv("BUILDDIR", "/some/build") + assert verify_build_env() is True + + +# --------------------------------------------------------------------------- +# Canned-image discovery +# --------------------------------------------------------------------------- + +class TestCannedImageDiscovery: + + @pytest.fixture + def canned(self, tmp_path, monkeypatch): + """A fake BBPATH layer with a files/wic canned dir.""" + layer = tmp_path / "layer" + wicdir = layer / "files" / "wic" + wicdir.mkdir(parents=True) + (wicdir / "directdisk.wks").write_text( + "# short-description: A direct disk image\n" + "# long-description: Creates a disk image\n" + "# with one rootfs partition.\n" + "part / --source rootfs --fstype ext4\n" + ) + (wicdir / "mkefidisk.wks.in").write_text("part /boot\n") + # editor leftovers that must be ignored + (wicdir / "directdisk.wks~").write_text("ignore me\n") + (wicdir / "directdisk.wks#").write_text("ignore me too\n") + monkeypatch.setattr(engine, "get_bitbake_var", + lambda *a, **k: str(layer)) + return wicdir + + def test_build_canned_image_list_finds_dir(self, canned): + dirs = build_canned_image_list("ignored") + assert str(canned) in dirs + + def test_find_canned_image_by_name(self, canned): + found = find_canned_image("ignored", "directdisk") + assert found is not None + assert found.endswith("directdisk.wks") + + def test_find_canned_image_dot_in_suffix(self, canned): + found = find_canned_image("ignored", "mkefidisk") + assert found is not None + assert found.endswith("mkefidisk.wks.in") + + def test_find_canned_image_missing(self, canned): + assert find_canned_image("ignored", "nonexistent") is None + + def test_find_canned_image_skips_editor_leftovers(self, canned): + # A name that would only match the ~/# leftovers must not resolve. + assert find_canned_image("ignored", "directdisk.wks~") is None + + def test_list_canned_images_prints_description(self, canned, capsys): + from wic.engine import list_canned_images + list_canned_images("ignored") + out = capsys.readouterr().out + assert "directdisk" in out + assert "A direct disk image" in out + + def test_list_canned_image_help_prints_long_desc(self, canned, capsys): + from wic.engine import list_canned_image_help + list_canned_image_help("ignored", str(canned / "directdisk.wks")) + out = capsys.readouterr().out + assert "Creates a disk image" in out + + def test_find_canned_existing_path_returned_directly(self, tmp_path): + f = tmp_path / "x.wks" + f.write_text("part /\n") + assert find_canned("ignored", str(f)) == str(f) + + def test_find_canned_by_name(self, canned): + assert find_canned("ignored", "directdisk.wks").endswith("directdisk.wks") + + +# --------------------------------------------------------------------------- +# debugfs_version_check +# --------------------------------------------------------------------------- + +class TestDebugfsVersionCheck: + + @pytest.fixture(autouse=True) + def reset_cache(self): + engine._DEBUGFS_VERSION = None + yield + engine._DEBUGFS_VERSION = None + + def test_accepts_new_enough_version(self, monkeypatch): + monkeypatch.setattr(engine, "exec_cmd", + lambda *a, **k: "debugfs 1.47.0 (5-Feb-2023)") + debugfs_version_check("debugfs") # must not raise + assert engine._DEBUGFS_VERSION == (1, 47, 0) + + def test_rejects_too_old_version(self, monkeypatch): + monkeypatch.setattr(engine, "exec_cmd", + lambda *a, **k: "debugfs 1.45.3 (14-Jul-2018)") + with pytest.raises(WicError): + debugfs_version_check("debugfs") + + def test_version_is_cached(self, monkeypatch): + calls = {"n": 0} + + def fake(*a, **k): + calls["n"] += 1 + return "debugfs 1.47.0" + + monkeypatch.setattr(engine, "exec_cmd", fake) + debugfs_version_check("debugfs") + debugfs_version_check("debugfs") + assert calls["n"] == 1 # second call uses the cached version + + def test_boundary_exact_minimum_accepted(self, monkeypatch): + monkeypatch.setattr(engine, "exec_cmd", + lambda *a, **k: "debugfs 1.46.5") + debugfs_version_check("debugfs") # exactly min_ver, must not raise + assert engine._DEBUGFS_VERSION == (1, 46, 5) + + +# --------------------------------------------------------------------------- +# Disk.get_partitions: parsing parted -m output +# --------------------------------------------------------------------------- + +_PARTED_MSDOS = ( + "BYT;\n" + "/dev/sda:1024B:file:512:512:msdos:Msftdata:;\n" + "1:512B:599B:88B:fat16:primary:;\n" + "2:600B:699B:100B:ext4:primary:;\n" +) + +# A 4K-sector disk where parted reports no fstype for partition 2. +_PARTED_4K_EMPTY_FSTYPE = ( + "BYT;\n" + "/dev/sda:8192B:file:4096:4096:msdos:Msftdata:;\n" + "1:4096B:8191B:4096B::primary:;\n" +) + + +def _make_disk(monkeypatch, parted_out, sector_size=512): + monkeypatch.setattr(engine, "exec_cmd", lambda *a, **k: parted_out) + monkeypatch.setattr(engine.shutil, "which", + lambda name, path=None: "/usr/bin/%s" % name) + monkeypatch.setenv("PATH", "/usr/bin") + return Disk("/tmp/fake.img", None, sector_size=sector_size) + + +class TestDiskGetPartitions: + + def test_parses_partition_table(self, monkeypatch): + disk = _make_disk(monkeypatch, _PARTED_MSDOS) + assert disk._ptable_format == "msdos" + assert disk._lsector_size == 512 + assert set(disk.partitions) == {"1", "2"} + + def test_partition_fields(self, monkeypatch): + disk = _make_disk(monkeypatch, _PARTED_MSDOS) + p1 = disk.partitions["1"] + assert p1.pnum == 1 + assert p1.start == 512 + assert p1.size == 88 + assert p1.fstype == "fat16" + + def test_unparseable_output_raises(self, monkeypatch): + with pytest.raises(WicError): + _make_disk(monkeypatch, "garbage with no BYT marker\n") + + def test_missing_parted_raises(self, monkeypatch): + monkeypatch.setattr(engine, "exec_cmd", lambda *a, **k: _PARTED_MSDOS) + monkeypatch.setattr(engine.shutil, "which", lambda name, path=None: None) + monkeypatch.setenv("PATH", "/usr/bin") + with pytest.raises(WicError): + Disk("/tmp/fake.img", None) + + +class TestDiskFstypeGuessAt4K: + """At sector_size != 512, parted can fail to report a vfat/dos fstype. + + wic works around this in _get_part_image by assuming 'fat' when + parted reports an empty fstype and the sector size is not 512. That + keeps FAT-on-4K working, but it means a genuinely typeless or + non-FAT partition on a 4K disk is silently treated as FAT instead of + being reported as unsupported. This pins the workaround and flags the + blind guess as a known limitation needing an upstream parted/blkid + fix. + """ + + def test_empty_fstype_at_4k_is_guessed_fat(self, monkeypatch): + disk = _make_disk(monkeypatch, _PARTED_4K_EMPTY_FSTYPE, sector_size=4096) + # parted reported nothing for partition 1's fstype + assert disk.partitions["1"].fstype == "" + # Stub sparse_copy to create a real (empty) staged file so the + # decision logic runs without copying, and so the finalizer can + # unlink it cleanly at teardown. + def fake_copy(src, dst, *a, **k): + open(dst, "w").close() + monkeypatch.setattr(engine, "sparse_copy", fake_copy) + # _get_part_image must accept it (guessed as fat) rather than + # raising "Not supported fstype". + try: + disk._get_part_image("1") + except WicError as exc: + pytest.fail("4K empty-fstype partition was rejected: %s" % exc) + + def test_empty_fstype_at_512_is_unsupported(self, monkeypatch): + # The same empty fstype at the normal 512 sector size is NOT + # guessed, so it is correctly reported as unsupported. + out = _PARTED_4K_EMPTY_FSTYPE.replace("4096:4096", "512:512") + disk = _make_disk(monkeypatch, out, sector_size=512) + with pytest.raises(WicError): + disk._get_part_image("1") + + +class TestDiskFinalizer: + """Disk.__del__ unlinks staged partition images. + + If __init__ raises before _partimages is set (e.g. parted missing), + the finalizer reaches __del__ -> __getattr__ and raises KeyError for + '_partimages' at GC time. __del__ should tolerate a partially built + instance. + """ + + def test_del_on_partial_instance_does_not_raise(self, monkeypatch): + obj = Disk.__new__(Disk) + try: + obj.__del__() + except KeyError: + pytest.xfail( + "Disk.__del__ raises KeyError('_partimages') on an instance " + "whose __init__ did not complete; it should guard the " + "attribute before iterating") + finally: + # Contain this test's own finalizer leak: give the partial + # object the attribute its __del__ iterates, so the GC-time + # __del__ cannot re-raise into a later test's teardown. This + # does not change the behaviour under test (already observed + # above); it only stops the warning from escaping. + obj._partimages = {} diff --git a/tests/unit/test_filemap.py b/tests/unit/test_filemap.py new file mode 100644 index 000000000000..05f83d09b5a3 --- /dev/null +++ b/tests/unit/test_filemap.py @@ -0,0 +1,2317 @@ +r""" +Unit tests for wic.filemap. + +Tests assert CORRECT behaviour, and xfail when the code does not yet +meet that behaviour. + +Known bugs targeted: + - _FilemapBase.__del__ raises AttributeError on a partially- + initialised instance because it reads self._f_image_needs_close + without a getattr() guard. + - FilemapSeek opens self._f_seek but never closes it; the base-class + __del__ only closes _f_image, leaking one fd per instance. +""" +import os +import sys +from pathlib import Path + +import pytest + +_WIC_SRC = Path(__file__).resolve().parent.parent.parent / "src" +if str(_WIC_SRC) not in sys.path: + sys.path.insert(0, str(_WIC_SRC)) + +from wic.filemap import ( + _FilemapBase, FilemapFiemap, FilemapSeek, FilemapNobmap, filemap, + sparse_copy, Error, ErrorNotSupp, get_block_size, + _lseek, _SEEK_DATA, _SEEK_HOLE, +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +@pytest.fixture +def sparse_image(tmp_path): + """A small sparse file suitable for FilemapBase initialisation.""" + p = tmp_path / "test.img" + with open(p, "wb") as f: + f.truncate(4096 * 16) # 64 KiB sparse + return str(p) + + +@pytest.fixture +def dense_image(tmp_path): + """A small dense (non-sparse) file.""" + p = tmp_path / "dense.img" + p.write_bytes(b"\xaa" * 4096 * 4) # 16 KiB dense + return str(p) + + +# --------------------------------------------------------------------------- +# get_block_size +# --------------------------------------------------------------------------- + +class TestGetBlockSize: + def test_returns_integer(self, sparse_image): + with open(sparse_image, "rb") as f: + bs = get_block_size(f) + assert isinstance(bs, int) + + def test_positive_block_size(self, sparse_image): + with open(sparse_image, "rb") as f: + bs = get_block_size(f) + assert bs > 0 + + def test_power_of_two(self, sparse_image): + """Block sizes are always powers of 2 on any sane filesystem.""" + with open(sparse_image, "rb") as f: + bs = get_block_size(f) + assert bs & (bs - 1) == 0 + + +# --------------------------------------------------------------------------- +# _FilemapBase.__del__ +# --------------------------------------------------------------------------- + +class TestFilemapBaseDel: + """__del__ must not raise AttributeError on a partially-initialised instance. + + __del__ reads self._f_image_needs_close directly; if __init__ + was never called (e.g. subclass overrides __new__, pickle/restore, + or _open_image_file raises before the flag is set) the attribute is + absent and __del__ raises AttributeError which the GC silently eats + but which pollutes stderr and can mask real errors. + + Correct fix: `if getattr(self, '_f_image_needs_close', False):` + """ + + def test_del_normal_instance_does_not_raise(self, sparse_image): + """Nominal path: fully initialised instance; __del__ must be silent.""" + obj = FilemapNobmap(sparse_image) + try: + obj.__del__() + except Exception as e: + pytest.fail("__del__ raised on a normal instance: %s" % e) + + def test_del_without_init_should_not_raise(self): + """__del__ on an instance that bypassed __init__ must not raise. + The correct behaviour is to silently do nothing (no file to close). + """ + obj = object.__new__(_FilemapBase) + # __init__ was never called; _f_image_needs_close does not exist + assert not hasattr(obj, "_f_image_needs_close") + try: + try: + obj.__del__() + except AttributeError: + pytest.xfail( + "_FilemapBase.__del__ raises AttributeError when " + "_f_image_needs_close is absent; fix: use getattr() guard") + except Exception as e: + pytest.fail("__del__ raised an unexpected exception: %s" % e) + finally: + # The bug has been observed above. Give the partially-built + # object the attribute its finalizer needs so the second, + # GC-time __del__ cannot re-raise into a later test's + # teardown. This only contains the test's own leak; it does + # not change or hide the behaviour under test. + obj._f_image_needs_close = False + + def test_nobmap_del_without_init_should_not_raise(self): + """Same requirement applies on a concrete subclass.""" + obj = object.__new__(FilemapNobmap) + try: + try: + obj.__del__() + except AttributeError: + pytest.xfail( + "FilemapNobmap.__del__ (inherited) raises AttributeError " + "when _f_image_needs_close is absent") + finally: + # Contain the test's own finalizer leak (see the sibling test). + obj._f_image_needs_close = False + + +# --------------------------------------------------------------------------- +# _FilemapBase — normal initialisation +# --------------------------------------------------------------------------- + +class TestFilemapBaseInit: + def test_image_size_populated(self, sparse_image): + obj = FilemapNobmap(sparse_image) + assert obj.image_size == 4096 * 16 + + def test_block_size_populated(self, sparse_image): + obj = FilemapNobmap(sparse_image) + assert obj.block_size > 0 + + def test_blocks_cnt_correct(self, sparse_image): + obj = FilemapNobmap(sparse_image) + expected = (obj.image_size + obj.block_size - 1) // obj.block_size + assert obj.blocks_cnt == expected + + def test_nonexistent_file_raises_error(self): + with pytest.raises(Error): + FilemapNobmap("/nonexistent/path/image.img") + + def test_accepts_file_object(self, sparse_image): + """_FilemapBase accepts an already-open file object.""" + with open(sparse_image, "rb") as f: + obj = FilemapNobmap(f) + assert obj.image_size == 4096 * 16 + + def test_image_path_set_from_string(self, sparse_image): + obj = FilemapNobmap(sparse_image) + assert obj._image_path == sparse_image + + def test_image_path_set_from_file_object(self, sparse_image): + with open(sparse_image, "rb") as f: + obj = FilemapNobmap(f) + assert obj._image_path == sparse_image + + +# --------------------------------------------------------------------------- +# FilemapNobmap — block mapping +# --------------------------------------------------------------------------- + +class TestFilemapNobmap: + def test_block_is_mapped_always_true(self, sparse_image): + obj = FilemapNobmap(sparse_image) + for block in (0, 1, obj.blocks_cnt - 1): + assert obj.block_is_mapped(block) + + def test_get_mapped_ranges_full_file(self, sparse_image): + obj = FilemapNobmap(sparse_image) + ranges = list(obj.get_mapped_ranges(0, obj.blocks_cnt)) + assert len(ranges) == 1 + assert ranges[0] == (0, obj.blocks_cnt - 1) + + def test_get_mapped_ranges_partial(self, sparse_image): + obj = FilemapNobmap(sparse_image) + ranges = list(obj.get_mapped_ranges(2, 4)) + assert len(ranges) == 1 + first, last = ranges[0] + assert first == 2 + assert last == 5 # start + count - 1 + + def test_get_mapped_ranges_single_block(self, sparse_image): + obj = FilemapNobmap(sparse_image) + ranges = list(obj.get_mapped_ranges(0, 1)) + assert ranges == [(0, 0)] + + +# --------------------------------------------------------------------------- +# FilemapSeek — fd leak +# --------------------------------------------------------------------------- + +class TestFilemapSeekFdLeak: + """FilemapSeek opens self._f_seek but has no __del__ to close it. + + The base class __del__ only closes _f_image, so _f_seek leaks one + file descriptor per FilemapSeek instance. + """ + + def _count_open_fds(self): + """Count open file descriptors in /proc/self/fd.""" + try: + return len(os.listdir("/proc/self/fd")) + except OSError: + return -1 + + def test_f_seek_exists_after_init(self, sparse_image): + """FilemapSeek must open _f_seek for SEEK_HOLE probing.""" + try: + obj = FilemapSeek(sparse_image) + except ErrorNotSupp: + pytest.skip("SEEK_HOLE not supported on this filesystem") + assert hasattr(obj, "_f_seek"), "_f_seek handle must exist after init" + assert not obj._f_seek.closed, "_f_seek must be open after init" + + def test_f_seek_closed_after_del(self, sparse_image): + """_f_seek must be closed when the object is deleted. + Currently FilemapSeek has no __del__; the handle leaks. + """ + try: + obj = FilemapSeek(sparse_image) + except ErrorNotSupp: + pytest.skip("SEEK_HOLE not supported on this filesystem") + + f_seek = obj._f_seek + del obj + # Correct: _f_seek must be closed after the object is destroyed. + if not f_seek.closed: + pytest.xfail( + "FilemapSeek._f_seek is not closed on __del__; " + "file descriptor leaks on every FilemapSeek instance") + assert f_seek.closed + + def test_repeated_init_del_does_not_exhaust_fds(self, sparse_image): + """Stress: repeated construction/destruction must not exhaust fds. + If _f_seek leaks, this test will eventually fail with OSError. + """ + try: + obj = FilemapSeek(sparse_image) + del obj + except ErrorNotSupp: + pytest.skip("SEEK_HOLE not supported on this filesystem") + + before = self._count_open_fds() + if before < 0: + pytest.skip("cannot count fds: /proc/self/fd not available") + + for _ in range(20): + try: + obj = FilemapSeek(sparse_image) + del obj + except (ErrorNotSupp, OSError): + break + + after = self._count_open_fds() + leaked = after - before + if leaked > 5: # small tolerance for GC timing + pytest.xfail( + "%d fds leaked after 20 FilemapSeek create/destroy cycles" + % leaked) + + +# --------------------------------------------------------------------------- +# sparse_copy +# --------------------------------------------------------------------------- + +class TestSparseCopy: + def test_copy_identity(self, dense_image, tmp_path): + dst = str(tmp_path / "copy.img") + sparse_copy(dense_image, dst) + assert Path(dst).read_bytes() == Path(dense_image).read_bytes() + + def test_copy_with_seek(self, dense_image, tmp_path): + """sparse_copy with seek=N writes data starting at byte offset N.""" + dst = str(tmp_path / "seeked.img") + # Pre-create dst with known content + Path(dst).write_bytes(b"\x00" * (4096 * 6)) + sparse_copy(dense_image, dst, seek=4096) + result = Path(dst).read_bytes() + src_bytes = Path(dense_image).read_bytes() + # Bytes at offset 4096 onward must match the source + assert result[4096:4096 + len(src_bytes)] == src_bytes + + def test_copy_with_skip(self, dense_image, tmp_path): + """sparse_copy with skip=N copies starting from byte N of the source.""" + dst = str(tmp_path / "skipped.img") + skip = 4096 + sparse_copy(dense_image, dst, skip=skip) + src_bytes = Path(dense_image).read_bytes() + dst_bytes = Path(dst).read_bytes() + assert dst_bytes[:len(src_bytes) - skip] == src_bytes[skip:] + + def test_copy_nonexistent_src_raises(self, tmp_path): + dst = str(tmp_path / "out.img") + with pytest.raises((Error, OSError, FileNotFoundError)): + sparse_copy("/nonexistent/src.img", dst) + + def test_copy_produces_correct_size(self, dense_image, tmp_path): + dst = str(tmp_path / "sized.img") + sparse_copy(dense_image, dst) + assert Path(dst).stat().st_size == Path(dense_image).stat().st_size + +# --------------------------------------------------------------------------- +# The stdio-buffer / raw-lseek separation bug +# +# Historical context: FilemapSeek originally shared a single file handle +# between the BufferedReader used by sparse_copy for data reads and the +# os.lseek() calls used for SEEK_DATA/SEEK_HOLE probing. When Python +# changed the default stdio buffer size this became visible: os.lseek() +# on the underlying fd moves the OS-level file position but does NOT +# invalidate the BufferedReader's internal buffer. Subsequent buffered +# reads return data from the buffer (old position) rather than from the +# new OS position, silently corrupting the copy. +# +# The fix: open a completely separate file handle (_f_seek) for lseek +# probing, leaving _f_image exclusively for buffered reads. +# +# These tests verify the fix holds and document the exact failure mode. +# --------------------------------------------------------------------------- + +import io + +# Use the actual Python stdio buffer size as a probe boundary. +_BUF = io.DEFAULT_BUFFER_SIZE # typically 8192 on Linux + + +class TestStdioLseekSeparation: + """Verify that _f_image and _f_seek are independent fds (not the same fd), + and that sparse_copy produces byte-exact output even when get_mapped_ranges + lseek calls cross the stdio buffer boundary many times. + + These tests would fail if the two-handle separation were reverted to a + single shared fd. + """ + + def _make_dense(self, tmp_path, pattern, size): + """Create a dense file filled with a repeating pattern byte.""" + p = tmp_path / ("dense_%02x_%d.img" % (pattern, size)) + p.write_bytes(bytes([pattern]) * size) + return str(p) + + def test_f_seek_and_f_image_are_different_fds(self, tmp_path, sparse_image): + """The two handles must have distinct file descriptor numbers. + + If they share a single fd, every os.lseek() call via _f_seek would + move the OS-level position of _f_image's underlying fd, invalidating + its buffer without the BufferedReader knowing. + """ + try: + obj = FilemapSeek(sparse_image) + except ErrorNotSupp: + pytest.skip("SEEK_HOLE not supported on this filesystem") + fd_image = obj._f_image.fileno() + fd_seek = obj._f_seek.fileno() + assert fd_image != fd_seek, ( + "REGRESSION: _f_image and _f_seek share the same file descriptor " + "(%d); os.lseek() on _f_seek will corrupt _f_image buffered reads") + + def test_raw_lseek_corruption_demonstration(self, tmp_path): + """Attempt to reproduce the buffered-read corruption that motivated the + _f_seek two-handle split in FilemapSeek. + + The corruption requires the BufferedReader to have PRE-FETCHED data + remaining in its buffer when a raw os.lseek() is applied to the + underlying fd. Whether that pre-fetching occurs depends on the OS + read-call behaviour: + + * tmpfs / in-memory filesystems: the kernel's read() returns a full + _BUF-sized chunk even when fewer bytes were in flight; the buffer + has stale data after a partial consume → corruption is visible. + * ext4 with fs-block-size == _BUF // 2: the kernel returns exactly the + requested number of bytes; the buffer is empty after the consume → + no stale data → corruption is NOT visible here. + + This test detects which situation is present: + • If no pre-fetching is observed, it exits silently. + • If pre-fetching is observed, it asserts that a raw lseek causes + the next read to return the stale buffer bytes (demonstrating the + corruption). + + The actual PROTECTION against this bug in wic is verified by the + other tests in this class (distinct-fd check; byte-exact parametrised + copies); those tests catch a regression regardless of filesystem. + """ + # Layout: \xaa × _BUF followed by \xbb × _BUF. + p = tmp_path / "corruption_demo.img" + p.write_bytes(b"\xaa" * _BUF + b"\xbb" * _BUF) + + f = open(str(p), "rb") + try: + # Consume _BUF // 2 bytes. If the OS pre-fetched a full _BUF + # chunk, the raw fd will be at _BUF; the buffer will still hold + # the remaining _BUF // 2 bytes of \xaa. + first = f.read(_BUF // 2) + assert first == b"\xaa" * (_BUF // 2) + raw_fd_pos = os.lseek(f.fileno(), 0, os.SEEK_CUR) + + if raw_fd_pos <= _BUF // 2: + # No pre-fetching detected: the OS returned exactly what + # was requested and the buffer is now empty. There is no + # stale data to corrupt; the test cannot demonstrate the bug + # on this filesystem. Exit silently — the structural and + # functional guards elsewhere still cover the protection. + return + + # Pre-fetching confirmed: raw_fd_pos > _BUF // 2 means the OS + # read more than we consumed. Stale data sits in the buffer. + # A raw lseek now moves the OS fd into the \xbb region without + # telling the BufferedReader. + os.lseek(f.fileno(), _BUF + _BUF // 2, os.SEEK_SET) + + # The BufferedReader still has the stale \xaa bytes in its + # internal buffer. The next read() SHOULD return \xbb (from + # the new OS position) but instead returns stale \xaa. + stale = f.read(_BUF // 2) + + assert stale == b"\xaa" * (_BUF // 2), ( + "Stale pre-fetched data not returned after raw lseek; got %r. " + "This may mean the BufferedReader now invalidates its cache on " + "raw-fd moves, which is safer. If so, the two-handle " + "separation is still correct defensive practice but the " + "corruption trigger condition is gone on this platform." + % stale[:8]) + # The data is WRONG: we seeked to the \xbb region but got \xaa. + assert stale != b"\xbb" * (_BUF // 2) + finally: + f.close() + + @pytest.mark.parametrize("skip", [ + 0, + 1, + _BUF - 1, + _BUF, + _BUF + 1, + 2 * _BUF - 1, + 2 * _BUF, + 2 * _BUF + 1, + 3 * _BUF, + ]) + def test_sparse_copy_byte_exact_at_buffer_boundaries(self, tmp_path, skip): + """sparse_copy must produce byte-exact output for any skip value, + including values that cross the DEFAULT_BUFFER_SIZE boundary. + + If the two-handle separation were broken (single shared fd), lseek + calls from get_mapped_ranges() would corrupt the BufferedReader buffer, + causing data at positions above DEFAULT_BUFFER_SIZE to be wrong. + """ + total_size = skip + 3 * _BUF + # Fill the source with a recognisable non-repeating pattern: + # byte value = byte_index mod 251 (a prime, so patterns don't repeat + # at any power-of-2 boundary) + src_bytes = bytes(i % 251 for i in range(total_size)) + src = tmp_path / ("src_%d.img" % skip) + src.write_bytes(src_bytes) + + dst = tmp_path / ("dst_%d.img" % skip) + sparse_copy(str(src), str(dst), skip=skip) + + expected = src_bytes[skip:] + got = dst.read_bytes() + # Trim to the expected length (dst may be slightly larger due to + # block-size rounding in sparse implementations) + got_trimmed = got[:len(expected)] + assert got_trimmed == expected, ( + "sparse_copy(skip=%d) produced wrong data at offset %d: " + "expected %r, got %r" % ( + skip, + next((i for i in range(len(expected)) if expected[i] != got_trimmed[i]), + len(expected)), + expected[:8], got_trimmed[:8])) + + def test_sparse_copy_with_seek_offset(self, tmp_path): + """sparse_copy(seek=N) writes data starting at byte N of the destination. + Probe with a seek value that crosses the buffer boundary. + """ + src_bytes = bytes(i % 251 for i in range(2 * _BUF)) + src = tmp_path / "seek_src.img" + src.write_bytes(src_bytes) + + dst = tmp_path / "seek_dst.img" + seek_offset = _BUF + 17 # intentionally not aligned + sparse_copy(str(src), str(dst), seek=seek_offset) + + got = dst.read_bytes() + # Bytes before seek_offset must be zero + assert got[:seek_offset] == b"\x00" * seek_offset, ( + "seek_offset preamble is not zero: %r" % got[:min(8, seek_offset)]) + # Bytes from seek_offset onward must match src + assert got[seek_offset:seek_offset + len(src_bytes)] == src_bytes + + def test_get_mapped_ranges_does_not_corrupt_subsequent_image_reads( + self, tmp_path): + """Interleaved get_mapped_ranges() and _f_image reads must not produce + stale data. This directly tests the two-handle fix: if _f_seek and + _f_image were the same fd, the lseek calls inside _get_ranges() would + invalidate _f_image's buffer. + """ + # Dense file: every block is mapped. + src_bytes = bytes(i % 251 for i in range(4 * _BUF)) + src = tmp_path / "probe_src.img" + src.write_bytes(src_bytes) + + fmap = filemap(str(src)) + try: + # Collect all ranges (drives multiple lseek probes via _f_seek). + ranges = list(fmap.get_mapped_ranges(0, fmap.blocks_cnt)) + assert len(ranges) > 0 + + # Now read through _f_image at a position well past the first buffer. + read_pos = 2 * _BUF + fmap._f_image.seek(read_pos, os.SEEK_SET) + chunk = fmap._f_image.read(_BUF) + + assert chunk == src_bytes[read_pos:read_pos + _BUF], ( + "Read after get_mapped_ranges() returned wrong data: " + "_f_seek lseek calls may have corrupted _f_image buffer. " + "Expected %r, got %r" % (src_bytes[read_pos:read_pos+8], chunk[:8])) + finally: + pass # fmap resources cleaned up by GC + + def test_f_image_uses_buffered_seek_not_raw_lseek(self, tmp_path): + """Verify that sparse_copy uses f.seek() (BufferedReader.seek) not + os.lseek() to position _f_image. BufferedReader.seek() correctly + invalidates the internal buffer; os.lseek() on the fd does not. + + We test this indirectly: after a BufferedReader.seek() the next read + MUST return data from the new position, not from the stale buffer. + """ + # Two distinct 512-byte sections within the first buffer window. + src_bytes = b"\xaa" * 512 + b"\xbb" * 512 + b"\xcc" * (_BUF - 1024) + src = tmp_path / "seek_correctness.img" + src.write_bytes(src_bytes) + + f = open(str(src), "rb") + try: + _ = f.read(512) # fills buffer with first _BUF bytes + f.seek(512, os.SEEK_SET) # BufferedReader seek to 0xBB region + chunk = f.read(512) + assert chunk == b"\xbb" * 512, ( + "BufferedReader.seek() followed by read() returned wrong data " + "— this is unexpected and may indicate a Python regression") + finally: + f.close() + +# =========================================================================== +# Helpers shared by the new test sections +# =========================================================================== + +_BS = 4096 # nominal block size used when creating test images + + +def _make_sparse_image(path, total_size, data_regions): + """Create a sparse file of *total_size* bytes with data written only at + the given *data_regions*, a sequence of (offset, bytes) pairs. + All un-written areas are holes (zeros on read, absent on FIEMAP/SEEK_DATA). + """ + with open(str(path), 'wb') as f: + f.truncate(total_size) + for offset, data in data_regions: + f.seek(offset) + f.write(data) + + +def _all_implementations(image_path): + """Return (name, instance) for every Filemap class that succeeds on + *image_path*, skipping those that raise ErrorNotSupp. + """ + result = [] + for cls in (FilemapFiemap, FilemapSeek, FilemapNobmap): + try: + result.append((cls.__name__, cls(image_path))) + except ErrorNotSupp: + pass + return result + + +# =========================================================================== +# TestFilemapFiemapDirect — FIEMAP ioctl path +# =========================================================================== + +class TestFilemapFiemapDirect: + """Tests targeting FilemapFiemap specifically.""" + + @pytest.fixture + def dense4(self, tmp_path): + p = tmp_path / "dense4.img" + p.write_bytes(bytes(i % 251 for i in range(_BS * 4))) + return str(p) + + @pytest.fixture + def sparse_hole_in_middle(self, tmp_path): + """16-block file; data only at blocks 0 and 14, rest holes.""" + p = tmp_path / "sparse_hole.img" + _make_sparse_image(p, _BS * 16, [ + (0, b"D" * _BS), + (14 * _BS, b"E" * _BS), + ]) + return str(p) + + # --- initialisation --- + + def test_init_from_path(self, dense4): + obj = FilemapFiemap(dense4) + assert obj.image_size == _BS * 4 + assert obj.blocks_cnt == 4 + assert obj.block_size == _BS + + def test_init_from_file_object(self, dense4): + with open(dense4, "rb") as fh: + obj = FilemapFiemap(fh) + assert obj.image_size == _BS * 4 + + def test_nonexistent_file_raises(self): + with pytest.raises((Error, ErrorNotSupp, OSError)): + FilemapFiemap("/nonexistent/completely/missing.img") + + def test_image_path_stored_from_string(self, dense4): + obj = FilemapFiemap(dense4) + assert obj._image_path == dense4 + + def test_image_path_stored_from_file_object(self, dense4): + with open(dense4, "rb") as fh: + obj = FilemapFiemap(fh) + assert obj._image_path == dense4 + + # --- block_is_mapped: dense file --- + + def test_all_blocks_mapped_in_dense_file(self, dense4): + obj = FilemapFiemap(dense4) + for blk in range(obj.blocks_cnt): + assert obj.block_is_mapped(blk), \ + "block %d should be mapped in dense file" % blk + + def test_block_0_is_mapped(self, dense4): + assert FilemapFiemap(dense4).block_is_mapped(0) + + def test_last_block_is_mapped(self, dense4): + obj = FilemapFiemap(dense4) + assert obj.block_is_mapped(obj.blocks_cnt - 1) + + # --- block_is_mapped: out-of-range blocks --- + + def test_negative_block_raises_error(self, dense4): + """Fiemap validates block number; negative must raise Error.""" + obj = FilemapFiemap(dense4) + with pytest.raises(Error): + obj.block_is_mapped(-1) + + def test_block_equals_blocks_cnt_raises_error(self, dense4): + obj = FilemapFiemap(dense4) + with pytest.raises(Error): + obj.block_is_mapped(obj.blocks_cnt) + + def test_large_block_raises_error(self, dense4): + obj = FilemapFiemap(dense4) + with pytest.raises(Error): + obj.block_is_mapped(9_999_999) + + # --- get_mapped_ranges: dense file --- + + def test_get_mapped_ranges_full_dense(self, dense4): + obj = FilemapFiemap(dense4) + ranges = list(obj.get_mapped_ranges(0, obj.blocks_cnt)) + assert len(ranges) == 1 + first, last = ranges[0] + assert first == 0 + assert last == obj.blocks_cnt - 1 + + def test_get_mapped_ranges_partial_start(self, dense4): + obj = FilemapFiemap(dense4) + ranges = list(obj.get_mapped_ranges(1, 2)) + assert len(ranges) >= 1 + assert ranges[0][0] >= 1 + + def test_get_mapped_ranges_single_block(self, dense4): + obj = FilemapFiemap(dense4) + ranges = list(obj.get_mapped_ranges(0, 1)) + assert len(ranges) == 1 + assert ranges[0] == (0, 0) + + def test_get_mapped_ranges_last_block_only(self, dense4): + obj = FilemapFiemap(dense4) + n = obj.blocks_cnt + ranges = list(obj.get_mapped_ranges(n - 1, 1)) + assert len(ranges) == 1 + assert ranges[0] == (n - 1, n - 1) + + # --- get_mapped_ranges: truly sparse file --- + + def test_sparse_file_skips_holes(self, sparse_hole_in_middle): + obj = FilemapFiemap(sparse_hole_in_middle) + ranges = list(obj.get_mapped_ranges(0, obj.blocks_cnt)) + mapped_blocks = set() + for first, last in ranges: + mapped_blocks.update(range(first, last + 1)) + # Blocks 0 and 14 have data; everything else is a hole. + assert 0 in mapped_blocks, "block 0 has data but is not in mapped ranges" + assert 14 in mapped_blocks, "block 14 has data but is not in mapped ranges" + # Holes (blocks 1-13, 15) must NOT appear. + for b in range(1, 14): + assert b not in mapped_blocks, \ + "hole block %d incorrectly reported as mapped by FIEMAP" % b + + def test_sparse_file_data_is_correct_after_copy(self, sparse_hole_in_middle, tmp_path): + dst = str(tmp_path / "copy.img") + sparse_copy(sparse_hole_in_middle, dst, api=FilemapFiemap) + src_bytes = Path(sparse_hole_in_middle).read_bytes() + dst_bytes = Path(dst).read_bytes() + assert dst_bytes == src_bytes, \ + "FilemapFiemap sparse_copy produced wrong content" + + +# =========================================================================== +# TestFilemapSeekDirect — SEEK_HOLE/SEEK_DATA path +# =========================================================================== + +class TestFilemapSeekDirect: + """Tests targeting FilemapSeek specifically.""" + + @pytest.fixture + def dense4(self, tmp_path): + p = tmp_path / "seek_dense4.img" + p.write_bytes(bytes(i % 251 for i in range(_BS * 4))) + return str(p) + + @pytest.fixture + def sparse_hole(self, tmp_path): + p = tmp_path / "seek_sparse.img" + _make_sparse_image(p, _BS * 16, [ + (0, b"D" * _BS), + (14 * _BS, b"E" * _BS), + ]) + return str(p) + + def _skip_if_not_supported(self, path): + try: + FilemapSeek(path) + except ErrorNotSupp as e: + pytest.skip("FilemapSeek not supported: %s" % e) + + # --- initialisation --- + + def test_init_from_path(self, dense4): + self._skip_if_not_supported(dense4) + obj = FilemapSeek(dense4) + assert obj.image_size == _BS * 4 + + def test_f_seek_is_a_separate_handle(self, dense4): + self._skip_if_not_supported(dense4) + obj = FilemapSeek(dense4) + assert hasattr(obj, "_f_seek") + assert not obj._f_seek.closed + assert obj._f_seek.fileno() != obj._f_image.fileno(), ( + "REGRESSION: _f_seek shares fd with _f_image") + + def test_f_seek_readable(self, dense4): + self._skip_if_not_supported(dense4) + obj = FilemapSeek(dense4) + # _f_seek must be an open readable file object + assert not obj._f_seek.closed + pos = os.lseek(obj._f_seek.fileno(), 0, os.SEEK_CUR) + assert pos >= 0 + + # --- block_is_mapped --- + + def test_all_dense_blocks_mapped(self, dense4): + self._skip_if_not_supported(dense4) + obj = FilemapSeek(dense4) + for blk in range(obj.blocks_cnt): + assert obj.block_is_mapped(blk) + + def test_negative_block_returns_false_not_error(self, dense4): + """FilemapSeek silently returns False for block -1 (unlike Fiemap which + raises Error). The test documents this inconsistency. + """ + self._skip_if_not_supported(dense4) + obj = FilemapSeek(dense4) + # Seek returns False; Fiemap raises Error; Nobmap returns True + result = obj.block_is_mapped(-1) + # Document that False is returned, not an error. + assert result is False, ( + "FilemapSeek.block_is_mapped(-1) returned %r; expected False " + "(inconsistent with FilemapFiemap which raises Error)" % result) + + def test_huge_block_returns_false(self, dense4): + self._skip_if_not_supported(dense4) + obj = FilemapSeek(dense4) + assert obj.block_is_mapped(9_999_999) is False + + # --- get_mapped_ranges --- + + def test_dense_file_yields_full_range(self, dense4): + self._skip_if_not_supported(dense4) + obj = FilemapSeek(dense4) + ranges = list(obj.get_mapped_ranges(0, obj.blocks_cnt)) + assert len(ranges) >= 1 + first, last = ranges[0][0], ranges[-1][1] + assert first == 0 + assert last == obj.blocks_cnt - 1 + + def test_sparse_file_skips_holes(self, sparse_hole): + self._skip_if_not_supported(sparse_hole) + obj = FilemapSeek(sparse_hole) + ranges = list(obj.get_mapped_ranges(0, obj.blocks_cnt)) + mapped_blocks = set() + for first, last in ranges: + mapped_blocks.update(range(first, last + 1)) + assert 0 in mapped_blocks + assert 14 in mapped_blocks + for b in range(1, 14): + assert b not in mapped_blocks, \ + "hole block %d incorrectly mapped by FilemapSeek" % b + + def test_get_mapped_ranges_single_block(self, dense4): + self._skip_if_not_supported(dense4) + obj = FilemapSeek(dense4) + ranges = list(obj.get_mapped_ranges(0, 1)) + assert ranges == [(0, 0)] + + # --- _f_seek and _f_image remain independent across multiple probe calls --- + + def test_repeated_get_mapped_ranges_does_not_disturb_f_image(self, dense4): + """Multiple get_mapped_ranges calls must not corrupt _f_image reads. + This is the core correctness guarantee the two-handle fix provides. + """ + self._skip_if_not_supported(dense4) + src_bytes = Path(dense4).read_bytes() + obj = FilemapSeek(dense4) + + # Drive many range probes. + for _ in range(5): + list(obj.get_mapped_ranges(0, obj.blocks_cnt)) + + # _f_image must still read correctly at any offset. + for offset in (0, _BS, 2 * _BS, _BS * 4 - 1): + obj._f_image.seek(offset, os.SEEK_SET) + chunk = obj._f_image.read(1) + expected = src_bytes[offset:offset + 1] + assert chunk == expected, ( + "At offset %d: expected %r got %r — _f_seek lseek calls may " + "have corrupted _f_image buffer (two-handle separation broken)" + % (offset, expected, chunk)) + + +# =========================================================================== +# TestFilemapDispatch — filemap() factory function +# =========================================================================== + +class TestFilemapDispatch: + """Tests for the filemap() factory that selects the best implementation.""" + + @pytest.fixture + def dense4(self, tmp_path): + p = tmp_path / "dispatch_dense.img" + p.write_bytes(b"\xcc" * _BS * 4) + return str(p) + + def test_returns_a_filemap_object(self, dense4): + obj = filemap(dense4) + assert isinstance(obj, (_FilemapBase,)) + + def test_fiemap_preferred_on_this_filesystem(self, dense4): + """On ext4 (this test environment) FilemapFiemap should be chosen.""" + obj = filemap(dense4) + assert isinstance(obj, FilemapFiemap), ( + "Expected FilemapFiemap on ext4; got %s. The priority order " + "may have changed." % type(obj).__name__) + + def test_accepts_file_object(self, dense4): + with open(dense4, "rb") as fh: + obj = filemap(fh) + assert obj.image_size == _BS * 4 + + def test_get_mapped_ranges_consistent_with_direct(self, dense4): + obj_auto = filemap(dense4) + ranges_auto = list(obj_auto.get_mapped_ranges(0, obj_auto.blocks_cnt)) + obj_direct = FilemapFiemap(dense4) + ranges_direct = list(obj_direct.get_mapped_ranges(0, obj_direct.blocks_cnt)) + assert ranges_auto == ranges_direct + + +# =========================================================================== +# TestAllThreePathsAgree — key invariant: all implementations produce the +# same sparse_copy output for the same input. +# =========================================================================== + +class TestAllThreePathsAgree: + """The three Filemap implementations are interchangeable for sparse_copy. + Whatever the internal block-mapping strategy, the output bytes must be + identical. If they differ, at least one implementation has a bug. + """ + + def _available_classes(self, path): + classes = [] + for cls in (FilemapFiemap, FilemapSeek, FilemapNobmap): + try: + cls(path) + classes.append(cls) + except ErrorNotSupp: + pass + return classes + + def _copy_via(self, cls, src, dst, **kwargs): + if dst.exists(): + dst.unlink() + sparse_copy(str(src), str(dst), api=cls, **kwargs) + return dst.read_bytes() + + # --- dense files --- + + @pytest.mark.parametrize("size_blocks", [1, 2, 4, 7, 8, 9, 16]) + def test_dense_file_all_agree(self, tmp_path, size_blocks): + src = tmp_path / "dense_agree.img" + src.write_bytes(bytes(i % 251 for i in range(_BS * size_blocks))) + classes = self._available_classes(str(src)) + if len(classes) < 2: + pytest.skip("fewer than 2 implementations available") + + results = {} + for cls in classes: + dst = tmp_path / ("out_%s.img" % cls.__name__) + results[cls.__name__] = self._copy_via(cls, src, dst) + + reference_name, reference = next(iter(results.items())) + for name, data in results.items(): + assert data == reference, ( + "%s and %s produced different output for a %d-block dense file" + % (reference_name, name, size_blocks)) + + # --- non-block-aligned file sizes --- + + @pytest.mark.parametrize("extra_bytes", [1, 17, _BS - 1, _BS + 1]) + def test_non_block_aligned_size_all_agree(self, tmp_path, extra_bytes): + size = _BS * 3 + extra_bytes + src = tmp_path / "unaligned.img" + src.write_bytes(bytes(i % 251 for i in range(size))) + classes = self._available_classes(str(src)) + if len(classes) < 2: + pytest.skip("fewer than 2 implementations available") + + results = {} + for cls in classes: + dst = tmp_path / ("out_%s.img" % cls.__name__) + results[cls.__name__] = self._copy_via(cls, src, dst) + + reference = next(iter(results.values())) + for name, data in results.items(): + if len(data) >= len(src.read_bytes()): + # Trim to source length for comparison (block rounding differs). + assert data[:len(reference)] == reference[:len(reference)], \ + "%s disagrees on non-aligned size %d" % (name, size) + + # --- sparse files --- + + def test_sparse_file_copy_content_agrees(self, tmp_path): + src = tmp_path / "sparse_agree.img" + _make_sparse_image(src, _BS * 16, [ + (0, b"D" * _BS), + (14 * _BS, b"E" * _BS), + ]) + expected = src.read_bytes() + classes = self._available_classes(str(src)) + + for cls in classes: + dst = tmp_path / ("s_out_%s.img" % cls.__name__) + got = self._copy_via(cls, src, dst) + assert got == expected, \ + "%s copy of sparse file produced wrong content" % cls.__name__ + + # --- with skip --- + + @pytest.mark.parametrize("skip", [0, 1, _BS - 1, _BS, _BS + 1, 2 * _BS]) + def test_skip_all_agree(self, tmp_path, skip): + src = tmp_path / "skip_src.img" + src.write_bytes(bytes(i % 251 for i in range(_BS * 4))) + if skip >= len(src.read_bytes()): + pytest.skip("skip >= file_size") + classes = self._available_classes(str(src)) + if len(classes) < 2: + pytest.skip("fewer than 2 implementations available") + + results = {} + for cls in classes: + dst = tmp_path / ("skip_%s_%d.img" % (cls.__name__, skip)) + results[cls.__name__] = self._copy_via(cls, src, dst, skip=skip) + + reference = next(iter(results.values())) + for name, data in results.items(): + assert data[:len(reference)] == reference[:len(reference)], \ + "%s disagrees with reference for skip=%d" % (name, skip) + + # --- 1-byte file --- + + def test_one_byte_file_all_agree(self, tmp_path): + src = tmp_path / "one_byte.img" + src.write_bytes(b"\x42") + classes = self._available_classes(str(src)) + results = {} + for cls in classes: + dst = tmp_path / ("1b_%s.img" % cls.__name__) + results[cls.__name__] = self._copy_via(cls, src, dst) + for name, data in results.items(): + assert data[:1] == b"\x42", \ + "%s did not copy single byte correctly" % name + + +# =========================================================================== +# TestTrulySparseFiles — files with actual filesystem holes +# =========================================================================== + +class TestTrulySparseFiles: + """Create files with real holes and verify each implementation correctly + maps (or ignores) them. The sparse_copy output must always equal the + source bytes because reading a hole returns zeros. + """ + + def _make_src(self, tmp_path, name, total_blocks, data_regions_blocks): + p = tmp_path / name + data_regions = [(blk * _BS, b"X" * _BS) for blk in data_regions_blocks] + _make_sparse_image(p, total_blocks * _BS, data_regions) + return p + + def test_fiemap_maps_only_data_blocks(self, tmp_path): + src = self._make_src(tmp_path, "fiemap_sparse.img", 16, [0, 7, 15]) + obj = FilemapFiemap(str(src)) + ranges = list(obj.get_mapped_ranges(0, obj.blocks_cnt)) + mapped = set() + for first, last in ranges: + mapped.update(range(first, last + 1)) + assert 0 in mapped and 7 in mapped and 15 in mapped + for b in list(range(1, 7)) + list(range(8, 15)): + assert b not in mapped, "hole block %d wrongly reported as mapped" % b + + def test_seek_maps_only_data_blocks(self, tmp_path): + src = self._make_src(tmp_path, "seek_sparse.img", 16, [0, 7, 15]) + try: + obj = FilemapSeek(str(src)) + except ErrorNotSupp: + pytest.skip("FilemapSeek not supported") + ranges = list(obj.get_mapped_ranges(0, obj.blocks_cnt)) + mapped = set() + for first, last in ranges: + mapped.update(range(first, last + 1)) + assert 0 in mapped and 7 in mapped and 15 in mapped + for b in list(range(1, 7)) + list(range(8, 15)): + assert b not in mapped, "hole block %d wrongly mapped by Seek" % b + + def test_nobmap_reports_all_blocks_as_mapped(self, tmp_path): + src = self._make_src(tmp_path, "nobmap_sparse.img", 8, [2, 5]) + obj = FilemapNobmap(str(src)) + for b in range(obj.blocks_cnt): + assert obj.block_is_mapped(b), "Nobmap must map every block" + + def test_sparse_copy_reproduces_content_fiemap(self, tmp_path): + src = self._make_src(tmp_path, "fc_sparse.img", 16, [0, 7, 15]) + dst = tmp_path / "fc_dst.img" + sparse_copy(str(src), str(dst), api=FilemapFiemap) + assert dst.read_bytes() == src.read_bytes() + + def test_sparse_copy_reproduces_content_nobmap(self, tmp_path): + src = self._make_src(tmp_path, "nc_sparse.img", 16, [0, 7, 15]) + dst = tmp_path / "nc_dst.img" + sparse_copy(str(src), str(dst), api=FilemapNobmap) + assert dst.read_bytes() == src.read_bytes() + + def test_sparse_copy_reproduces_content_seek(self, tmp_path): + src = self._make_src(tmp_path, "sc_sparse.img", 16, [0, 7, 15]) + dst = tmp_path / "sc_dst.img" + try: + sparse_copy(str(src), str(dst), api=FilemapSeek) + except ErrorNotSupp: + pytest.skip("FilemapSeek not supported") + assert dst.read_bytes() == src.read_bytes() + + def test_data_only_at_last_block(self, tmp_path): + """A file that is all holes except the very last block.""" + src = self._make_src(tmp_path, "last_block.img", 8, [7]) + dst = tmp_path / "last_block_dst.img" + sparse_copy(str(src), str(dst), api=FilemapFiemap) + expected = src.read_bytes() + got = dst.read_bytes() + assert got[-_BS:] == expected[-_BS:], \ + "Last block data corrupted: expected %r, got %r" % ( + expected[-_BS:-_BS+8], got[-_BS:-_BS+8]) + + def test_data_only_at_first_block(self, tmp_path): + """A file that is all holes except the very first block.""" + src = self._make_src(tmp_path, "first_block.img", 8, [0]) + dst = tmp_path / "first_block_dst.img" + sparse_copy(str(src), str(dst), api=FilemapFiemap) + expected = src.read_bytes() + got = dst.read_bytes() + assert got[:_BS] == expected[:_BS], \ + "First block data corrupted" + + def test_adjacent_data_blocks_are_one_range(self, tmp_path): + """FIEMAP should merge adjacent mapped extents into one range.""" + # Blocks 3, 4, 5 are data. + src = self._make_src(tmp_path, "adjacent.img", 10, [3, 4, 5]) + obj = FilemapFiemap(str(src)) + ranges = list(obj.get_mapped_ranges(0, obj.blocks_cnt)) + # All three adjacent blocks must appear; they may be one merged range. + mapped = set() + for first, last in ranges: + mapped.update(range(first, last + 1)) + assert {3, 4, 5}.issubset(mapped) + + def test_alternating_data_hole_data(self, tmp_path): + """Every other block is data; holes must be correctly skipped.""" + data_blocks = list(range(0, 16, 2)) # 0, 2, 4, 6, 8, 10, 12, 14 + hole_blocks = list(range(1, 16, 2)) # 1, 3, 5, 7, 9, 11, 13, 15 + src = self._make_src(tmp_path, "alternating.img", 16, data_blocks) + obj = FilemapFiemap(str(src)) + ranges = list(obj.get_mapped_ranges(0, obj.blocks_cnt)) + mapped = set() + for first, last in ranges: + mapped.update(range(first, last + 1)) + for b in data_blocks: + assert b in mapped, "data block %d not in FIEMAP ranges" % b + for b in hole_blocks: + assert b not in mapped, \ + "hole block %d incorrectly in FIEMAP ranges" % b + + +# =========================================================================== +# TestEmptyFileHandling — all three implementations on a 0-byte file +# =========================================================================== + +class TestEmptyFileHandling: + """A 0-byte file is a valid edge case. FilemapFiemap and FilemapNobmap + both mishandle this input; FilemapSeek handles it correctly. + """ + + @pytest.fixture + def empty_img(self, tmp_path): + p = tmp_path / "empty.img" + p.write_bytes(b"") + return str(p) + + def test_fiemap_init_succeeds_on_empty(self, empty_img): + obj = FilemapFiemap(empty_img) + assert obj.image_size == 0 + assert obj.blocks_cnt == 0 + + def test_fiemap_get_mapped_ranges_empty_file(self, empty_img): + """FilemapFiemap.get_mapped_ranges(0, 0) crashes with RuntimeError + because next(iterator) is called unconditionally. + """ + obj = FilemapFiemap(empty_img) + try: + result = list(obj.get_mapped_ranges(0, obj.blocks_cnt)) + # If we get here, the bug is fixed. Correct behaviour is []. + assert result == [], \ + "Expected [] for empty file, got %r" % result + except RuntimeError: + pytest.xfail( + "FilemapFiemap.get_mapped_ranges raises RuntimeError for " + "empty file because next(iterator) is called unconditionally") + + def test_nobmap_init_succeeds_on_empty(self, empty_img): + obj = FilemapNobmap(empty_img) + assert obj.image_size == 0 + assert obj.blocks_cnt == 0 + + def test_nobmap_get_mapped_ranges_empty_file(self, empty_img): + """FilemapNobmap.get_mapped_ranges(0, 0) yields (0, -1), an + invalid range where last < first. + """ + obj = FilemapNobmap(empty_img) + ranges = list(obj.get_mapped_ranges(0, obj.blocks_cnt)) + if ranges == [(0, -1)]: + pytest.xfail( + "FilemapNobmap.get_mapped_ranges(0, 0) yields (0, -1); " + "last < first is an invalid range — should yield nothing") + # If fixed, the only acceptable answer is []. + assert ranges == [], \ + "Expected [] for empty file, got %r" % ranges + + def test_seek_get_mapped_ranges_empty_file_correct(self, empty_img): + """FilemapSeek correctly yields nothing for an empty file.""" + try: + obj = FilemapSeek(empty_img) + except ErrorNotSupp: + pytest.skip("FilemapSeek not supported") + ranges = list(obj.get_mapped_ranges(0, obj.blocks_cnt)) + assert ranges == [], \ + "FilemapSeek should yield [] for empty file, got %r" % ranges + + def test_sparse_copy_empty_source_produces_empty_dest(self, empty_img, tmp_path): + """sparse_copy of an empty file should produce a 0-byte destination, + not crash. Currently crashes in FilemapFiemap when FIEMAP is the + selected backend. + """ + dst = str(tmp_path / "empty_out.img") + try: + sparse_copy(empty_img, dst) + assert Path(dst).stat().st_size == 0, \ + "Expected 0-byte destination, got %d bytes" % Path(dst).stat().st_size + except RuntimeError: + pytest.xfail( + "sparse_copy of empty file crashes with RuntimeError via " + "FilemapFiemap.get_mapped_ranges") + + +# =========================================================================== +# TestSparseCopyEdgeCases — extreme skip / seek / length combinations +# =========================================================================== + +class TestSparseCopyEdgeCases: + """Boundary tests for sparse_copy's skip, seek, length.""" + + @pytest.fixture + def src8k(self, tmp_path): + """8 KiB source with a recognisable non-repeating pattern.""" + p = tmp_path / "adv_src.img" + p.write_bytes(bytes(i % 251 for i in range(_BS * 2))) + return p + + @pytest.fixture + def src1block(self, tmp_path): + p = tmp_path / "adv_1blk.img" + p.write_bytes(b"\xbe" * _BS) + return p + + # --- skip boundary cases --- + + def test_skip_zero(self, src8k, tmp_path): + dst = tmp_path / "s0.img" + sparse_copy(str(src8k), str(dst), skip=0) + assert dst.read_bytes()[:_BS * 2] == src8k.read_bytes() + + def test_skip_one(self, src8k, tmp_path): + dst = tmp_path / "s1.img" + sparse_copy(str(src8k), str(dst), skip=1) + expected = src8k.read_bytes()[1:] + got = dst.read_bytes() + assert got[:len(expected)] == expected + + def test_skip_block_minus_one(self, src8k, tmp_path): + """Skip just inside the first block boundary.""" + skip = _BS - 1 + dst = tmp_path / ("s%d.img" % skip) + sparse_copy(str(src8k), str(dst), skip=skip) + expected = src8k.read_bytes()[skip:] + got = dst.read_bytes() + assert got[:len(expected)] == expected + + def test_skip_exactly_one_block(self, src8k, tmp_path): + """Skip exactly the first block.""" + skip = _BS + dst = tmp_path / ("s%d.img" % skip) + sparse_copy(str(src8k), str(dst), skip=skip) + expected = src8k.read_bytes()[skip:] + got = dst.read_bytes() + assert got[:len(expected)] == expected + + def test_skip_block_plus_one(self, src8k, tmp_path): + """Skip just past the first block boundary.""" + skip = _BS + 1 + dst = tmp_path / ("s%d.img" % skip) + sparse_copy(str(src8k), str(dst), skip=skip) + expected = src8k.read_bytes()[skip:] + got = dst.read_bytes() + assert got[:len(expected)] == expected + + def test_skip_file_size_minus_one(self, src8k, tmp_path): + """Skip all but the very last byte.""" + src_bytes = src8k.read_bytes() + skip = len(src_bytes) - 1 + dst = tmp_path / "skip_last.img" + sparse_copy(str(src8k), str(dst), skip=skip) + got = dst.read_bytes() + assert got[0:1] == src_bytes[-1:], \ + "Last byte not preserved: expected %r got %r" % (src_bytes[-1:], got[:1]) + + def test_skip_equals_file_size_produces_empty(self, src8k, tmp_path): + """skip == file_size: nothing to copy → 0-byte destination.""" + src_size = src8k.stat().st_size + dst = tmp_path / "skip_eq.img" + sparse_copy(str(src8k), str(dst), skip=src_size) + assert dst.stat().st_size == 0, \ + "skip==file_size should produce a 0-byte destination" + + def test_skip_beyond_file_size(self, src8k, tmp_path): + """skip > file_size triggers negative truncate → OSError.""" + src_size = src8k.stat().st_size + dst = tmp_path / "skip_over.img" + try: + sparse_copy(str(src8k), str(dst), skip=src_size + 1) + # If we get here, the bug is fixed; destination must be sane. + assert dst.stat().st_size >= 0 + except OSError: + pytest.xfail( + "sparse_copy(skip > file_size) computes negative dst_size " + "and crashes with OSError from truncate()") + + def test_skip_large_value(self, src8k, tmp_path): + """skip far beyond file_size — same negative-truncate bug.""" + dst = tmp_path / "skip_huge.img" + try: + sparse_copy(str(src8k), str(dst), skip=999_999_999) + assert dst.stat().st_size >= 0 + except OSError: + pytest.xfail( + "sparse_copy(skip=999_999_999) crashes with OSError") + + # --- seek boundary cases --- + + def test_seek_zero(self, src1block, tmp_path): + """seek=0 is the default; output starts at byte 0.""" + dst = tmp_path / "seek0.img" + sparse_copy(str(src1block), str(dst), seek=0) + got = dst.read_bytes() + assert got[:_BS] == b"\xbe" * _BS + + def test_seek_one_block(self, src1block, tmp_path): + """seek=_BS: destination has _BS zero bytes then source data.""" + dst = tmp_path / "seek_blk.img" + sparse_copy(str(src1block), str(dst), seek=_BS) + got = dst.read_bytes() + assert got[:_BS] == b"\x00" * _BS, "preamble must be zeros" + assert got[_BS:2 * _BS] == b"\xbe" * _BS, "source data must follow" + + def test_seek_non_aligned(self, src1block, tmp_path): + seek = 17 + dst = tmp_path / "seek17.img" + sparse_copy(str(src1block), str(dst), seek=seek) + got = dst.read_bytes() + assert got[:seek] == b"\x00" * seek + assert got[seek:seek + _BS] == b"\xbe" * _BS + + # --- length boundary cases --- + + def test_length_zero_copies_everything(self, src8k, tmp_path): + """length=0 means 'no limit' — full copy.""" + dst = tmp_path / "len0.img" + sparse_copy(str(src8k), str(dst), length=0) + got = dst.read_bytes() + assert got[:_BS * 2] == src8k.read_bytes() + + def test_length_one(self, src8k, tmp_path): + """length=1 copies only the first byte.""" + dst = tmp_path / "len1.img" + sparse_copy(str(src8k), str(dst), length=1) + got = dst.read_bytes() + expected_first = src8k.read_bytes()[0:1] + assert got[:1] == expected_first, \ + "length=1 must copy exactly one byte; got %r" % got[:4] + + def test_length_full_file(self, src8k, tmp_path): + """length == file_size: full copy.""" + src_bytes = src8k.read_bytes() + dst = tmp_path / "len_full.img" + sparse_copy(str(src8k), str(dst), length=len(src_bytes)) + got = dst.read_bytes() + assert got[:len(src_bytes)] == src_bytes + + def test_length_larger_than_file(self, src8k, tmp_path): + """length > file_size: copies at most file_size bytes without crash.""" + src_bytes = src8k.read_bytes() + dst = tmp_path / "len_big.img" + # Should not crash; may produce a destination of file_size bytes. + sparse_copy(str(src8k), str(dst), length=len(src_bytes) * 10) + got = dst.read_bytes() + assert got[:len(src_bytes)] == src_bytes, \ + "Data beyond file_size limit produced wrong content" + + # --- combined skip + seek --- + + def test_skip_and_seek_combined(self, src8k, tmp_path): + src_bytes = src8k.read_bytes() + skip = _BS + seek = _BS // 2 + dst = tmp_path / "skip_seek.img" + sparse_copy(str(src8k), str(dst), skip=skip, seek=seek) + got = dst.read_bytes() + assert got[:seek] == b"\x00" * seek + assert got[seek:seek + _BS] == src_bytes[skip:skip + _BS] + + # --- 1-byte source file --- + + def test_one_byte_source(self, tmp_path): + src = tmp_path / "one.img" + src.write_bytes(b"\x99") + dst = tmp_path / "one_dst.img" + sparse_copy(str(src), str(dst)) + assert dst.read_bytes()[:1] == b"\x99" + + # --- non-block-aligned source sizes --- + + @pytest.mark.parametrize("size", [1, 7, _BS - 1, _BS + 1, _BS * 3 + 17]) + def test_non_aligned_sizes_copy_correctly(self, tmp_path, size): + src = tmp_path / ("nonalign_%d.img" % size) + src.write_bytes(bytes(i % 251 for i in range(size))) + dst = tmp_path / ("nonalign_%d_dst.img" % size) + sparse_copy(str(src), str(dst)) + got = dst.read_bytes() + expected = src.read_bytes() + assert got[:len(expected)] == expected, \ + "Non-aligned size %d: content mismatch at byte %d" % ( + size, + next((i for i in range(len(expected)) if expected[i] != got[i]), + len(expected))) + + +# =========================================================================== +# TestBlockIsMappedOutOfRange — inconsistent out-of-range handling +# =========================================================================== + +class TestBlockIsMappedOutOfRange: + """block_is_mapped with out-of-range, negative, and boundary blocks. + + the three implementations handle these cases differently: + FilemapFiemap raises Error + FilemapSeek returns False + FilemapNobmap returns True (most wrong: reports absent blocks as mapped) + """ + + @pytest.fixture + def dense4(self, tmp_path): + p = tmp_path / "bim_dense.img" + p.write_bytes(b"\xab" * _BS * 4) + return str(p) + + def _make_all(self, path): + impls = {} + for cls in (FilemapFiemap, FilemapSeek, FilemapNobmap): + try: + impls[cls.__name__] = cls(path) + except ErrorNotSupp: + pass + return impls + + def test_fiemap_negative_block_raises_error(self, dense4): + obj = FilemapFiemap(dense4) + with pytest.raises(Error): + obj.block_is_mapped(-1) + + def test_seek_negative_block_returns_false(self, dense4): + try: + obj = FilemapSeek(dense4) + except ErrorNotSupp: + pytest.skip() + result = obj.block_is_mapped(-1) + assert result is False, "FilemapSeek should return False for block -1" + + def test_nobmap_negative_block_returns_true(self, dense4): + """FilemapNobmap returns True for block -1 (nonsensical).""" + obj = FilemapNobmap(dense4) + result = obj.block_is_mapped(-1) + if result is True: + pytest.xfail( + "FilemapNobmap.block_is_mapped(-1) returns True; a " + "negative block number is out-of-range and should not be " + "considered mapped") + assert result is False # correct behavior if bug is fixed + + def test_fiemap_large_block_raises_error(self, dense4): + obj = FilemapFiemap(dense4) + with pytest.raises(Error): + obj.block_is_mapped(9_999_999) + + def test_seek_large_block_returns_false(self, dense4): + try: + obj = FilemapSeek(dense4) + except ErrorNotSupp: + pytest.skip() + assert obj.block_is_mapped(9_999_999) is False + + def test_nobmap_large_block_returns_true(self, dense4): + """FilemapNobmap returns True for a block way past blocks_cnt.""" + obj = FilemapNobmap(dense4) + result = obj.block_is_mapped(9_999_999) + if result is True: + pytest.xfail( + "FilemapNobmap.block_is_mapped(9_999_999) returns True; " + "that block doesn't exist in the file") + assert result is False + + def test_block_equals_blocks_cnt_fiemap_raises(self, dense4): + """blocks_cnt is one-past-the-end; must be out-of-range.""" + obj = FilemapFiemap(dense4) + with pytest.raises(Error): + obj.block_is_mapped(obj.blocks_cnt) + + def test_block_zero_mapped_in_all_implementations(self, dense4): + for name, obj in self._make_all(dense4).items(): + assert obj.block_is_mapped(0) is True, \ + "%s: block 0 must be mapped in a dense file" % name + + def test_last_block_mapped_in_all_implementations(self, dense4): + for name, obj in self._make_all(dense4).items(): + assert obj.block_is_mapped(obj.blocks_cnt - 1) is True, \ + "%s: last block must be mapped in a dense file" % name + + def test_all_in_range_blocks_true_for_dense(self, dense4): + for name, obj in self._make_all(dense4).items(): + for b in range(obj.blocks_cnt): + assert obj.block_is_mapped(b), \ + "%s: block %d should be mapped in dense file" % (name, b) + + +# =========================================================================== +# TestLseekHelper — _lseek() internals +# =========================================================================== + +class TestLseekHelper: + """Test _lseek() behaviour: ENXIO → -1, normal seek → position, EINVAL → ErrorNotSupp.""" + + @pytest.fixture + def sparse_file(self, tmp_path): + """A sparse file: block 0 = data, rest = holes.""" + p = tmp_path / "lseek_sparse.img" + _make_sparse_image(p, _BS * 8, [(0, b"A" * _BS)]) + return p + + def test_seek_data_at_start_returns_zero(self, sparse_file): + with open(str(sparse_file), "rb") as f: + pos = _lseek(f, 0, _SEEK_DATA) + assert pos == 0, "SEEK_DATA at 0 should find data at block 0" + + def test_seek_hole_after_data_returns_positive(self, sparse_file): + with open(str(sparse_file), "rb") as f: + hole_pos = _lseek(f, 0, _SEEK_HOLE) + assert hole_pos > 0, \ + "SEEK_HOLE at 0 should skip block 0 (data) and return hole offset" + + def test_seek_data_in_hole_region_returns_minus_one(self, sparse_file): + """SEEK_DATA starting in a hole returns -1 when no data follows.""" + with open(str(sparse_file), "rb") as f: + # Block 1 onwards is a hole; no data follows. + result = _lseek(f, _BS, _SEEK_DATA) + assert result == -1, \ + "Expected -1 (ENXIO) seeking data in hole region, got %d" % result + + def test_seek_beyond_eof_returns_minus_one(self, sparse_file): + file_size = sparse_file.stat().st_size + with open(str(sparse_file), "rb") as f: + result = _lseek(f, file_size + 1, _SEEK_DATA) + assert result == -1, \ + "SEEK_DATA beyond EOF must return -1, got %d" % result + + def test_seek_hole_beyond_eof_returns_minus_one(self, sparse_file): + file_size = sparse_file.stat().st_size + with open(str(sparse_file), "rb") as f: + result = _lseek(f, file_size + 1, _SEEK_HOLE) + assert result == -1 + + def test_seek_data_at_exactly_eof_returns_minus_one(self, sparse_file): + file_size = sparse_file.stat().st_size + with open(str(sparse_file), "rb") as f: + result = _lseek(f, file_size, _SEEK_DATA) + assert result == -1 + + def test_dense_file_seek_hole_returns_eof(self, tmp_path): + """On a fully dense file SEEK_HOLE at 0 should return the file size.""" + p = tmp_path / "dense_lseek.img" + p.write_bytes(b"Z" * _BS * 4) + file_size = p.stat().st_size + with open(str(p), "rb") as f: + hole = _lseek(f, 0, _SEEK_HOLE) + assert hole == file_size or hole == -1, ( + "SEEK_HOLE on dense file should return file_size or -1, got %d" % hole) + + def test_seek_does_not_move_caller_fd_position(self, sparse_file): + """_lseek uses os.lseek which moves the fd; caller must be aware.""" + with open(str(sparse_file), "rb") as f: + before = os.lseek(f.fileno(), 0, os.SEEK_CUR) + _lseek(f, 0, _SEEK_HOLE) + after = os.lseek(f.fileno(), 0, os.SEEK_CUR) + # The fd IS moved by os.lseek — _lseek is a raw lseek wrapper. + # This test documents that behaviour (it is intentional for _f_seek, + # problematic if mistakenly applied to _f_image). + assert after != before or True # just document; always passes + + +# =========================================================================== +# TestGetBlockSizeEdgeCases — get_block_size edge cases +# =========================================================================== + +class TestGetBlockSizeEdgeCases: + """get_block_size caps at 4 KiB and must always return a power of two.""" + + def test_result_is_power_of_two(self, tmp_path): + p = tmp_path / "bs.img" + p.write_bytes(b"\x00" * 4096) + with open(str(p), "rb") as f: + bs = get_block_size(f) + assert bs > 0 and (bs & (bs - 1)) == 0 + + def test_result_capped_at_4096(self, tmp_path): + """If the filesystem block size is > 4096, it must be clamped.""" + p = tmp_path / "bs_cap.img" + p.write_bytes(b"\x00" * 4096) + with open(str(p), "rb") as f: + bs = get_block_size(f) + assert bs <= 4096, "get_block_size must cap at 4096; got %d" % bs + + def test_result_divides_into_blocks_cnt_correctly(self, tmp_path): + """blocks_cnt computation must be exact (ceiling division).""" + for size in (1, 4095, 4096, 4097, 8192, 8193): + p = tmp_path / ("bs_div_%d.img" % size) + p.write_bytes(b"\x00" * size) + obj = FilemapNobmap(str(p)) + expected = (size + obj.block_size - 1) // obj.block_size + assert obj.blocks_cnt == expected, \ + "blocks_cnt wrong for size %d: got %d expected %d" % ( + size, obj.blocks_cnt, expected) + + +# =========================================================================== +# TestSparseCopyDestination — destination file edge cases +# =========================================================================== + +class TestSparseCopyDestination: + """sparse_copy destination creation and pre-existing file behaviour.""" + + @pytest.fixture + def src(self, tmp_path): + p = tmp_path / "src_dest.img" + p.write_bytes(b"\xcd" * _BS * 2) + return p + + def test_nonexistent_destination_is_created(self, src, tmp_path): + dst = tmp_path / "new_dst.img" + assert not dst.exists() + sparse_copy(str(src), str(dst)) + assert dst.exists() + assert dst.read_bytes()[:_BS * 2] == src.read_bytes() + + def test_existing_smaller_destination_is_overwritten(self, src, tmp_path): + dst = tmp_path / "small_dst.img" + dst.write_bytes(b"\x00" * _BS) # smaller than src + sparse_copy(str(src), str(dst)) + got = dst.read_bytes() + assert got[:_BS * 2] == src.read_bytes() + + def test_existing_larger_destination_preserves_prefix(self, src, tmp_path): + """When destination already exists and is larger, sparse_copy opens it + with 'r+b' (no truncate). The source bytes must be present at offset 0. + """ + dst = tmp_path / "big_dst.img" + # Pre-fill with recognisable sentinel bytes. + dst.write_bytes(b"\xff" * _BS * 4) + sparse_copy(str(src), str(dst)) + got = dst.read_bytes() + # Source bytes overwrite the beginning. + assert got[:_BS * 2] == src.read_bytes(), \ + "Source data not at offset 0 of larger destination" + + def test_missing_destination_directory_raises(self, src, tmp_path): + """If the destination directory does not exist, sparse_copy must raise.""" + dst = str(tmp_path / "no_such_dir" / "out.img") + with pytest.raises((Error, OSError, FileNotFoundError)): + sparse_copy(str(src), dst) + + def test_destination_content_correct_after_partial_overwrite(self, src, tmp_path): + """seek=_BS: the first block of dst must be unchanged.""" + dst = tmp_path / "partial.img" + sentinel = b"\xee" * _BS * 3 + dst.write_bytes(sentinel) + sparse_copy(str(src), str(dst), seek=_BS) + got = dst.read_bytes() + # First block untouched. + assert got[:_BS] == sentinel[:_BS], \ + "seek should not overwrite bytes before the seek offset" + # Source bytes start at _BS. + assert got[_BS:_BS * 3] == src.read_bytes(), \ + "Source data must appear at seek offset" + + +# =========================================================================== +# TestFilemapBaseEdgeCases — _FilemapBase edge cases not yet covered +# =========================================================================== + +class TestFilemapBaseEdgeCases: + """Edge cases in _FilemapBase initialisation.""" + + def test_blocks_cnt_ceiling_for_non_aligned_size(self, tmp_path): + """A file whose size is not a multiple of block_size needs ceiling div.""" + for extra in (1, 2, _BS - 1): + p = tmp_path / ("ceil_%d.img" % extra) + p.write_bytes(b"\xaa" * (_BS + extra)) + obj = FilemapNobmap(str(p)) + assert obj.blocks_cnt == 2, \ + "blocks_cnt for size %d should be 2, got %d" % ( + _BS + extra, obj.blocks_cnt) + + def test_blocks_cnt_exact_multiple(self, tmp_path): + p = tmp_path / "exact.img" + p.write_bytes(b"\xbb" * _BS * 3) + obj = FilemapNobmap(str(p)) + assert obj.blocks_cnt == 3 + + def test_image_size_reported_exactly(self, tmp_path): + for size in (1, 100, _BS, _BS * 3 + 7): + p = tmp_path / ("sz_%d.img" % size) + p.write_bytes(b"\x00" * size) + obj = FilemapNobmap(str(p)) + assert obj.image_size == size, \ + "image_size wrong for size %d: got %d" % (size, obj.image_size) + + def test_block_is_mapped_not_implemented_in_base(self, tmp_path): + """The base class block_is_mapped must raise Error (not NotImplemented).""" + p = tmp_path / "base.img" + p.write_bytes(b"\x00" * _BS) + obj = object.__new__(_FilemapBase) + _FilemapBase.__init__(obj, str(p)) + with pytest.raises(Error): + obj.block_is_mapped(0) + + def test_get_mapped_ranges_not_implemented_in_base(self, tmp_path): + p = tmp_path / "base2.img" + p.write_bytes(b"\x00" * _BS) + obj = object.__new__(_FilemapBase) + _FilemapBase.__init__(obj, str(p)) + with pytest.raises(Error): + list(obj.get_mapped_ranges(0, 1)) + + def test_f_image_is_readable_after_init(self, tmp_path): + """_f_image must be open and readable after __init__.""" + p = tmp_path / "readable.img" + p.write_bytes(b"\xcc" * _BS) + obj = FilemapNobmap(str(p)) + assert not obj._f_image.closed + obj._f_image.seek(0) + assert obj._f_image.read(4) == b"\xcc\xcc\xcc\xcc" + + def test_image_path_is_absolute_after_relative_input(self, tmp_path): + """_image_path should be stored as given (path, not resolved).""" + p = tmp_path / "rel.img" + p.write_bytes(b"\x00" * _BS) + obj = FilemapNobmap(str(p)) + # We passed an absolute path; _image_path must equal what was passed. + assert obj._image_path == str(p) + +# =========================================================================== +# TestNonAlignedSizes — hammer with sizes that are NOT powers of 2 and +# NOT multiples of 4096. The filemap block math, extent rounding, and +# skip/seek calculations all have latent off-by-one hazards when the file +# size or transfer length doesn't fall on a convenient boundary. +# +# Chosen sizes hit as many distinct alignment classes as possible: +# • Not a power of 2 +# • Not a multiple of 512 (disk sector) +# • Not a multiple of 4096 (typical fs block / _BS) +# • Prime-number sizes (hardest for any alignment assumption to survive) +# • Sizes just above/below every relevant boundary +# =========================================================================== + +# Prime sizes (defeat all alignment assumptions) +_PRIMES = [ + 3, 7, 13, 31, 97, 251, 509, 1021, 2039, 4093, 4099, + 8191, 8209, 16381, 16411, 32749, 65537, +] + +# Sizes that straddle block/sector/buffer boundaries but are not multiples +_STRADDLERS = [ + _BS + 1, # 4097 + _BS - 1, # 4095 + _BS * 2 + 3, # 8195 + _BS * 2 - 5, # 8187 + _BS * 3 + 7, # 12295 + _BS * 4 - 13, # 16371 + _BS * 8 + 17, # 32785 + _BS * 8 - 17, # 32751 + _BUF + 1, # 8193 (just over stdio buffer) + _BUF - 1, # 8191 (just under stdio buffer) + _BUF * 2 + 3, # 16387 + _BUF * 3 - 7, # 24569 +] + +# Non-power-of-2 multiples of small odd numbers +_ODD_MULTIPLES = [ + 3 * 512, # 1536 + 5 * 512, # 2560 + 3 * 1000, # 3000 + 7 * 512, # 3584 + 3 * _BS, # 12288 — IS a multiple of _BS, but included for skip/seek combos + 5 * _BS + 100, # 20580 + 7 * _BS + 111, # 28783 +] + +_ALL_WEIRD_SIZES = sorted(set(_PRIMES + _STRADDLERS + _ODD_MULTIPLES)) + + +class TestNonAlignedSizesFilemapInit: + """FilemapBase init with every weird size: image_size, block_size, and + blocks_cnt must all be consistent regardless of how un-aligned the size is. + """ + + @pytest.mark.parametrize("size", _ALL_WEIRD_SIZES) + def test_image_size_exact(self, tmp_path, size): + p = tmp_path / ("sz_%d.img" % size) + p.write_bytes(bytes(i % 251 for i in range(size))) + obj = FilemapNobmap(str(p)) + assert obj.image_size == size, \ + "image_size %d != expected %d" % (obj.image_size, size) + + @pytest.mark.parametrize("size", _ALL_WEIRD_SIZES) + def test_blocks_cnt_ceiling(self, tmp_path, size): + p = tmp_path / ("bc_%d.img" % size) + p.write_bytes(bytes(i % 251 for i in range(size))) + obj = FilemapNobmap(str(p)) + expected = (size + obj.block_size - 1) // obj.block_size + assert obj.blocks_cnt == expected, \ + "blocks_cnt %d != ceiling(%d/%d)=%d" % ( + obj.blocks_cnt, size, obj.block_size, expected) + + @pytest.mark.parametrize("size", _ALL_WEIRD_SIZES) + def test_fiemap_init_weird_size(self, tmp_path, size): + p = tmp_path / ("fz_%d.img" % size) + p.write_bytes(bytes(i % 251 for i in range(size))) + try: + obj = FilemapFiemap(str(p)) + except ErrorNotSupp: + pytest.skip("FIEMAP not supported") + assert obj.image_size == size + + @pytest.mark.parametrize("size", _ALL_WEIRD_SIZES) + def test_seek_init_weird_size(self, tmp_path, size): + p = tmp_path / ("sk_%d.img" % size) + p.write_bytes(bytes(i % 251 for i in range(size))) + try: + obj = FilemapSeek(str(p)) + except ErrorNotSupp: + pytest.skip("FilemapSeek not supported") + assert obj.image_size == size + + +class TestNonAlignedSizesCopyIdentity: + """sparse_copy of a weird-sized file must reproduce byte-for-byte content. + Tests all three filemap implementations. A single failing size points to + a block-boundary rounding error in that implementation. + """ + + def _src(self, tmp_path, size): + p = tmp_path / ("id_src_%d.img" % size) + # Non-repeating content: index mod a prime so no power-of-2 period + p.write_bytes(bytes(i % 251 for i in range(size))) + return p + + @pytest.mark.parametrize("size", _ALL_WEIRD_SIZES) + def test_nobmap_copy_identity(self, tmp_path, size): + src = self._src(tmp_path, size) + dst = tmp_path / ("id_dst_nb_%d.img" % size) + sparse_copy(str(src), str(dst), api=FilemapNobmap) + got = dst.read_bytes()[:size] + assert got == src.read_bytes(), \ + "FilemapNobmap identity copy wrong at size %d" % size + + @pytest.mark.parametrize("size", _ALL_WEIRD_SIZES) + def test_fiemap_copy_identity(self, tmp_path, size): + src = self._src(tmp_path, size) + dst = tmp_path / ("id_dst_fi_%d.img" % size) + try: + sparse_copy(str(src), str(dst), api=FilemapFiemap) + except ErrorNotSupp: + pytest.skip("FIEMAP not supported") + got = dst.read_bytes()[:size] + assert got == src.read_bytes(), \ + "FilemapFiemap identity copy wrong at size %d" % size + + @pytest.mark.parametrize("size", _ALL_WEIRD_SIZES) + def test_seek_copy_identity(self, tmp_path, size): + src = self._src(tmp_path, size) + dst = tmp_path / ("id_dst_sk_%d.img" % size) + try: + sparse_copy(str(src), str(dst), api=FilemapSeek) + except ErrorNotSupp: + pytest.skip("FilemapSeek not supported") + got = dst.read_bytes()[:size] + assert got == src.read_bytes(), \ + "FilemapSeek identity copy wrong at size %d" % size + + +class TestNonAlignedSkipValues: + """sparse_copy with skip values that are NOT multiples of _BS or _BUF. + Block-aligned skip values are already covered elsewhere; these sizes + probe the intermediate math at every un-aligned offset. + """ + + # Weird skip values to test — a mix of primes, near-boundary, and odd + _WEIRD_SKIPS = [ + 1, 3, 7, 13, 97, 251, 509, + _BS - 1, _BS + 1, _BS + 13, + _BUF - 1, _BUF + 1, _BUF + 7, + _BS * 2 - 3, _BS * 2 + 3, + _BS * 3 + 97, + ] + + def _make_src(self, tmp_path, total_size): + p = tmp_path / ("skip_src_%d.img" % total_size) + p.write_bytes(bytes(i % 251 for i in range(total_size))) + return p + + @pytest.mark.parametrize("skip", _WEIRD_SKIPS) + def test_skip_odd_offset_nobmap(self, tmp_path, skip): + total = skip + _BS * 4 + 97 # ensure there's data after skip + src = self._make_src(tmp_path, total) + dst = tmp_path / ("skip_nb_%d.img" % skip) + sparse_copy(str(src), str(dst), api=FilemapNobmap, skip=skip) + expected = src.read_bytes()[skip:] + got = dst.read_bytes()[:len(expected)] + assert got == expected, \ + "Nobmap skip=%d produced wrong data at first differing byte %d" % ( + skip, + next((i for i in range(len(expected)) if expected[i] != got[i]), + len(expected))) + + @pytest.mark.parametrize("skip", _WEIRD_SKIPS) + def test_skip_odd_offset_fiemap(self, tmp_path, skip): + total = skip + _BS * 4 + 97 + src = self._make_src(tmp_path, total) + dst = tmp_path / ("skip_fi_%d.img" % skip) + try: + sparse_copy(str(src), str(dst), api=FilemapFiemap, skip=skip) + except ErrorNotSupp: + pytest.skip("FIEMAP not supported") + expected = src.read_bytes()[skip:] + got = dst.read_bytes()[:len(expected)] + assert got == expected, \ + "FilemapFiemap skip=%d produced wrong data" % skip + + @pytest.mark.parametrize("skip", _WEIRD_SKIPS) + def test_skip_odd_offset_seek(self, tmp_path, skip): + total = skip + _BS * 4 + 97 + src = self._make_src(tmp_path, total) + dst = tmp_path / ("skip_sk_%d.img" % skip) + try: + sparse_copy(str(src), str(dst), api=FilemapSeek, skip=skip) + except ErrorNotSupp: + pytest.skip("FilemapSeek not supported") + expected = src.read_bytes()[skip:] + got = dst.read_bytes()[:len(expected)] + assert got == expected, \ + "FilemapSeek skip=%d produced wrong data" % skip + + @pytest.mark.parametrize("skip", _WEIRD_SKIPS) + def test_all_three_agree_on_odd_skip(self, tmp_path, skip): + """The three implementations must agree byte-for-byte on every skip.""" + total = skip + _BS * 4 + 97 + src = self._make_src(tmp_path, total) + results = {} + for cls in (FilemapFiemap, FilemapSeek, FilemapNobmap): + dst = tmp_path / ("ag_%s_%d.img" % (cls.__name__, skip)) + try: + sparse_copy(str(src), str(dst), api=cls, skip=skip) + results[cls.__name__] = dst.read_bytes() + except ErrorNotSupp: + pass + if len(results) < 2: + pytest.skip("fewer than 2 implementations available") + names = list(results) + ref = results[names[0]] + for name in names[1:]: + d = results[name] + trim = min(len(ref), len(d)) + assert ref[:trim] == d[:trim], \ + "%s vs %s disagree at skip=%d, first diff at byte %d" % ( + names[0], name, skip, + next((i for i in range(trim) if ref[i] != d[i]), trim)) + + +class TestNonAlignedLengthValues: + """sparse_copy with length values that are NOT multiples of 4096 or powers + of 2. Any block-rounding of the length would produce the wrong byte count + or wrong content in the destination. + """ + + _WEIRD_LENGTHS = [ + 1, 3, 7, 13, 97, 251, 509, 1021, + _BS - 1, _BS + 1, _BS + 13, + _BUF - 1, _BUF + 1, _BUF + 7, + _BS * 2 - 3, _BS * 2 + 3, + ] + + def _make_src(self, tmp_path, size): + p = tmp_path / ("len_src_%d.img" % size) + p.write_bytes(bytes(i % 251 for i in range(size))) + return p + + @pytest.mark.parametrize("length", _WEIRD_LENGTHS) + def test_length_exact_byte_count_nobmap(self, tmp_path, length): + """With FilemapNobmap, sparse_copy(length=L) must copy exactly L bytes.""" + total = max(length * 2, _BS * 4) + src = self._make_src(tmp_path, total) + dst = tmp_path / ("len_nb_%d.img" % length) + sparse_copy(str(src), str(dst), api=FilemapNobmap, length=length) + got = dst.read_bytes() + expected = src.read_bytes()[:length] + assert got[:length] == expected, \ + "Nobmap length=%d: first %d bytes wrong" % (length, length) + + @pytest.mark.parametrize("length", _WEIRD_LENGTHS) + def test_length_exact_byte_count_fiemap(self, tmp_path, length): + total = max(length * 2, _BS * 4) + src = self._make_src(tmp_path, total) + dst = tmp_path / ("len_fi_%d.img" % length) + try: + sparse_copy(str(src), str(dst), api=FilemapFiemap, length=length) + except ErrorNotSupp: + pytest.skip() + got = dst.read_bytes() + expected = src.read_bytes()[:length] + assert got[:length] == expected, \ + "FilemapFiemap length=%d: first %d bytes wrong" % (length, length) + + @pytest.mark.parametrize("length", _WEIRD_LENGTHS) + def test_length_exact_byte_count_seek(self, tmp_path, length): + total = max(length * 2, _BS * 4) + src = self._make_src(tmp_path, total) + dst = tmp_path / ("len_sk_%d.img" % length) + try: + sparse_copy(str(src), str(dst), api=FilemapSeek, length=length) + except ErrorNotSupp: + pytest.skip() + got = dst.read_bytes() + expected = src.read_bytes()[:length] + assert got[:length] == expected, \ + "FilemapSeek length=%d: first %d bytes wrong" % (length, length) + + +class TestNonAlignedSeekValues: + """sparse_copy with seek offsets that are not multiples of _BS or _BUF. + The preamble (bytes before seek) must be exactly zero and the source data + must start at the exact byte offset with no shift or rounding. + """ + + _WEIRD_SEEKS = [ + 1, 3, 7, 13, 97, 251, 509, + _BS - 1, _BS + 1, _BS + 13, + _BUF - 1, _BUF + 1, _BUF + 7, + ] + + def _make_src(self, tmp_path): + p = tmp_path / "seek_src_w.img" + p.write_bytes(bytes(i % 251 for i in range(_BS * 3))) + return p + + @pytest.mark.parametrize("seek", _WEIRD_SEEKS) + def test_seek_preamble_is_zero(self, tmp_path, seek): + src = self._make_src(tmp_path) + dst = tmp_path / ("seek_w_%d.img" % seek) + sparse_copy(str(src), str(dst), seek=seek) + got = dst.read_bytes() + assert got[:seek] == b"\x00" * seek, \ + "Bytes before seek=%d must be zero; got %r" % (seek, got[:min(8, seek)]) + + @pytest.mark.parametrize("seek", _WEIRD_SEEKS) + def test_seek_data_starts_at_exact_offset(self, tmp_path, seek): + src = self._make_src(tmp_path) + src_bytes = src.read_bytes() + dst = tmp_path / ("seek_data_%d.img" % seek) + sparse_copy(str(src), str(dst), seek=seek) + got = dst.read_bytes() + assert got[seek:seek + len(src_bytes)] == src_bytes, \ + "Source data not at exact offset seek=%d" % seek + + +class TestNonAlignedSparseHoles: + """Sparse files whose holes and data extents are NOT block-aligned in + the 'nice' sense: data written at weird offsets within the file forces + FIEMAP/SEEK_HOLE to deal with partial extents and sub-block precision. + """ + + def _make_sparse_weird(self, tmp_path, name, total, data_at): + """data_at: list of (byte_offset, size) to write with non-repeating bytes.""" + p = tmp_path / name + with open(str(p), "wb") as f: + f.truncate(total) + written = 0 + for off, sz in data_at: + f.seek(off) + f.write(bytes((written + i) % 251 for i in range(sz))) + written += sz + return p + + def test_data_at_non_block_offset_within_block(self, tmp_path): + """Data written at byte 13 (inside block 0). FIEMAP should mark block 0 + as a mapped extent; block 1 onward are holes. + The copy must reproduce byte 13 correctly. + """ + total = _BS * 4 + src = self._make_sparse_weird(tmp_path, "non_blk_off.img", total, + [(13, 100)]) + src_bytes = src.read_bytes() + dst = tmp_path / "non_blk_off_dst.img" + sparse_copy(str(src), str(dst), api=FilemapFiemap) + got = dst.read_bytes() + assert got[13:113] == src_bytes[13:113], \ + "Data at byte offset 13 corrupted after FIEMAP sparse_copy" + + def test_data_spanning_block_boundary_at_odd_offset(self, tmp_path): + """Data starts at byte (_BS - 7): spans the block-0/block-1 boundary. + Both blocks must appear mapped; copy must be byte-exact at the crossing. + """ + total = _BS * 4 + start = _BS - 7 + length = 14 # 7 bytes in block 0, 7 bytes in block 1 + src = self._make_sparse_weird(tmp_path, "cross_blk.img", total, + [(start, length)]) + src_bytes = src.read_bytes() + dst = tmp_path / "cross_blk_dst.img" + sparse_copy(str(src), str(dst), api=FilemapFiemap) + got = dst.read_bytes() + assert got[start:start + length] == src_bytes[start:start + length], \ + "Block-boundary-crossing data corrupted" + + @pytest.mark.parametrize("offset", [1, 3, 7, 13, 97, 251, _BS // 2 - 1, + _BS // 2, _BS // 2 + 1, _BS - 1]) + def test_single_byte_data_at_weird_intra_block_offsets(self, tmp_path, offset): + """One byte of data at various sub-block offsets. The copy must reproduce + exactly that one byte at exactly that offset. + """ + if offset >= _BS: + pytest.skip("offset >= block size") + total = _BS * 2 + src = self._make_sparse_weird(tmp_path, "single_%d.img" % offset, total, + [(offset, 1)]) + src_bytes = src.read_bytes() + dst = tmp_path / ("single_%d_dst.img" % offset) + sparse_copy(str(src), str(dst), api=FilemapFiemap) + got = dst.read_bytes() + assert got[offset] == src_bytes[offset], \ + "Single byte at offset %d wrong after copy: expected %r got %r" % ( + offset, src_bytes[offset], got[offset]) + + def test_all_implementations_agree_on_odd_sparse_layout(self, tmp_path): + """Three different blocks have data at non-block-aligned internal offsets. + All three Filemap implementations must produce identical copies. + """ + total = _BS * 8 + src = self._make_sparse_weird(tmp_path, "multi_weird.img", total, [ + (7, 250), # inside block 0, odd start + (_BS + 13, 100), # inside block 1, odd start + (_BS * 5 - 17, 34), # spans blocks 4-5 + ]) + src_bytes = src.read_bytes() + results = {} + for cls in (FilemapFiemap, FilemapSeek, FilemapNobmap): + dst = tmp_path / ("mw_%s.img" % cls.__name__) + try: + sparse_copy(str(src), str(dst), api=cls) + results[cls.__name__] = dst.read_bytes() + except ErrorNotSupp: + pass + if len(results) < 2: + pytest.skip("fewer than 2 implementations") + ref_name, ref = next(iter(results.items())) + for name, data in results.items(): + trim = min(len(ref), len(data)) + diff_at = next((i for i in range(trim) if ref[i] != data[i]), None) + assert diff_at is None, \ + "%s vs %s differ at byte %d (src=%r ref=%r got=%r)" % ( + ref_name, name, diff_at, + src_bytes[diff_at], + ref[diff_at], + data[diff_at]) + + +class TestNonAlignedCombined: + """Combined weird sizes × weird skips — the most demanding combinations. + A mishandled block-boundary in EITHER the image size OR the skip offset + produces wrong data. Each test case hits a distinct pair. + """ + + # Curated pairs: (file_size, skip) — both weird, independently chosen + _COMBOS = [ + # (file_size, skip) + (251, 0), + (251, 1), + (509, 13), + (1021, 97), + (_BS + 1, 1), + (_BS + 1, _BS - 1), + (_BS - 1, 1), + (_BUF + 3, _BS + 1), + (_BUF + 3, _BUF - 7), + (_BUF * 2 + 13, _BUF + 1), + (_BS * 3 + 97, _BS + 13), + (_BS * 5 - 7, _BUF + 17), + ] + + @pytest.mark.parametrize("size,skip", _COMBOS) + def test_combined_weird_size_and_skip(self, tmp_path, size, skip): + if skip >= size: + pytest.skip("skip >= size") + src = tmp_path / ("combo_src_%d_%d.img" % (size, skip)) + src.write_bytes(bytes(i % 251 for i in range(size))) + src_bytes = src.read_bytes() + expected = src_bytes[skip:] + + for cls in (FilemapFiemap, FilemapSeek, FilemapNobmap): + dst = tmp_path / ("combo_%s_%d_%d.img" % (cls.__name__, size, skip)) + try: + sparse_copy(str(src), str(dst), api=cls, skip=skip) + except ErrorNotSupp: + continue + got = dst.read_bytes()[:len(expected)] + diff_at = next((i for i in range(len(expected)) + if i < len(got) and expected[i] != got[i]), None) + assert diff_at is None, \ + "%s size=%d skip=%d: wrong at byte %d (expected %r got %r)" % ( + cls.__name__, size, skip, diff_at, + expected[diff_at], got[diff_at]) + + @pytest.mark.parametrize("size,skip", _COMBOS) + def test_all_implementations_agree_on_combined(self, tmp_path, size, skip): + if skip >= size: + pytest.skip("skip >= size") + src = tmp_path / ("cag_src_%d_%d.img" % (size, skip)) + src.write_bytes(bytes(i % 251 for i in range(size))) + results = {} + for cls in (FilemapFiemap, FilemapSeek, FilemapNobmap): + dst = tmp_path / ("cag_%s_%d_%d.img" % (cls.__name__, size, skip)) + try: + sparse_copy(str(src), str(dst), api=cls, skip=skip) + results[cls.__name__] = dst.read_bytes() + except ErrorNotSupp: + pass + if len(results) < 2: + pytest.skip("fewer than 2 implementations") + names = list(results) + ref = results[names[0]] + for name in names[1:]: + d = results[name] + trim = min(len(ref), len(d)) + diff = next((i for i in range(trim) if ref[i] != d[i]), None) + assert diff is None, \ + "%s vs %s disagree at byte %d for size=%d skip=%d" % ( + names[0], name, diff, size, skip) + + +class TestBufferBoundaryExhaustive: + """Exhaustively test skip values that straddle DEFAULT_BUFFER_SIZE (_BUF). + Any rounding of skip to a multiple of _BUF would corrupt output here. + The file is large enough that there is data on both sides of every boundary. + """ + + # Every value from _BUF-16 to _BUF+16 plus _BUF*2 ± small offsets + _BUF_SKIPS = ( + list(range(_BUF - 16, _BUF + 17)) + # tight band around _BUF + list(range(_BUF * 2 - 8, _BUF * 2 + 9)) + # tight band around 2*_BUF + [_BUF + k for k in (0, 1, 3, 7, 13, 97, 251)] # scattered above _BUF + ) + + @staticmethod + @pytest.fixture(scope="class") + def large_src(tmp_path_factory): + tmp = tmp_path_factory.mktemp("bufsrc") + total = _BUF * 3 + 300 + p = tmp / "buf_src.img" + p.write_bytes(bytes(i % 251 for i in range(total))) + return p + + @pytest.mark.parametrize("skip", _BUF_SKIPS) + def test_buffer_boundary_skip_nobmap(self, large_src, tmp_path, skip): + src_bytes = large_src.read_bytes() + if skip >= len(src_bytes): + pytest.skip("skip >= file size") + dst = tmp_path / ("bb_nb_%d.img" % skip) + sparse_copy(str(large_src), str(dst), api=FilemapNobmap, skip=skip) + expected = src_bytes[skip:] + got = dst.read_bytes()[:len(expected)] + diff = next((i for i in range(len(expected)) if expected[i] != got[i]), None) + assert diff is None, \ + "Nobmap buf-boundary skip=%d wrong at byte %d (exp %r got %r)" % ( + skip, diff, expected[diff], got[diff]) + + @pytest.mark.parametrize("skip", _BUF_SKIPS) + def test_buffer_boundary_skip_fiemap(self, large_src, tmp_path, skip): + src_bytes = large_src.read_bytes() + if skip >= len(src_bytes): + pytest.skip("skip >= file size") + dst = tmp_path / ("bb_fi_%d.img" % skip) + try: + sparse_copy(str(large_src), str(dst), api=FilemapFiemap, skip=skip) + except ErrorNotSupp: + pytest.skip() + expected = src_bytes[skip:] + got = dst.read_bytes()[:len(expected)] + diff = next((i for i in range(len(expected)) if expected[i] != got[i]), None) + assert diff is None, \ + "Fiemap buf-boundary skip=%d wrong at byte %d" % (skip, diff) + + @pytest.mark.parametrize("skip", _BUF_SKIPS) + def test_buffer_boundary_skip_seek(self, large_src, tmp_path, skip): + src_bytes = large_src.read_bytes() + if skip >= len(src_bytes): + pytest.skip("skip >= file size") + dst = tmp_path / ("bb_sk_%d.img" % skip) + try: + sparse_copy(str(large_src), str(dst), api=FilemapSeek, skip=skip) + except ErrorNotSupp: + pytest.skip() + expected = src_bytes[skip:] + got = dst.read_bytes()[:len(expected)] + diff = next((i for i in range(len(expected)) if expected[i] != got[i]), None) + assert diff is None, \ + "Seek buf-boundary skip=%d wrong at byte %d" % (skip, diff) diff --git a/tests/unit/test_help.py b/tests/unit/test_help.py new file mode 100644 index 000000000000..9ee74bde3b1c --- /dev/null +++ b/tests/unit/test_help.py @@ -0,0 +1,118 @@ +r""" +Unit tests for wic/help.py. + +The help module is mostly dispatch logic plus large help-text strings. +display_help(), wic_help(), and invoke_subcommand() are pure (apart +from display_help's pager, which is only reached for a known topic), so +they can be driven directly. + +Where wic does not yet behave correctly, the test is marked xfail with +a short reason. +""" + +import sys +import types +from pathlib import Path + +import pytest + +_SRC = Path(__file__).resolve().parent.parent.parent / "src" +if str(_SRC) not in sys.path: + sys.path.insert(0, str(_SRC)) + +import wic.help as helpmod +from wic.help import display_help, invoke_subcommand + + +# --------------------------------------------------------------------------- +# display_help +# --------------------------------------------------------------------------- + +class TestDisplayHelp: + + def test_unknown_subcommand_returns_false(self): + assert display_help("nope", {}) is False + + def test_unknown_subcommand_with_populated_map(self): + subs = {"create": (lambda *a: None, "usage", "help text")} + assert display_help("missing", subs) is False + + +# --------------------------------------------------------------------------- +# invoke_subcommand dispatch +# --------------------------------------------------------------------------- + +class _Parser: + def __init__(self): + self.printed = False + + def print_help(self): + self.printed = True + + +class TestInvokeSubcommand: + + def test_no_command_returns_1_and_prints_help(self): + parser = _Parser() + rc = invoke_subcommand(types.SimpleNamespace(command=None), + parser, "usage", {}) + assert rc == 1 + assert parser.printed is True + + def test_unknown_command_returns_1(self): + parser = _Parser() + rc = invoke_subcommand(types.SimpleNamespace(command="bogus"), + parser, "usage", {}) + assert rc == 1 + assert parser.printed is True + + def test_known_command_dispatches_with_usage(self): + captured = {} + + def handler(args, usage): + captured["usage"] = usage + + subs = {"create": (handler, "CREATE_USAGE", "help text")} + invoke_subcommand(types.SimpleNamespace(command="create"), + _Parser(), "usage", subs) + assert captured["usage"] == "CREATE_USAGE" + + def test_help_command_dispatches_to_wic_help(self): + """`wic help` routes through invoke_subcommand to wic_help(). + + wic_help is defined as a function early in help.py but a later + module-level string of the same name shadows it, so by import + time the name is a str. invoke_subcommand then calls it and + raises 'str object is not callable' for the plain `wic help` + path. + """ + subs = {"create": (lambda *a: None, "u", "h")} + args = types.SimpleNamespace(command="help", help_topic=None) + try: + invoke_subcommand(args, _Parser(), "main usage", subs) + except TypeError as exc: + if "not callable" in str(exc): + pytest.xfail( + "wic_help() is shadowed by a later module-level string " + "of the same name; `wic help` raises 'str object is not " + "callable' instead of printing usage") + raise + + +# --------------------------------------------------------------------------- +# wic_help name resolution +# --------------------------------------------------------------------------- + +class TestWicHelpName: + + def test_wic_help_is_callable(self): + """The wic_help symbol should be the dispatcher function. + + It is currently overwritten by a help-text string assigned later + in the module, so the callable is lost. + """ + if not callable(getattr(helpmod, "wic_help")): + pytest.xfail( + "wic_help is a str at import time (shadowed by a later " + "module-level assignment); the dispatcher function is lost") + assert callable(helpmod.wic_help) diff --git a/tests/unit/test_ksparser_parse.py b/tests/unit/test_ksparser_parse.py new file mode 100644 index 000000000000..5ed0accd5008 --- /dev/null +++ b/tests/unit/test_ksparser_parse.py @@ -0,0 +1,209 @@ +r""" +End-to-end tests for the KickStart .wks parser (wic/ksparser.py). + +KickStart(confpath) reads a .wks file line by line, runs each line +through an argparse-based grammar, validates filesystem/option +combinations, and builds Partition objects plus a bootloader config. +It is pure logic apart from get_bitbake_var() (used for ${VAR} +expansion and the bootloader APPEND), which these tests neutralise so +no bitbake or host tools are needed. + +Each test writes a small .wks file and asserts on the parsed result or +on the specific error raised. Where wic does not yet behave correctly, +the test is marked xfail with a short reason. +""" + +import sys +from pathlib import Path + +import pytest + +_SRC = Path(__file__).resolve().parent.parent.parent / "src" +if str(_SRC) not in sys.path: + sys.path.insert(0, str(_SRC)) + +import wic.misc as misc +import wic.ksparser as ksparser +from wic.ksparser import KickStart, KickStartError + + +@pytest.fixture(autouse=True) +def no_bitbake(monkeypatch): + """Neutralise bitbake variable lookups in both modules. + + get_bitbake_var is imported into ksparser's namespace and also used + via misc; patch both so ${VAR} expansion and APPEND resolve to None + instead of trying to run `bitbake -e`. + """ + monkeypatch.setattr(misc, "get_bitbake_var", lambda *a, **k: None) + monkeypatch.setattr(ksparser, "get_bitbake_var", lambda *a, **k: None) + + +def _wks(tmp_path, text): + path = tmp_path / "test.wks" + path.write_text(text) + return str(path) + + +def _parse(tmp_path, text): + return KickStart(_wks(tmp_path, text)) + + +# --------------------------------------------------------------------------- +# Happy path +# --------------------------------------------------------------------------- + +class TestParseHappyPath: + + def test_single_part(self, tmp_path): + ks = _parse(tmp_path, "part / --source rootfs --fstype ext4 --size 100M\n") + assert len(ks.partitions) == 1 + p = ks.partitions[0] + assert p.fstype == "ext4" + assert p.size == 100 * 1024 + assert p.mountpoint == "/" + + def test_defaults_applied_without_fixed_size(self, tmp_path): + ks = _parse(tmp_path, "part / --source rootfs --fstype ext4 --size 1M\n") + p = ks.partitions[0] + assert p.overhead_factor == KickStart.DEFAULT_OVERHEAD_FACTOR + assert p.extra_filesystem_space == KickStart.DEFAULT_EXTRA_FILESYSTEM_SPACE + + def test_partnum_increments(self, tmp_path): + ks = _parse( + tmp_path, + "part /boot --source bootimg --fstype vfat --size 50M\n" + "part / --source rootfs --fstype ext4 --size 100M\n", + ) + assert [p.lineno for p in ks.partitions] == [1, 2] + assert len(ks.partitions) == 2 + + def test_bootloader_parsed(self, tmp_path): + ks = _parse(tmp_path, "bootloader --ptable gpt\n") + assert ks.bootloader.ptable == "gpt" + + def test_comment_and_blank_lines_ignored(self, tmp_path): + ks = _parse( + tmp_path, + "# a comment\n" + "\n" + " \n" + "part / --source rootfs --fstype ext4 --size 1M\n", + ) + assert len(ks.partitions) == 1 + + +# --------------------------------------------------------------------------- +# Degenerate files +# --------------------------------------------------------------------------- + +class TestParseDegenerateFiles: + + def test_empty_file(self, tmp_path): + ks = _parse(tmp_path, "") + assert ks.partitions == [] + assert ks.bootloader is not None # defaults filled in + + def test_comment_only_file(self, tmp_path): + ks = _parse(tmp_path, "# only comments\n# nothing else\n") + assert ks.partitions == [] + + def test_crlf_line_endings(self, tmp_path): + ks = _parse(tmp_path, "part / --source rootfs --fstype ext4 --size 1M\r\n") + assert len(ks.partitions) == 1 + + def test_unknown_directive_rejected(self, tmp_path): + with pytest.raises(KickStartError): + _parse(tmp_path, "frobnicate everything\n") + + +# --------------------------------------------------------------------------- +# Mutually-exclusive and invalid option combinations +# --------------------------------------------------------------------------- + +class TestParseInvalidCombinations: + + def test_size_and_fixed_size_conflict(self, tmp_path): + with pytest.raises(KickStartError): + _parse(tmp_path, "part / --source rootfs --fstype ext4 " + "--size 100M --fixed-size 200M\n") + + def test_fixed_size_with_overhead_factor_rejected(self, tmp_path): + with pytest.raises(KickStartError): + _parse(tmp_path, "part / --source rootfs --fstype ext4 " + "--fixed-size 100M --overhead-factor 1.5\n") + + def test_squashfs_with_label_rejected(self, tmp_path): + with pytest.raises(KickStartError): + _parse(tmp_path, "part / --source rootfs --fstype squashfs --label boot\n") + + def test_squashfs_with_fsuuid_rejected(self, tmp_path): + with pytest.raises(KickStartError): + _parse(tmp_path, "part / --source rootfs --fstype squashfs --fsuuid 0x1234\n") + + def test_erofs_with_label_rejected(self, tmp_path): + with pytest.raises(KickStartError): + _parse(tmp_path, "part / --source rootfs --fstype erofs --label x --size 1M\n") + + def test_use_label_without_label_rejected(self, tmp_path): + with pytest.raises(KickStartError): + _parse(tmp_path, "part / --source rootfs --fstype ext4 --size 1M --use-label\n") + + def test_msdos_fsuuid_too_long_rejected(self, tmp_path): + with pytest.raises(KickStartError): + _parse(tmp_path, "part / --source rootfs --fstype msdos " + "--size 1M --fsuuid 0xDEADBEEF12\n") + + +# --------------------------------------------------------------------------- +# Multiple bootloaders +# --------------------------------------------------------------------------- + +class TestParseBootloader: + + def test_two_bootloaders_rejected(self, tmp_path): + with pytest.raises(KickStartError): + _parse(tmp_path, "bootloader --ptable gpt\nbootloader --ptable msdos\n") + + def test_msdos_diskid_integer(self, tmp_path): + ks = _parse(tmp_path, "bootloader --ptable msdos --diskid 0x12345678\n") + assert ks.bootloader.diskid == 0x12345678 + + def test_gpt_bad_diskid_raises_kickstart_error(self, tmp_path): + with pytest.raises(KickStartError): + _parse(tmp_path, "bootloader --ptable gpt --diskid not-a-uuid\n") + + def test_msdos_bad_diskid_raises_kickstart_error(self, tmp_path): + """A bad --diskid on an msdos ptable must raise a described + KickStartError. The error branch instead references self.ptable, + an attribute KickStart does not have, so it raises AttributeError + before the intended message is built. + """ + try: + _parse(tmp_path, "bootloader --ptable msdos --diskid not-an-int\n") + except AttributeError: + pytest.xfail( + "msdos --diskid error path references self.ptable (no such " + "attribute on KickStart); raises AttributeError instead of " + "a KickStartError describing the bad --diskid") + except KickStartError: + return + pytest.fail("a non-integer msdos --diskid should be rejected") + + +# --------------------------------------------------------------------------- +# Include handling +# --------------------------------------------------------------------------- + +class TestParseInclude: + + def test_include_missing_file_rejected(self, tmp_path): + with pytest.raises(KickStartError): + _parse(tmp_path, "include does-not-exist.wks\n") + + def test_include_pulls_in_partitions(self, tmp_path): + inc = tmp_path / "inc.wks" + inc.write_text("part / --source rootfs --fstype ext4 --size 5M\n") + ks = _parse(tmp_path, "include inc.wks\n") + assert len(ks.partitions) == 1 + assert ks.partitions[0].size == 5 * 1024 diff --git a/tests/unit/test_ksparser_types.py b/tests/unit/test_ksparser_types.py new file mode 100644 index 000000000000..556a3aac609a --- /dev/null +++ b/tests/unit/test_ksparser_types.py @@ -0,0 +1,400 @@ +""" +Unit tests for ksparser.py custom argparse types. + +For every type parser, every parameter is probed at: + - valid values at and around the documented boundary + - the empty string, whitespace-only, None + - special characters, very long strings + - values that are syntactically plausible but semantically wrong + +Where wic does not yet behave correctly, the test is marked xfail with a +short reason. An xfail still runs: it pins the expected behaviour and +flips to XPASS (then to a plain pass once the code is fixed). +""" + +import sys +import pytest +from pathlib import Path + +_SRC = Path(__file__).resolve().parent.parent.parent / "src" +if str(_SRC) not in sys.path: + sys.path.insert(0, str(_SRC)) + +from argparse import ArgumentTypeError +from wic.ksparser import sizetype, overheadtype, systemidtype + + +# --------------------------------------------------------------------------- +# sizetype +# --------------------------------------------------------------------------- +# +# sizetype(default, size_in_bytes=False) returns a callable f(arg) that +# converts "1M", "512K", "1G" etc. to integers. +# +# Return unit when size_in_bytes=False, mult=1: +# K/k -> size * 1 (raw KiB count) +# M -> size * 1024 (KiB) +# G -> size * 1024*1024 (KiB) +# bare -> uses default suffix +# +# Return unit when size_in_bytes=True, mult=1024: +# S/s -> size * 512 (bytes) +# K/k -> size * 1024 (bytes) +# M -> size * 1024^2 (bytes) +# G -> size * 1024^3 (bytes) +# --------------------------------------------------------------------------- + +class TestSizetype: + + def _sz(self, default="M", size_in_bytes=False): + return sizetype(default, size_in_bytes) + + # --- valid inputs (size_in_bytes=False, default="M") -------------------- + + @pytest.mark.parametrize("arg,expected", [ + ("1M", 1 * 1024), + ("10M", 10 * 1024), + ("512M", 512 * 1024), + ("1G", 1 * 1024 * 1024), + ("2G", 2 * 1024 * 1024), + ("1K", 1), + ("1k", 1), + ("1024K", 1024), + ("512k", 512), + ("0M", 0), + ("0", 0), + ]) + def test_valid_with_M_default(self, arg, expected): + assert self._sz("M")(arg) == expected + + def test_bare_integer_uses_default_suffix(self): + # "100" with default="M" -> 100 * 1024 KiB + assert self._sz("M")("100") == 100 * 1024 + + def test_bare_integer_with_K_default(self): + assert self._sz("K")("512") == 512 + + # --- size_in_bytes=True ------------------------------------------------- + + def test_size_in_bytes_S_uppercase(self): + assert self._sz("K", size_in_bytes=True)("1S") == 512 + + def test_size_in_bytes_s_lowercase(self): + assert self._sz("K", size_in_bytes=True)("1s") == 512 + + def test_size_in_bytes_K(self): + assert self._sz("K", size_in_bytes=True)("1K") == 1024 + + def test_size_in_bytes_M(self): + assert self._sz("M", size_in_bytes=True)("1M") == 1024 * 1024 + + def test_size_in_bytes_G(self): + assert self._sz("M", size_in_bytes=True)("1G") == 1024 ** 3 + + # --- suffix case sensitivity -------------------------------------------- + + def test_uppercase_M_valid(self): + assert self._sz("M")("1M") == 1024 + + def test_uppercase_G_valid(self): + assert self._sz("M")("1G") == 1024 * 1024 + + def test_uppercase_K_valid(self): + assert self._sz("M")("1K") == 1 + + def test_lowercase_k_valid(self): + assert self._sz("M")("1k") == 1 + + def test_lowercase_m_rejected(self): + # 'm' is not in the accepted suffix set + with pytest.raises(ArgumentTypeError): + self._sz("M")("1m") + + def test_lowercase_g_rejected(self): + with pytest.raises(ArgumentTypeError): + self._sz("M")("1g") + + # --- whitespace handling ------------------------------------------------ + + def test_leading_whitespace_accepted(self): + # Python int() strips leading whitespace, so " 1M" -> size=1 + assert self._sz("M")(" 1M") == 1024 + + def test_embedded_whitespace_accepted(self): + # "1 M" -> suffix="M", size=int("1 ")=1 (int() strips trailing space) + assert self._sz("M")("1 M") == 1024 + + def test_trailing_whitespace_on_suffix_rejected(self): + # "1M " -> suffix=" ", not a valid suffix + with pytest.raises(ArgumentTypeError): + self._sz("M")("1M ") + + # --- invalid format ----------------------------------------------------- + + @pytest.mark.parametrize("bad", [ + "abc", # no digits + "M", # suffix only + "1X", # unsupported suffix + "1.5M", # float not accepted + "1M ", # trailing space on the suffix position + "--1M", # double minus + "1e3M", # scientific notation + ]) + def test_invalid_format_raises(self, bad): + with pytest.raises((ArgumentTypeError, ValueError)): + self._sz("M")(bad) + + def test_empty_string_raises(self): + with pytest.raises((ArgumentTypeError, ValueError)): + self._sz("M")("") + + def test_whitespace_only_raises(self): + with pytest.raises((ArgumentTypeError, ValueError)): + self._sz("M")(" ") + + def test_none_raises(self): + with pytest.raises((TypeError, AttributeError)): + self._sz("M")(None) + + def test_special_chars_raise(self): + for bad in ["1;M", "1|M", "$(1)M", "1`M"]: + with pytest.raises((ArgumentTypeError, ValueError)): + self._sz("M")(bad) + + # --- known bugs --------------------------------------------------------- + + @pytest.mark.xfail(reason="sizetype accepts negative sizes silently") + def test_negative_size_rejected(self): + # "-1M" -> suffix="M", size=int("-1")=-1 -> returns -1024 + # A negative size is nonsensical and should be rejected. + with pytest.raises(ArgumentTypeError): + self._sz("M")("-1M") + + def test_very_large_value_does_not_crash(self): + # Extremely large values should either be accepted or raise cleanly + try: + result = self._sz("M")("999999999M") + assert isinstance(result, int) + except (ArgumentTypeError, OverflowError): + pass # also acceptable + + +# --------------------------------------------------------------------------- +# overheadtype +# --------------------------------------------------------------------------- +# +# overheadtype(arg) -> float, must be strictly > 1.0 +# +# Known bugs: +# the error-path format string "...message..." % arg crashes with +# TypeError when arg is passed to a string with no %s placeholder. +# Any value < 1.0 causes TypeError, not ArgumentTypeError. +# float('nan') and float('inf') bypass the < 1.0 guard. +# the boundary condition uses < instead of <=, so 1.0 is accepted +# even though the message says "should be > 1.0". +# --------------------------------------------------------------------------- + +class TestOverheadtype: + + # --- valid inputs ------------------------------------------------------- + + @pytest.mark.parametrize("arg,approx", [ + ("1.1", 1.1), + ("1.3", 1.3), + ("2.0", 2.0), + ("100.0", 100.0), + ("1.001", 1.001), + ("2", 2.0), # integer string -> float + ]) + def test_valid_values(self, arg, approx): + assert overheadtype(arg) == pytest.approx(approx) + + def test_very_large_value_accepted(self): + result = overheadtype("1000000.0") + assert result == pytest.approx(1000000.0) + + # --- boundary: exactly 1.0 ---------------------------------------------- + + @pytest.mark.xfail(reason="overheadtype uses '<' not '<='; 1.0 is accepted but should not be") + def test_exactly_one_point_zero_rejected(self): + with pytest.raises(ArgumentTypeError): + overheadtype("1.0") + + def test_just_above_one_accepted(self): + assert overheadtype("1.001") > 1.0 + + # --- values below 1.0: TypeError ---------------------------- + # The format string "Overhead factor should be > 1.0" % arg has no %s, + # so Python raises TypeError when arg is substituted. + + @pytest.mark.xfail(raises=TypeError, + reason="format string bug raises TypeError instead of ArgumentTypeError") + def test_negative_value_raises_argument_type_error(self): + with pytest.raises(ArgumentTypeError): + overheadtype("-1.0") + + @pytest.mark.xfail(raises=TypeError, + reason="format string bug raises TypeError instead of ArgumentTypeError") + def test_zero_raises_argument_type_error(self): + with pytest.raises(ArgumentTypeError): + overheadtype("0.0") + + def test_negative_value_raises_something(self): + # Whatever exception is raised, it must not be a silent return. + with pytest.raises((ArgumentTypeError, TypeError, ValueError)): + overheadtype("-1.0") + + def test_zero_raises_something(self): + with pytest.raises((ArgumentTypeError, TypeError, ValueError)): + overheadtype("0.0") + + # --- nan and inf -------------------------------------------------- + + @pytest.mark.xfail(reason="float('nan') bypasses the < 1.0 guard; nan is accepted") + def test_nan_rejected(self): + with pytest.raises(ArgumentTypeError): + overheadtype("nan") + + @pytest.mark.xfail(reason="float('inf') > 1.0 is True; inf is accepted as a valid factor") + def test_inf_rejected(self): + with pytest.raises(ArgumentTypeError): + overheadtype("inf") + + def test_negative_inf_raises_something(self): + # -inf < 1.0 triggers the TypeError + with pytest.raises((ArgumentTypeError, TypeError)): + overheadtype("-inf") + + # --- whitespace: Python float() strips it, so these are valid ----------- + + def test_leading_whitespace_accepted(self): + assert overheadtype(" 1.1") == pytest.approx(1.1) + + def test_trailing_whitespace_accepted(self): + assert overheadtype("1.1 ") == pytest.approx(1.1) + + # --- invalid format ----------------------------------------------------- + + @pytest.mark.parametrize("bad", [ + "abc", "1.1.1", "M", "1,3", "1 1", "--1.1", + ]) + def test_non_numeric_raises(self, bad): + with pytest.raises(ArgumentTypeError): + overheadtype(bad) + + def test_empty_string_raises(self): + with pytest.raises(ArgumentTypeError): + overheadtype("") + + def test_none_raises(self): + with pytest.raises((ArgumentTypeError, TypeError)): + overheadtype(None) + + # --- error message quality ---------------------------------------- + + def test_error_message_not_empty_for_non_numeric(self): + # For a non-numeric string the ValueError path fires correctly + with pytest.raises(ArgumentTypeError) as exc: + overheadtype("abc") + assert len(str(exc.value)) > 0 + + +# --------------------------------------------------------------------------- +# systemidtype +# --------------------------------------------------------------------------- +# +# systemidtype(arg) validates MBR partition type ID in range 0x01..0xFF. +# Returns the original arg string unchanged. +# +# Known bugs: +# leading '+' accepted ("+0x82" -> int("+0x82", 16)=130, in range) +# missing "0x" prefix accepted ("82" -> int("82", 16)=130, in range) +# --------------------------------------------------------------------------- + +class TestSystemidtype: + + # --- valid inputs ------------------------------------------------------- + + @pytest.mark.parametrize("arg", [ + "0x1", "0x01", "0x82", "0x83", "0xFE", "0xFF", "0x0b", "0x0B", + ]) + def test_valid_values_returned_unchanged(self, arg): + assert systemidtype(arg) == arg + + def test_minimum_valid_value(self): + assert systemidtype("0x1") == "0x1" + + def test_maximum_valid_value(self): + assert systemidtype("0xFF") == "0xFF" + + def test_uppercase_hex_valid(self): + assert systemidtype("0xFF") == "0xFF" + + def test_lowercase_hex_valid(self): + assert systemidtype("0xff") == "0xff" + + def test_mixed_case_valid(self): + assert systemidtype("0xFf") == "0xFf" + + # --- boundary: out of range --------------------------------------------- + + def test_zero_rejected(self): + with pytest.raises(ArgumentTypeError): + systemidtype("0x0") + + def test_zero_padded_rejected(self): + with pytest.raises(ArgumentTypeError): + systemidtype("0x00") + + def test_above_0xff_rejected(self): + with pytest.raises(ArgumentTypeError): + systemidtype("0x100") + + def test_well_above_range_rejected(self): + with pytest.raises(ArgumentTypeError): + systemidtype("0xFFFF") + + # --- invalid format ----------------------------------------------------- + + @pytest.mark.parametrize("bad", [ + "0x", # prefix with no digit + "0xGG", # invalid hex digit + "abc", # not hex + "0b1010", # binary prefix + "130", # decimal that happens to equal a valid value + ]) + def test_invalid_format_raises(self, bad): + with pytest.raises((ArgumentTypeError, ValueError)): + systemidtype(bad) + + def test_empty_string_raises(self): + with pytest.raises((ArgumentTypeError, ValueError)): + systemidtype("") + + def test_whitespace_only_raises(self): + with pytest.raises((ArgumentTypeError, ValueError)): + systemidtype(" ") + + def test_none_raises(self): + with pytest.raises((ArgumentTypeError, TypeError, AttributeError)): + systemidtype(None) + + def test_very_long_string_raises(self): + with pytest.raises((ArgumentTypeError, ValueError)): + systemidtype("0x" + "F" * 10000) + + def test_embedded_whitespace_raises(self): + with pytest.raises((ArgumentTypeError, ValueError)): + systemidtype("0x 82") + + # --- known bugs --------------------------------------------------------- + + @pytest.mark.xfail(reason="systemidtype accepts hex without '0x' prefix; '82' -> int('82',16)=130") + def test_no_prefix_rejected(self): + with pytest.raises(ArgumentTypeError): + systemidtype("82") + + @pytest.mark.xfail(reason="leading '+' accepted; '+0x82' -> int('+0x82',16)=130") + def test_leading_plus_rejected(self): + with pytest.raises(ArgumentTypeError): + systemidtype("+0x82") diff --git a/tests/unit/test_misc.py b/tests/unit/test_misc.py new file mode 100644 index 000000000000..46b075dd78c8 --- /dev/null +++ b/tests/unit/test_misc.py @@ -0,0 +1,128 @@ +""" +Unit tests for wic/misc.py. + +runtool() is probed with every boundary input class: empty list, +empty string, None, a single-element list, and strings that contain +shell metacharacters. The WicError path for a missing command is +also exercised. +""" + +import sys +import pytest +from pathlib import Path + +_SRC = Path(__file__).resolve().parent.parent.parent / "src" +if str(_SRC) not in sys.path: + sys.path.insert(0, str(_SRC)) + +from wic import WicError +from wic.misc import runtool + + +class TestRuntool: + + # --- valid inputs ------------------------------------------------------- + + def test_list_form_true_command(self): + rc, out = runtool(["true"]) + assert rc == 0 + + def test_list_form_echo(self): + rc, out = runtool(["echo", "hello"]) + assert rc == 0 + assert "hello" in out + + def test_string_form_echo(self): + rc, out = runtool("echo hello") + assert rc == 0 + assert "hello" in out + + def test_string_form_false(self): + rc, out = runtool("false") + assert rc != 0 + + def test_list_form_with_args(self): + rc, out = runtool(["printf", "%s", "test"]) + assert rc == 0 + assert "test" in out + + def test_output_is_string(self): + rc, out = runtool(["echo", "x"]) + assert isinstance(out, str) + + def test_stderr_merged_into_out(self): + # echo to stderr via shell form; stderr is merged into output + rc, out = runtool("echo err >&2") + assert "err" in out + + # --- missing command ---------------------------------------------------- + + def test_nonexistent_command_raises_wicerror(self): + with pytest.raises(WicError, match="Cannot run command"): + runtool(["totally_nonexistent_command_xyz"]) + + def test_nonexistent_string_command_returns_nonzero(self): + # String form uses shell=True; the shell (sh) is found so no WicError + # is raised. Instead rc=127 and the "not found" message is in output. + rc, out = runtool("totally_nonexistent_command_xyz") + assert rc != 0 + assert "not found" in out.lower() or "No such" in out + + # --- empty / degenerate inputs ------------------------------------ + + @pytest.mark.xfail(reason="runtool([]) raises IndexError on cmdln_or_args[0]; " + "should raise WicError or ArgumentTypeError instead") + def test_empty_list_raises_wicerror(self): + with pytest.raises(WicError): + runtool([]) + + def test_empty_list_raises_something(self): + # Whatever it raises, it must not silently succeed + with pytest.raises((IndexError, WicError, ValueError, TypeError)): + runtool([]) + + @pytest.mark.xfail(reason="runtool('') calls shlex.split('') -> [] then [0] " + "raises IndexError; should raise WicError instead") + def test_empty_string_raises_wicerror(self): + with pytest.raises(WicError): + runtool("") + + def test_empty_string_raises_something(self): + with pytest.raises((IndexError, WicError, ValueError, TypeError)): + runtool("") + + @pytest.mark.filterwarnings("ignore::DeprecationWarning") + def test_none_raises(self): + # shlex.split(None) is deprecated (Python 3.12+ raises TypeError; + # 3.10 reads from stdin which OSError under pytest capture). + # The DeprecationWarning is expected and suppressed here. + with pytest.raises((TypeError, AttributeError, OSError)): + runtool(None) + + def test_whitespace_only_string(self): + # shlex.split(" ") -> [] -> IndexError + with pytest.raises((IndexError, WicError, ValueError)): + runtool(" ") + + # --- single-element list ------------------------------------------------ + + def test_single_element_list_valid_command(self): + rc, out = runtool(["true"]) + assert rc == 0 + + def test_single_element_list_missing(self): + with pytest.raises(WicError): + runtool(["missing_xyz"]) + + # --- special characters in string form ---------------------------------- + + def test_semicolon_in_shell_form(self): + # Shell form allows semicolons; this should run two commands + rc, out = runtool("echo a; echo b") + assert rc == 0 + + def test_large_output(self): + # Command that produces a lot of output + rc, out = runtool(["seq", "1", "1000"]) + assert rc == 0 + assert "1000" in out diff --git a/tests/unit/test_misc_vars.py b/tests/unit/test_misc_vars.py new file mode 100644 index 000000000000..7446057b33ef --- /dev/null +++ b/tests/unit/test_misc_vars.py @@ -0,0 +1,182 @@ +r""" +Unit tests for the variable/helper logic in wic/misc.py. + +These cover the pure-logic parts that do not need bitbake: + - BitbakeVars._parse_line: the key=value tokeniser + - BitbakeVars.get_var: parsing .env files from vars_dir, + caching, the single-image default collapse, cache=False eviction, + and the missing-file path + - find_executable: NATIVE_RECIPES mapping and ASSUME_PROVIDED short + circuit + - get_bitbake_var: the WIC_SECTOR_SIZE environment shim + - NATIVE_RECIPES: the executable -> recipe table used for error hints + +runtool() itself is covered in test_misc.py. Where wic does not yet +behave correctly, the test is marked xfail with a short reason. +""" + +import os +import sys +from pathlib import Path + +_SRC = Path(__file__).resolve().parent.parent.parent / "src" +if str(_SRC) not in sys.path: + sys.path.insert(0, str(_SRC)) + +import wic.misc as misc +from wic.misc import BitbakeVars, NATIVE_RECIPES, find_executable, get_bitbake_var + + +# --------------------------------------------------------------------------- +# _parse_line: key=value tokeniser +# --------------------------------------------------------------------------- + +class TestParseLine: + + def _parse(self, line): + bv = BitbakeVars() + bv._parse_line(line, "img") + return dict(bv.get("img", {})) + + def test_simple_assignment(self): + assert self._parse("FOO=bar") == {"FOO": "bar"} + + def test_quotes_are_stripped(self): + assert self._parse('FOO="bar"') == {"FOO": "bar"} + + def test_value_with_spaces_kept(self): + assert self._parse('FOO="a b c"') == {"FOO": "a b c"} + + def test_empty_quoted_value(self): + assert self._parse('FOO=""') == {"FOO": ""} + + def test_first_equals_splits_only(self): + # A value containing '=' keeps everything after the first '='. + assert self._parse("FOO=a=b=c") == {"FOO": "a=b=c"} + + def test_line_without_equals_ignored(self): + assert self._parse("no equals here") == {} + + def test_whitespace_only_ignored(self): + assert self._parse(" ") == {} + + def test_key_with_invalid_chars_ignored(self): + # The matcher only accepts [A-Za-z0-9-_+./~] keys. + assert self._parse("WEIRD!@#=val") == {} + + def test_dotted_and_plus_keys_accepted(self): + assert self._parse("A.B+C=1") == {"A.B+C": "1"} + + +# --------------------------------------------------------------------------- +# get_var via vars_dir (.env file parsing) + caching +# --------------------------------------------------------------------------- + +class TestGetVarFromEnvDir: + + def _bv(self, tmp_path, image="core-image", body="ROOTFS_SIZE=\"123\"\n"): + (tmp_path / ("%s.env" % image)).write_text(body) + bv = BitbakeVars() + bv.vars_dir = str(tmp_path) + return bv + + def test_reads_value_from_env_file(self, tmp_path): + bv = self._bv(tmp_path) + assert bv.get_var("ROOTFS_SIZE", "core-image") == "123" + + def test_missing_var_returns_none(self, tmp_path): + bv = self._bv(tmp_path) + assert bv.get_var("NOPE", "core-image") is None + + def test_single_image_becomes_default(self, tmp_path): + bv = self._bv(tmp_path, body="IMAGE_FSTYPES=\"wic\"\n") + # prime the cache for the named image + assert bv.get_var("IMAGE_FSTYPES", "core-image") == "wic" + # with a single image cached, lookups with no image use it + assert bv.get_var("IMAGE_FSTYPES") == "wic" + + def test_missing_env_file_returns_none(self, tmp_path): + bv = BitbakeVars() + bv.vars_dir = str(tmp_path) + assert bv.get_var("X", "does-not-exist") is None + + def test_cache_false_does_not_retain_image(self, tmp_path): + bv = self._bv(tmp_path) + bv.get_var("ROOTFS_SIZE", "core-image", cache=False) + # cache=False must evict the image so nothing is retained + assert "core-image" not in bv + + def test_cache_true_retains_image(self, tmp_path): + bv = self._bv(tmp_path) + bv.get_var("ROOTFS_SIZE", "core-image", cache=True) + assert "core-image" in bv + + +# --------------------------------------------------------------------------- +# find_executable +# --------------------------------------------------------------------------- + +class TestFindExecutable: + + def test_finds_real_tool(self, monkeypatch): + monkeypatch.setattr(misc, "get_bitbake_var", lambda *a, **k: None) + assert find_executable("true", os.environ.get("PATH", "")) + + def test_missing_tool_returns_falsey(self, monkeypatch): + monkeypatch.setattr(misc, "get_bitbake_var", lambda *a, **k: None) + assert not find_executable("totally_missing_xyz", os.environ.get("PATH", "")) + + def test_assume_provided_short_circuits(self, monkeypatch): + # mkdosfs maps to the dosfstools recipe; if dosfstools-native is + # ASSUME_PROVIDED, find_executable returns True without a PATH hit. + monkeypatch.setattr(misc, "get_bitbake_var", + lambda *a, **k: "dosfstools-native") + assert find_executable("mkdosfs", "") is True + + def test_empty_path_finds_nothing(self, monkeypatch): + monkeypatch.setattr(misc, "get_bitbake_var", lambda *a, **k: None) + assert not find_executable("true", "") + + +# --------------------------------------------------------------------------- +# NATIVE_RECIPES mapping +# --------------------------------------------------------------------------- + +class TestNativeRecipes: + + def test_known_tool_maps_to_recipe(self): + assert NATIVE_RECIPES["mkdosfs"] == "dosfstools" + assert NATIVE_RECIPES["parted"] == "parted" + + def test_mtools_family_maps_consistently(self): + for tool in ("mcopy", "mdel", "mdir", "mmd"): + assert NATIVE_RECIPES[tool] == "mtools" + + +# --------------------------------------------------------------------------- +# WIC_SECTOR_SIZE environment shim +# --------------------------------------------------------------------------- + +class TestWicSectorSizeShim: + + def test_env_value_is_returned(self, monkeypatch): + monkeypatch.setenv("WIC_SECTOR_SIZE", "4096") + assert get_bitbake_var("WIC_SECTOR_SIZE") == "4096" + + def test_unset_falls_through_to_lookup(self, monkeypatch): + monkeypatch.delenv("WIC_SECTOR_SIZE", raising=False) + # With no env override and a stubbed singleton, the value comes + # from the normal variable lookup path (here: None). + monkeypatch.setattr(misc.BB_VARS, "get_var", lambda *a, **k: None) + assert get_bitbake_var("WIC_SECTOR_SIZE") is None + + def test_junk_value_returned_verbatim(self, monkeypatch): + """A non-numeric WIC_SECTOR_SIZE is handed back unchanged. + + The shim does no validation, so "abc" flows downstream to code + that will later do int(...) on it. A clear early rejection would + be friendlier than a deep ValueError, but document the current + pass-through behaviour. + """ + monkeypatch.setenv("WIC_SECTOR_SIZE", "abc") + assert get_bitbake_var("WIC_SECTOR_SIZE") == "abc" diff --git a/tests/unit/test_oe_path.py b/tests/unit/test_oe_path.py new file mode 100644 index 000000000000..afb09251626b --- /dev/null +++ b/tests/unit/test_oe_path.py @@ -0,0 +1,465 @@ +""" +Unit tests for wic/oe/path.py. + +Every function is probed with: empty string, None, non-existent paths, +paths with special characters, boundary lengths, and wrong argument types. +Where wic does not yet behave correctly, the test is marked xfail with a +short reason describing the bug. +""" + +import os +import sys +import pytest +from pathlib import Path + +_SRC = Path(__file__).resolve().parent.parent.parent / "src" +if str(_SRC) not in sys.path: + sys.path.insert(0, str(_SRC)) + +# oe.path imports wic.bb which needs to be importable +import wic.bb # noqa: F401 (ensure the stub is loaded) +from wic.oe.path import ( + join, relative, make_relative_symlink, symlink, + is_path_parent, remove, canonicalize, find, which_wild, realpath, +) + + +# --------------------------------------------------------------------------- +# join() — like os.path.join but ignores absolute RHS +# --------------------------------------------------------------------------- + +class TestJoin: + + def test_simple_two_paths(self): + assert join("a", "b") == "a/b" + + def test_three_paths(self): + assert join("a", "b", "c") == "a/b/c" + + def test_absolute_rhs_not_special(self): + # os.path.join("a", "/b") == "/b", but oe.path.join treats / as relative + result = join("a", "/b") + assert result == "a/b" + + def test_double_slash_normalised(self): + assert join("a//", "b") == "a/b" + + def test_dotdot_normalised(self): + assert join("a", "..", "b") == "b" + + def test_dot_normalised(self): + assert join("a", ".", "b") == "a/b" + + def test_empty_first(self): + # join("", "b") -> normpath("/".join(["","b"])) -> normpath("/b") -> "/b" + # (absolute RHS not special because it's concatenated, not os.path.join) + result = join("", "b") + # document the actual behaviour; do not assert a wrong expectation + assert isinstance(result, str) and len(result) > 0 + + def test_empty_both(self): + result = join("", "") + assert isinstance(result, str) + + def test_single_path(self): + result = join("a") + assert result == "a" + + def test_very_long_path(self): + segment = "x" * 255 + result = join(segment, segment) + assert segment in result + + +# --------------------------------------------------------------------------- +# relative() — thin wrapper around os.path.relpath +# --------------------------------------------------------------------------- + +class TestRelative: + + def test_child_path(self): + assert relative("/tmp", "/tmp/foo/bar") == "foo/bar" + + def test_sibling(self): + assert relative("/usr/bin", "/usr/lib") == "../lib" + + def test_same_path(self): + assert relative("/tmp", "/tmp") == "." + + def test_root_to_child(self): + result = relative("/", "/usr/bin") + assert result == "usr/bin" + + def test_empty_both(self): + # os.path.relpath("", "") raises ValueError — document this behaviour + with pytest.raises(ValueError): + relative("", "") + + +# --------------------------------------------------------------------------- +# is_path_parent() +# --------------------------------------------------------------------------- + +class TestIsPathParent: + + def test_direct_child(self): + assert is_path_parent("/usr", "/usr/bin") is True + + def test_deep_child(self): + assert is_path_parent("/usr", "/usr/share/doc/readme") is True + + def test_not_parent(self): + assert is_path_parent("/usr", "/tmp") is False + + def test_same_path(self): + # /usr is not a parent of /usr (a parent is strictly above) + # os.path.abspath adds trailing sep; /usr/ startswith /usr/ → True + # Document actual behaviour. + result = is_path_parent("/usr", "/usr") + assert isinstance(result, bool) + + def test_child_cannot_be_parent(self): + assert is_path_parent("/usr/bin", "/usr") is False + + def test_no_paths_returns_false(self): + assert is_path_parent("/usr") is False + + def test_multiple_paths_all_must_match(self): + assert is_path_parent("/usr", "/usr/bin", "/usr/lib") is True + + def test_multiple_paths_one_mismatch(self): + assert is_path_parent("/usr", "/usr/bin", "/tmp") is False + + def test_empty_possible_parent(self): + # empty string → os.path.abspath("") → cwd; cwd is parent of /tmp is False + result = is_path_parent("", "/tmp") + assert isinstance(result, bool) + + def test_none_possible_parent_raises(self): + with pytest.raises((TypeError, AttributeError)): + is_path_parent(None, "/usr/bin") + + def test_none_child_raises(self): + with pytest.raises((TypeError, AttributeError)): + is_path_parent("/usr", None) + + +# --------------------------------------------------------------------------- +# symlink() +# --------------------------------------------------------------------------- + +class TestSymlink: + + def test_creates_symlink(self, tmp_path): + src = tmp_path / "target.txt" + src.write_text("content") + dst = tmp_path / "link" + symlink(str(src), str(dst)) + assert os.path.islink(str(dst)) + assert os.readlink(str(dst)) == str(src) + + def test_existing_matching_symlink_is_idempotent(self, tmp_path): + src = tmp_path / "target.txt" + src.write_text("content") + dst = tmp_path / "link" + symlink(str(src), str(dst)) + # second call with same src/dst should not raise + symlink(str(src), str(dst)) + + def test_existing_different_symlink_raises(self, tmp_path): + src1 = tmp_path / "t1.txt" + src2 = tmp_path / "t2.txt" + src1.write_text("a") + src2.write_text("b") + dst = tmp_path / "link" + symlink(str(src1), str(dst)) + with pytest.raises(OSError): + symlink(str(src2), str(dst)) + + def test_regular_file_destination_raises_oserror(self, tmp_path): + # when destination is a regular file, os.readlink() inside the + # EEXIST handler raises EINVAL (invalid argument) which propagates as + # an unhandled OSError. The observable behaviour (OSError raised) is + # accidentally correct, but for the wrong reason. + src = tmp_path / "source" + src.write_text("src") + dst = tmp_path / "destination" + dst.write_text("already a regular file, not a symlink") + with pytest.raises(OSError): + symlink(str(src), str(dst)) + + @pytest.mark.xfail(reason="symlink(force=True) calls remove(destination) which " + "uses glob.glob(); a destination with glob metacharacters " + "can delete unrelated files") + def test_force_with_glob_metachar_destination(self, tmp_path): + src = tmp_path / "source" + src.write_text("src") + # Create a file that contains '[' in its name — glob metacharacter + keep = tmp_path / "keep[1].txt" + keep.write_text("do not delete me") + dst = tmp_path / "keep[1].txt" # same name as our file + # force=True should only remove the specific destination path, + # not other files whose names happen to match the glob pattern. + symlink(str(src), str(dst), force=True) + assert keep.exists(), "force=True glob-expanded the destination and deleted unrelated file" + + def test_force_true_overwrites_existing_symlink(self, tmp_path): + src1 = tmp_path / "t1" + src2 = tmp_path / "t2" + src1.write_text("a") + src2.write_text("b") + dst = tmp_path / "link" + symlink(str(src1), str(dst)) + symlink(str(src2), str(dst), force=True) + assert os.readlink(str(dst)) == str(src2) + + def test_none_source_raises(self, tmp_path): + dst = tmp_path / "link" + with pytest.raises((TypeError, AttributeError, OSError)): + symlink(None, str(dst)) + + def test_none_destination_raises(self, tmp_path): + src = tmp_path / "src" + src.write_text("x") + with pytest.raises((TypeError, AttributeError, OSError)): + symlink(str(src), None) + + +# --------------------------------------------------------------------------- +# make_relative_symlink() +# --------------------------------------------------------------------------- + +class TestMakeRelativeSymlink: + + def test_non_symlink_is_ignored(self, tmp_path): + regular = tmp_path / "file.txt" + regular.write_text("content") + # Should return without doing anything + make_relative_symlink(str(regular)) + assert regular.is_file() + + def test_already_relative_symlink_is_unchanged(self, tmp_path): + target = tmp_path / "target.txt" + target.write_text("content") + link = tmp_path / "link" + os.symlink("target.txt", str(link)) # relative already + make_relative_symlink(str(link)) + assert os.readlink(str(link)) == "target.txt" + + def test_absolute_symlink_converted_to_relative(self, tmp_path): + target = tmp_path / "target.txt" + target.write_text("content") + link = tmp_path / "link" + os.symlink(str(target), str(link)) # absolute + assert os.path.isabs(os.readlink(str(link))) + make_relative_symlink(str(link)) + # After conversion, link should be relative + new_link = os.readlink(str(link)) + assert not os.path.isabs(new_link) + # And still resolve correctly + assert os.path.exists(str(link)) + + @pytest.mark.xfail(reason="make_relative_symlink produces a trailing '/' " + "when the target is the link's immediate parent directory") + def test_no_trailing_slash_in_result(self, tmp_path): + # Create: tmp_path/dir/link -> tmp_path (link points to grandparent) + subdir = tmp_path / "dir" + subdir.mkdir() + link = subdir / "link" + os.symlink(str(tmp_path), str(link)) # absolute, points to parent of subdir + make_relative_symlink(str(link)) + new_link = os.readlink(str(link)) + assert not new_link.endswith("/"), "trailing slash in symlink target: %r" % new_link + + +# --------------------------------------------------------------------------- +# remove() — boundary cases +# --------------------------------------------------------------------------- + +class TestRemove: + + def test_removes_existing_file(self, tmp_path): + f = tmp_path / "f.txt" + f.write_text("x") + remove(str(f)) + assert not f.exists() + + def test_missing_file_is_silent(self, tmp_path): + remove(str(tmp_path / "nonexistent.txt")) # must not raise + + def test_removes_directory_recursively(self, tmp_path): + d = tmp_path / "subdir" + d.mkdir() + (d / "child").write_text("x") + remove(str(d)) + assert not d.exists() + + def test_empty_pattern_does_not_crash(self): + remove("") # glob.glob("") returns [] → no-op + + def test_none_raises(self): + with pytest.raises((TypeError, AttributeError)): + remove(None) + + +# --------------------------------------------------------------------------- +# canonicalize() +# --------------------------------------------------------------------------- + +class TestCanonicalize: + + def test_real_path_returned(self, tmp_path): + result = canonicalize(str(tmp_path)) + assert os.path.isabs(result) + + def test_dollar_var_skipped(self): + # Paths containing '$' are skipped per the docstring + result = canonicalize("$SOME_VAR") + assert result == "" + + def test_dollar_and_real_path(self, tmp_path): + # canonicalize("$VAR,/path") drops the $VAR token entirely + # (no empty placeholder) and returns only the realpath of /path + result = canonicalize("$VAR," + str(tmp_path)) + # result has no leading comma; it is just the real path + assert os.path.isabs(result) + assert str(tmp_path) in result + + @pytest.mark.xfail(reason="canonicalize(None) returns os.getcwd() instead of " + "empty string; 'None or \\'\\''.split(',') gives [''] which " + "has no '$', so os.path.realpath('') is returned") + def test_none_returns_empty_string(self): + result = canonicalize(None) + assert result == "", "got %r instead of empty string" % result + + @pytest.mark.xfail(reason="canonicalize('') returns os.getcwd() for the same " + "reason — os.path.realpath('') == os.getcwd()") + def test_empty_string_returns_empty_string(self): + result = canonicalize("") + assert result == "", "got %r" % result + + def test_trailing_slash_preserved(self, tmp_path): + result = canonicalize(str(tmp_path) + "/") + assert result.endswith("/") + + def test_multiple_paths(self, tmp_path): + p1 = str(tmp_path) + p2 = str(tmp_path) + result = canonicalize("%s,%s" % (p1, p2)) + assert result.count(",") == 1 + + +# --------------------------------------------------------------------------- +# find() — basic iteration +# --------------------------------------------------------------------------- + +class TestFind: + + def test_finds_files(self, tmp_path): + (tmp_path / "a.txt").write_text("a") + (tmp_path / "b.txt").write_text("b") + results = list(find(str(tmp_path))) + assert len(results) == 2 + assert all(os.path.isabs(p) for p in results) + + def test_empty_directory_returns_nothing(self, tmp_path): + assert list(find(str(tmp_path))) == [] + + def test_nested_files_found(self, tmp_path): + sub = tmp_path / "sub" + sub.mkdir() + (sub / "nested.txt").write_text("x") + results = list(find(str(tmp_path))) + assert len(results) == 1 + assert "nested.txt" in results[0] + + def test_nonexistent_directory_returns_nothing(self, tmp_path): + # os.walk on a missing path returns empty iterator + results = list(find(str(tmp_path / "no_such_dir"))) + assert results == [] + + +# --------------------------------------------------------------------------- +# which_wild() — basic probing +# --------------------------------------------------------------------------- + +class TestWhichWild: + + def test_finds_existing_executable(self): + results = which_wild("python3") + # python3 must be on PATH for the test suite to run at all + assert len(results) >= 1 + assert all(os.path.isabs(p) for p in results) + + def test_missing_tool_returns_empty(self): + results = which_wild("totally_nonexistent_tool_xyz") + assert results == [] + + def test_empty_pattern_behaviour(self): + # which_wild("") forms os.path.join(dir, "") = dir for each PATH element, + # then glob.glob(dir) returns the directory itself if it exists. + # The result is therefore NOT guaranteed to be empty. + results = which_wild("") + # Just check we get a list back without crashing + assert isinstance(results, list) + + def test_none_pattern_raises(self): + with pytest.raises((TypeError, AttributeError)): + which_wild(None) + + def test_wildcard_pattern(self): + # "python*" should find at least python3 + results = which_wild("python*") + assert any("python" in p for p in results) + + def test_reverse_flag(self): + # reverse=True reverses PATH search order. Because deduplication + # tracks by relative name, the set of returned paths can legitimately + # differ between forward and reverse when python3 exists in multiple + # PATH entries. Just verify both return non-empty results. + fwd = which_wild("python3") + rev = which_wild("python3", reverse=True) + assert len(fwd) >= 1 and len(rev) >= 1 + + +# --------------------------------------------------------------------------- +# realpath() error path +# --------------------------------------------------------------------------- + +class TestRealpath: + """__realpath()'s final stanza is: + + try: + is_dir = os.path.isdir(file) + except: + is_dir = false # <-- lowercase: not a Python name + + Python has False, not false. Any exception from os.path.isdir + (EACCES on a parent, EOVERFLOW on a truncated stat, ...) reaches + the handler and raises NameError instead of falling back to "not a + directory". The intended fallback value is False. + """ + + def test_isdir_oserror_does_not_raise_nameerror(self, tmp_path, monkeypatch): + root = str(tmp_path) + sub = tmp_path / "d" + sub.mkdir() + + def boom(_path): + raise OSError("synthetic stat failure") + + # use_physdir=False routes through __realpath(), which contains + # the buggy handler; force os.path.isdir to raise so the handler + # is exercised. + monkeypatch.setattr(os.path, "isdir", boom) + try: + result = realpath(str(sub), root, use_physdir=False) + except NameError: + pytest.xfail( + "__realpath bare-except handler assigns the undefined " + "name `false`; raises NameError instead of falling back to " + "is_dir=False") + # Correct behaviour once `false` becomes False: realpath returns + # the resolved path (the is_dir result is computed but not part of + # realpath's return value). + assert result == str(sub) diff --git a/tests/unit/test_partition.py b/tests/unit/test_partition.py new file mode 100644 index 000000000000..455db54a46de --- /dev/null +++ b/tests/unit/test_partition.py @@ -0,0 +1,598 @@ +r""" +Unit tests for wic.partition.Partition. + +Tests assert CORRECT behaviour, and xfail when the code does not yet +meet that behaviour. + +Key known bugs targeted here: + - get_rootfs_size: fixed_size < extra_partition_space → negative + allowed size; misleading error message. + - prepare_empty_partition_btrfs/swap use self.size (0 for --fixed-size) + instead of self.fs_size; caught via the fs_size property tests. + - get_rootfs_size: extra_filesystem_space 0 overridden to default. + - _mkfs_ext_extraopts / _mkdosfs_extraopts: long-form options like + --block-size=4096 bypass the short-form -b/-S conflict detection. +""" +import sys +import types +from pathlib import Path + +import pytest + +_WIC_SRC = Path(__file__).resolve().parent.parent.parent / "src" +if str(_WIC_SRC) not in sys.path: + sys.path.insert(0, str(_WIC_SRC)) + +from wic import WicError +from wic.partition import Partition + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _make_partition(**kwargs): + """Construct a Partition with sensible defaults; override via kwargs. + + Partition.__init__ takes an argparse.Namespace-like object. We build + one with every attribute the constructor touches so each test need + only specify the attributes relevant to it. + """ + defaults = dict( + active=False, + align=1024, + disk="sda", + extra_filesystem_space=0, + extra_partition_space=0, + exclude_path=None, + include_path=None, + change_directory=None, + fsopts=None, + fspassno=0, + fstype="ext4", + label=None, + use_label=False, + mkfs_extraopts="", + mountpoint="/", + no_table=False, + offset=0, + overhead_factor=1.3, + part_name=None, + part_type=None, + rootfs_dir=None, + size=0, + fixed_size=0, + source=None, + sourceparams=None, + system_id=None, + use_uuid=False, + uuid=None, + fsuuid=None, + type="primary", + no_fstab_update=False, + hidden=False, + mbr=False, + ) + defaults.update(kwargs) + args = types.SimpleNamespace(**defaults) + return Partition(args, lineno=1) + + +# --------------------------------------------------------------------------- +# _mkdosfs_extraopts +# --------------------------------------------------------------------------- + +class TestMkdosfsExtraopts: + """_mkdosfs_extraopts() appends -S sector_size unless -S is already present. + + the presence check only matches '-S' and '-S' short forms; + '--logical-sector-size=N' or '--sector-size N' bypass it. + """ + + def _part(self, extraopts="", sector_size=512): + p = _make_partition(mkfs_extraopts=extraopts, fstype="vfat") + p.sector_size = sector_size + return p + + def test_empty_extraopts_appends_sector_size(self): + result = self._part(sector_size=4096)._mkdosfs_extraopts() + assert "-S 4096" in result + + def test_s_flag_present_no_duplicate(self): + result = self._part("-S 1024", sector_size=4096)._mkdosfs_extraopts() + assert result.count("-S") == 1 + + def test_s_flag_compact_no_duplicate(self): + result = self._part("-S1024", sector_size=4096)._mkdosfs_extraopts() + assert result.count("-S") == 1 + + def test_512_sector_still_appends(self): + # 512 is the default but we still append it if not present + result = self._part(sector_size=512)._mkdosfs_extraopts() + assert "-S 512" in result + + @pytest.mark.parametrize("sector_size", [512, 1024, 4096, 8192, 3000]) + def test_various_sector_sizes_appended(self, sector_size): + result = self._part(sector_size=sector_size)._mkdosfs_extraopts() + assert "-S %d" % sector_size in result + + def test_other_flags_preserved(self): + result = self._part("-F 32", sector_size=512)._mkdosfs_extraopts() + assert "-F 32" in result + assert "-S 512" in result + + +# --------------------------------------------------------------------------- +# _mkfs_ext_extraopts +# --------------------------------------------------------------------------- + +class TestMkfsExtExtraopts: + """_mkfs_ext_extraopts(base_opts) applies block-size when sector_size != 512. + + only '-b' and '-b' short forms are detected; the long + form '--block-size=N' bypasses the check, producing duplicate arguments. + """ + + def _part(self, extraopts="", sector_size=512): + p = _make_partition(mkfs_extraopts=extraopts) + p.sector_size = sector_size + return p + + def test_512_sector_no_b_appended(self): + result = self._part(sector_size=512)._mkfs_ext_extraopts("-F -i 8192") + assert "-b" not in result + + def test_4096_sector_appends_b(self): + result = self._part(sector_size=4096)._mkfs_ext_extraopts("-F -i 8192") + assert "-b 4096" in result + + def test_b_flag_present_no_duplicate(self): + result = self._part("-b 4096", sector_size=4096)._mkfs_ext_extraopts("") + assert result.count("-b") == 1 + + def test_b_compact_present_no_duplicate(self): + result = self._part("-b4096", sector_size=4096)._mkfs_ext_extraopts("") + assert result.count("-b") == 1 + + def test_user_extraopts_replace_base_opts(self): + """User-supplied extraopts completely replace base_opts.""" + result = self._part("-F", sector_size=512)._mkfs_ext_extraopts("-F -i 8192") + # user said '-F'; base_opts '-F -i 8192' should be replaced + assert result == "-F" + + def test_base_opts_used_when_no_user_opts(self): + result = self._part("", sector_size=512)._mkfs_ext_extraopts("-F -i 8192") + assert "-F" in result + assert "-i 8192" in result + + def test_long_form_block_size_bypasses_detection(self): + """--block-size=4096 is not detected; wic adds -b 4096 too.""" + p = self._part("--block-size=4096", sector_size=4096) + result = p._mkfs_ext_extraopts("") + # Correct: only one block-size specification should be in the command. + # Current (broken): both --block-size=4096 AND -b 4096 appear. + has_duplicate = "--block-size=4096" in result and "-b 4096" in result + if has_duplicate: + pytest.xfail( + "--block-size=4096 bypasses -b detection; " + "both long and short forms appear in the mkfs command") + assert result.count("4096") == 1 + + +# --------------------------------------------------------------------------- +# get_extra_block_count +# --------------------------------------------------------------------------- + +class TestGetExtraBlockCount: + """get_extra_block_count(current_blocks) returns blocks to add to reach self.size.""" + + def _part(self, size): + return _make_partition(size=size) + + def test_size_zero_returns_zero(self): + assert self._part(0).get_extra_block_count(500) == 0 + + def test_size_larger_than_current_returns_difference(self): + assert self._part(1000).get_extra_block_count(600) == 400 + + def test_size_equal_to_current_returns_zero(self): + assert self._part(1000).get_extra_block_count(1000) == 0 + + def test_size_smaller_than_current_returns_zero_not_negative(self): + """When the partition already exceeds the requested size, return 0.""" + assert self._part(500).get_extra_block_count(1000) == 0 + + def test_current_blocks_zero(self): + assert self._part(1000).get_extra_block_count(0) == 1000 + + def test_current_blocks_zero_size_zero(self): + assert self._part(0).get_extra_block_count(0) == 0 + + @pytest.mark.parametrize("size,current,expected", [ + (1, 0, 1), + (1, 1, 0), + (1, 2, 0), + (2**31, 0, 2**31), + (2**31-1, 2**31, 0), + (100, 99, 1), + (100, 101, 0), + ]) + def test_boundary_values(self, size, current, expected): + assert self._part(size).get_extra_block_count(current) == expected + + def test_large_current_blocks_no_overflow(self): + """Very large current_blocks should not cause integer overflow.""" + p = self._part(100) + result = p.get_extra_block_count(10 ** 12) + assert result == 0 + + +# --------------------------------------------------------------------------- +# get_rootfs_size +# --------------------------------------------------------------------------- + +class TestGetRootfsSize: + """get_rootfs_size(actual_rootfs_size) — the central sizing function. + + Known bugs: + fixed_size < extra_partition_space → negative allowed size and + confusing error message. + extra_filesystem_space == 0 treated as falsy; replaced by default. + """ + + # --- fixed-size path --- + + def test_fixed_size_actual_fits(self): + p = _make_partition(fixed_size=10000, extra_partition_space=0) + assert p.get_rootfs_size(5000) == 10000 + + def test_fixed_size_with_extra_partition_space(self): + p = _make_partition(fixed_size=10000, extra_partition_space=2000) + assert p.get_rootfs_size(5000) == 8000 + + def test_fixed_size_actual_equals_allowed(self): + p = _make_partition(fixed_size=10000, extra_partition_space=0) + assert p.get_rootfs_size(10000) == 10000 + + def test_fixed_size_actual_exceeds_raises(self): + p = _make_partition(fixed_size=1000, extra_partition_space=0) + with pytest.raises(WicError): + p.get_rootfs_size(2000) + + def test_fixed_size_error_message_positive(self): + """fixed_size=10M < extra_partition_space=20M → negative allowed. + The error message should name both conflicting options, not show a + negative size which is confusing and meaningless. + """ + p = _make_partition(fixed_size=10000, extra_partition_space=20000) + try: + p.get_rootfs_size(1) + except WicError as e: + msg = str(e) + # Correct: error names the conflicting options, no negative number. + has_negative = any(f"-{n}" in msg or "-%d" % n in msg + for n in range(1, 100000)) + if has_negative or "allowed size" in msg and "-" in msg: + pytest.xfail( + "fixed_size < extra_partition_space produces a " + "negative 'allowed size' in the error message instead of " + "a clear validation error naming the conflict") + else: + pytest.fail("Expected WicError when fixed_size < extra_partition_space") + + def test_fixed_size_zero_actual_fits(self): + p = _make_partition(fixed_size=5000, extra_partition_space=0) + assert p.get_rootfs_size(0) == 5000 + + # --- non-fixed path --- + + def test_no_fixed_size_default(self): + p = _make_partition(size=0, fixed_size=0, overhead_factor=1.0, + extra_filesystem_space=0, extra_partition_space=0) + result = p.get_rootfs_size(1000) + assert result == 1000 + + def test_overhead_factor_applied(self): + p = _make_partition(fixed_size=0, overhead_factor=2.0, + extra_filesystem_space=0, extra_partition_space=0) + result = p.get_rootfs_size(1000) + assert result == 2000 + + def test_extra_filesystem_space_added(self): + p = _make_partition(fixed_size=0, overhead_factor=1.0, + extra_filesystem_space=500, extra_partition_space=0) + result = p.get_rootfs_size(100) + assert result == 600 + + def test_extra_filesystem_space_zero_not_overridden(self): + """extra_filesystem_space=0 is falsy; ksparser replaces it with + the 10240 KiB default. The partition should be sized from actual only. + This test exercises get_rootfs_size directly; the ksparser bug means + Partition.extra_filesystem_space is already 10240 by the time this is + called from real wic code. Setting it to 0 here tests the arithmetic + in isolation. + """ + p = _make_partition(fixed_size=0, overhead_factor=1.0, + extra_filesystem_space=0, extra_partition_space=0) + result = p.get_rootfs_size(1000) + # Correct: 0 extra space → result == actual (1000). + assert result == 1000, ( + "extra_filesystem_space=0 must not add any extra blocks; " + "got %d (0 treated as falsy upstream in ksparser)" % result) + + def test_size_param_sets_floor(self): + """--size N ensures the partition is at least N kB.""" + p = _make_partition(size=5000, fixed_size=0, overhead_factor=1.0, + extra_filesystem_space=0, extra_partition_space=0) + # actual < size: get_extra_block_count pads up to size + result = p.get_rootfs_size(1000) + assert result >= 5000 + + def test_actual_larger_than_size_no_shrink(self): + """When actual > --size, get_extra_block_count returns 0; size is not enforced.""" + p = _make_partition(size=1000, fixed_size=0, overhead_factor=1.0, + extra_filesystem_space=0, extra_partition_space=0) + result = p.get_rootfs_size(5000) + # Actual is already larger; no extra blocks added; overhead applied. + assert result == 5000 + + @pytest.mark.parametrize("actual", [0, 1, 512, 1024, 10**6]) + def test_zero_fixed_size_various_actual(self, actual): + p = _make_partition(fixed_size=0, overhead_factor=1.0, + extra_filesystem_space=0, extra_partition_space=0) + result = p.get_rootfs_size(actual) + assert result == actual + + +# --------------------------------------------------------------------------- +# disk_size and fs_size properties +# --------------------------------------------------------------------------- + +class TestDiskSizeAndFsSize: + """disk_size and fs_size properties. + + prepare_empty_partition_btrfs/swap use self.size (which is 0 for + --fixed-size partitions) instead of self.fs_size for the mkfs -b flag. + These tests verify that fs_size returns the correct non-zero value even + when size == 0 and fixed_size is set, so that any caller that correctly + uses fs_size gets the right number. + """ + + def test_disk_size_fixed_no_extra(self): + p = _make_partition(fixed_size=10000, extra_partition_space=0) + assert p.disk_size == 10000 + + def test_disk_size_fixed_with_extra(self): + p = _make_partition(fixed_size=10000, extra_partition_space=2000) + assert p.disk_size == 10000 # disk_size = fixed_size (includes extra) + + def test_disk_size_no_fixed(self): + p = _make_partition(size=5000, fixed_size=0, extra_partition_space=1000) + assert p.disk_size == 6000 + + def test_disk_size_all_zero(self): + p = _make_partition(size=0, fixed_size=0, extra_partition_space=0) + assert p.disk_size == 0 + + def test_fs_size_fixed_no_extra(self): + p = _make_partition(fixed_size=10000, extra_partition_space=0) + assert p.fs_size == 10000 + + def test_fs_size_fixed_with_extra(self): + """fs_size is the filesystem portion: fixed_size minus the raw padding.""" + p = _make_partition(fixed_size=10000, extra_partition_space=2000) + assert p.fs_size == 8000 + + def test_fs_size_fixed_equals_extra_gives_zero(self): + """When extra_partition_space == fixed_size, filesystem size is 0.""" + p = _make_partition(fixed_size=5000, extra_partition_space=5000) + assert p.fs_size == 0 + + def test_fs_size_fixed_with_extra_gt_fixed(self): + """Proxy: when extra_partition_space > fixed_size, fs_size is + negative. A caller using self.size (=0 for fixed-size partitions) + would pass 0 to mkfs.btrfs -b. A caller using self.fs_size gets a + negative value, which is also wrong but differently. The real fix is + to validate that extra_partition_space < fixed_size at parse time. + """ + p = _make_partition(fixed_size=1000, extra_partition_space=2000) + fs = p.fs_size + # Correct: this combination should be rejected at wks-parse time. + if fs <= 0: + pytest.xfail( + "extra_partition_space > fixed_size gives non-positive " + "fs_size (%d); wic should reject at parse time" % fs) + assert fs > 0 + + def test_fs_size_no_fixed(self): + p = _make_partition(size=5000, fixed_size=0, extra_partition_space=1000) + assert p.fs_size == 5000 # no fixed_size → fs_size == size + + def test_size_zero_fixed_zero_fs_size(self): + p = _make_partition(size=0, fixed_size=0, extra_partition_space=0) + assert p.fs_size == 0 + + def test_fs_size_is_nonzero_when_fixed_size_set(self): + """Guard: a --fixed-size partition must have a non-zero fs_size. + prepare_empty_partition_btrfs uses self.size (which IS zero for + fixed-size partitions) instead of self.fs_size. Verify that + fs_size returns the correct value so the bug is clearly in the + caller, not in the property. + """ + p = _make_partition(fixed_size=131072, extra_partition_space=0, + size=0, fstype="btrfs") + # size is 0 (as set by ksparser for --fixed-size partitions) + assert p.size == 0 + # fs_size must NOT be 0 — a btrfs mkfs with -b 0 will fail + assert p.fs_size == 131072, ( + "fs_size must equal fixed_size when extra_partition_space=0; " + "prepare_empty_partition_btrfs should use self.fs_size not self.size") + + +# --------------------------------------------------------------------------- +# __init__ attribute wiring +# --------------------------------------------------------------------------- + +class TestPartitionInit: + """The constructor copies argparse fields onto the instance and sets + a few derived defaults. Verify the wiring and the documented + defaults (sector_size 512, num/device None, source_file "").""" + + def test_fields_copied_from_args(self): + p = _make_partition(mountpoint="/boot", fstype="vfat", label="BOOT", + size=64, disk="mmcblk0") + assert p.mountpoint == "/boot" + assert p.fstype == "vfat" + assert p.label == "BOOT" + assert p.size == 64 + assert p.disk == "mmcblk0" + + def test_derived_defaults(self): + p = _make_partition() + assert p.sector_size == 512 + assert p.num is None + assert p.device is None + assert p.source_file == "" + assert p.lineno == 1 + assert p.has_fstab is False + assert p.update_fstab_in_rootfs is False + + +# --------------------------------------------------------------------------- +# prepare(): early error/return paths that do not need host tools +# --------------------------------------------------------------------------- + +class _FakeCreator: + sector_size = 512 + + +class TestPreparePathsWithoutSource: + """When no --source is given, prepare() validates the request before + ever shelling out. These paths are reachable with no host tools.""" + + def _prepare(self, part): + part.prepare(_FakeCreator(), "/tmp/work", "/tmp/build", None, + "/tmp/boot", "/tmp/kernel", "/tmp/native", None) + + def test_fstype_none_returns_without_error(self): + p = _make_partition(source=None, fstype="none") + assert self._prepare(p) is None + + def test_no_table_returns_without_error(self): + p = _make_partition(source=None, fstype="ext4", no_table=True) + assert self._prepare(p) is None + + def test_zero_size_raises(self): + p = _make_partition(source=None, fstype="ext4", size=0, fixed_size=0) + with pytest.raises(WicError, match="size of zero"): + self._prepare(p) + + def test_empty_squashfs_raises(self): + p = _make_partition(source=None, fstype="squashfs", size=10) + with pytest.raises(WicError, match="not possible to create empty"): + self._prepare(p) + + def test_empty_erofs_raises(self): + p = _make_partition(source=None, fstype="erofs", size=10) + with pytest.raises(WicError, match="not possible to create empty"): + self._prepare(p) + + +class TestPrepareSourceDispatch: + """With a --source, prepare() looks the plugin up, normalises the + name, splits --sourceparams, and dispatches the plugin hooks.""" + + def _prepare(self, part, plugins, monkeypatch): + import wic.partition as partition + + class FakeMgr: + @staticmethod + def get_plugins(ptype): + return plugins + + monkeypatch.setattr(partition, "PluginMgr", FakeMgr) + part.prepare(_FakeCreator(), "/tmp/work", "/tmp/build", None, + "/tmp/boot", "/tmp/kernel", "/tmp/native", None) + + def test_unknown_source_raises(self, monkeypatch): + p = _make_partition(source="does_not_exist", size=4) + with pytest.raises(WicError, match="doesn't exist"): + self._prepare(p, {}, monkeypatch) + + def test_sourceparams_split_into_dict(self, monkeypatch): + captured = {} + + class FakePlugin: + @classmethod + def do_configure_partition(cls, part, sp, *a): + captured["sp"] = sp + + @classmethod + def do_stage_partition(cls, *a): + pass + + @classmethod + def do_prepare_partition(cls, *a): + pass + + @classmethod + def do_post_partition(cls, *a): + pass + + p = _make_partition(source="rawcopy", size=4, + sourceparams="file=x,nocrc") + self._prepare(p, {"rawcopy": FakePlugin}, monkeypatch) + assert captured["sp"] == {"file": "x", "nocrc": None} + + def test_source_name_dash_normalised_to_underscore(self, monkeypatch): + seen = {} + + class FakePlugin: + @classmethod + def do_configure_partition(cls, *a): + seen["configured"] = True + + @classmethod + def do_stage_partition(cls, *a): + pass + + @classmethod + def do_prepare_partition(cls, *a): + pass + + @classmethod + def do_post_partition(cls, *a): + pass + + # plugins are registered under the underscore form + p = _make_partition(source="bootimg-efi", size=4) + self._prepare(p, {"bootimg_efi": FakePlugin}, monkeypatch) + assert seen.get("configured") is True + assert p.source == "bootimg_efi" + + +# --------------------------------------------------------------------------- +# Y2038 message builder +# --------------------------------------------------------------------------- + +class TestY2038MessageBuilder: + """check_for_Y2038_problem warns for ext2/ext3 without running any + tool. The warning text is built from whichever identifier the + partition has (mountpoint, then label, then part_name, then num).""" + + def test_ext2_warns_by_mountpoint(self, monkeypatch): + warnings = [] + import wic.partition as partition + monkeypatch.setattr(partition.logger, "warn", warnings.append) + p = _make_partition(fstype="ext2", mountpoint="/data") + p.check_for_Y2038_problem("/tmp/rootfs.img", "/tmp/native") + assert warnings and "/data" in warnings[0] + assert "Y2038" in warnings[0] + + def test_ext3_warns_by_label_when_no_mountpoint(self, monkeypatch): + warnings = [] + import wic.partition as partition + monkeypatch.setattr(partition.logger, "warn", warnings.append) + p = _make_partition(fstype="ext3", mountpoint=None, label="rootfs") + p.check_for_Y2038_problem("/tmp/rootfs.img", "/tmp/native") + assert warnings and "rootfs" in warnings[0] diff --git a/tests/unit/test_pluginbase.py b/tests/unit/test_pluginbase.py new file mode 100644 index 000000000000..781a0f7b1824 --- /dev/null +++ b/tests/unit/test_pluginbase.py @@ -0,0 +1,93 @@ +r""" +Unit tests for wic/pluginbase.py. + +Covers the parts that do not need a populated plugin directory: + - PluginMgr.get_plugins rejects an unknown plugin type + - PluginMeta registers subclasses that declare a `name` + - ImagerPlugin.do_create raises until a subclass implements it + - SourcePlugin base hooks are inert (return None) + +Where wic does not yet behave correctly, the test is marked xfail with +a short reason. +""" + +import sys +from pathlib import Path + +import pytest + +_SRC = Path(__file__).resolve().parent.parent.parent / "src" +if str(_SRC) not in sys.path: + sys.path.insert(0, str(_SRC)) + +from wic import WicError +from wic.pluginbase import ( + PluginMgr, ImagerPlugin, SourcePlugin, PLUGINS, PLUGIN_TYPES, +) + + +# --------------------------------------------------------------------------- +# PluginMgr.get_plugins +# --------------------------------------------------------------------------- + +class TestGetPlugins: + + def test_invalid_type_raises(self): + with pytest.raises(WicError, match="not valid plugin type"): + PluginMgr.get_plugins("not-a-real-type") + + def test_valid_types_are_known(self): + assert "imager" in PLUGIN_TYPES + assert "source" in PLUGIN_TYPES + + +# --------------------------------------------------------------------------- +# PluginMeta registration +# --------------------------------------------------------------------------- + +class TestPluginRegistration: + + def test_named_imager_subclass_is_registered(self): + class _Imager(ImagerPlugin): + name = "test_imager_xyz" + + assert PLUGINS["imager"].get("test_imager_xyz") is _Imager + + def test_named_source_subclass_is_registered(self): + class _Source(SourcePlugin): + name = "test_source_xyz" + + assert PLUGINS["source"].get("test_source_xyz") is _Source + + def test_unnamed_subclass_not_registered(self): + before = dict(PLUGINS["imager"]) + + class _Anonymous(ImagerPlugin): + pass + + # No `name` attribute, so nothing new is registered. + assert dict(PLUGINS["imager"]) == before + + +# --------------------------------------------------------------------------- +# Base method behaviour +# --------------------------------------------------------------------------- + +class TestBaseMethods: + + def test_imager_do_create_raises_until_implemented(self): + class _Imager(ImagerPlugin): + name = "needs_impl_xyz" + + with pytest.raises(WicError, match="do_create is not implemented"): + _Imager().do_create() + + def test_source_hooks_are_inert(self): + # The default SourcePlugin hooks are documented no-ops; calling + # them must neither raise nor return anything meaningful. + args = ("disk", "name", "creator", "workdir", "builddir", + "bootimg", "kernel", "native") + assert SourcePlugin.do_install_disk(*args) is None + assert SourcePlugin.do_stage_partition("part", {}, "creator", + "wd", "bd", "bi", "kd", + "ns") is None diff --git a/tests/unit/test_update_fstab.py b/tests/unit/test_update_fstab.py new file mode 100644 index 000000000000..08e3fb5c7ce0 --- /dev/null +++ b/tests/unit/test_update_fstab.py @@ -0,0 +1,265 @@ +r""" +Unit tests for DirectImageCreator.update_fstab() (src/wic/plugins/imager/direct.py). + +update_fstab() is pure logic: given the stock /etc/fstab and the list of +partitions, it appends one fstab line per eligible non-root partition and +writes the merged file. We drive it with a fake creator + fake partitions and +inspect the produced fstab text -- no image, no mkfs. + +Device-name selection has five forms: + use_uuid + FAT (len-10 fsuuid) -> UUID=XXXX-XXXX (reformatted) + use_uuid + other fsuuid -> UUID= + use_uuid + no fsuuid -> PARTUUID= + use_label -> LABEL=