"""Test zfs.""" import json from datetime import UTC, datetime from subprocess import CompletedProcess from typing import TYPE_CHECKING from unittest.mock import call import pytest from python.zfs import ( CommandResult, Dataset, Snapshot, Zpool, create_dataset, get_datasets, get_properties, list_dataset_names, run_zfs, run_zpool, set_property, ) # Not re-exported from python.zfs on purpose: the wrappers are the default. from python.zfs.command import run_command from python.zfs.dataset import _zfs_list from python.zfs.zpool import _zpool_list if TYPE_CHECKING: from pytest_mock import MockerFixture COMMAND = "python.zfs.command" DATASET = "python.zfs.dataset" ZPOOL = "python.zfs.zpool" SAMPLE_SNAPSHOT_DATA = { "createtxg": "123", "properties": { "creation": {"value": "1620000000"}, "defer_destroy": {"value": "off"}, "guid": {"value": "456"}, "objsetid": {"value": "789"}, "referenced": {"value": "1024"}, "used": {"value": "512"}, "userrefs": {"value": "0"}, "version": {"value": "1"}, "written": {"value": "2048"}, }, "name": "pool/dataset@snap1", } SAMPLE_DATASET_DATA = { "output_version": {"vers_major": 0, "vers_minor": 1, "command": "zfs list"}, "datasets": { "pool/dataset": { "properties": { "aclinherit": {"value": "restricted"}, "aclmode": {"value": "discard"}, "acltype": {"value": "off"}, "available": {"value": "1000000"}, "canmount": {"value": "on"}, "checksum": {"value": "on"}, "clones": {"value": ""}, "compression": {"value": "lz4"}, "copies": {"value": "1"}, "createtxg": {"value": "1234"}, "creation": {"value": "1620000000"}, "dedup": {"value": "off"}, "devices": {"value": "on"}, "encryption": {"value": "off"}, "exec": {"value": "on"}, "filesystem_limit": {"value": "none"}, "guid": {"value": "5678"}, "keystatus": {"value": "none"}, "logbias": {"value": "latency"}, "mlslabel": {"value": "none"}, "mounted": {"value": "yes"}, "mountpoint": {"value": "/pool/dataset"}, "quota": {"value": "0"}, "readonly": {"value": "off"}, "recordsize": {"value": "131072"}, "redundant_metadata": {"value": "all"}, "referenced": {"value": "512000"}, "refquota": {"value": "0"}, "refreservation": {"value": "0"}, "reservation": {"value": "0"}, "setuid": {"value": "on"}, "sharenfs": {"value": "off"}, "snapdir": {"value": "hidden"}, "snapshot_limit": {"value": "none"}, "sync": {"value": "standard"}, "used": {"value": "1024000"}, "usedbychildren": {"value": "512000"}, "usedbydataset": {"value": "256000"}, "usedbysnapshots": {"value": "256000"}, "version": {"value": "5"}, "volmode": {"value": "default"}, "volsize": {"value": "none"}, "vscan": {"value": "off"}, "written": {"value": "4096"}, "xattr": {"value": "on"}, } } }, } SAMPLE_ZPOOL_DATA = { "output_version": {"vers_major": 0, "vers_minor": 1, "command": "zpool list"}, "pools": { "testpool": { "properties": { "allocated": {"value": "1000000"}, "altroot": {"value": "none"}, "ashift": {"value": "12"}, "autoexpand": {"value": "off"}, "autoreplace": {"value": "off"}, "autotrim": {"value": "off"}, "capacity": {"value": "50"}, "comment": {"value": "test pool"}, "dedupratio": {"value": "1.00x"}, "delegation": {"value": "on"}, "expandsize": {"value": "0"}, "failmode": {"value": "wait"}, "fragmentation": {"value": "20"}, "free": {"value": "1000000"}, "freeing": {"value": "0"}, "guid": {"value": "12345678"}, "health": {"value": "ONLINE"}, "leaked": {"value": "0"}, "readonly": {"value": "off"}, "size": {"value": "2000000"}, } } }, } def test_dataset_initialization(mocker: MockerFixture) -> None: """Test Dataset class initialization with mocked ZFS data.""" mocker.patch(f"{DATASET}._zfs_list", return_value=SAMPLE_DATASET_DATA) dataset = Dataset("pool/dataset") assert dataset.__dict__ == { "aclinherit": "restricted", "aclmode": "discard", "acltype": "off", "available": 1000000, "canmount": "on", "checksum": "on", "clones": "", "compression": "lz4", "copies": 1, "createtxg": 1234, "creation": datetime(2021, 5, 3, 0, 0, tzinfo=UTC), "dedup": "off", "devices": "on", "encryption": "off", "exec": "on", "filesystem_limit": "none", "guid": 5678, "keystatus": "none", "logbias": "latency", "mlslabel": "none", "mounted": "yes", "mountpoint": "/pool/dataset", "name": "pool/dataset", "quota": 0, "readonly": "off", "recordsize": 131072, "redundant_metadata": "all", "referenced": 512000, "refquota": 0, "refreservation": 0, "reservation": 0, "setuid": "on", "sharenfs": "off", "snapdir": "hidden", "snapshot_limit": "none", "sync": "standard", "used": 1024000, "usedbychildren": 512000, "usedbydataset": 256000, "usedbysnapshots": 256000, "version": 5, "volmode": "default", "volsize": "none", "vscan": "off", "written": 4096, "xattr": "on", } def test_snapshot_initialization() -> None: """Test Snapshot class initialization with mocked ZFS data.""" snapshot = Snapshot(SAMPLE_SNAPSHOT_DATA) assert snapshot.__dict__ == { "createtxg": 123, "creation": datetime(2021, 5, 3, 0, 0, tzinfo=UTC), "defer_destroy": "off", "guid": 456, "name": "snap1", "objsetid": 789, "referenced": 1024, "used": 512, "userrefs": 0, "version": 1, "written": 2048, } def test_zfs_list_version_check(mocker: MockerFixture) -> None: """Test version validation in _zfs_list.""" mocker.patch( f"{DATASET}.bash_wrapper", return_value=( json.dumps({"output_version": {"vers_major": 1, "vers_minor": 0, "command": "zfs list"}}), 0, ), ) with pytest.raises(RuntimeError) as excinfo: _zfs_list("zfs list invalid -pHj -o all") assert "Datasets are not in the correct format" in str(excinfo.value) def test_get_datasets(mocker: MockerFixture) -> None: """Test get_datasets.""" mock_run = mocker.patch( f"{DATASET}.run_zfs", return_value=CommandResult( args=(), stdout="pool/dataset\npool/other\ninvalid", stderr="", return_code=0, ), ) mock_dataset = mocker.patch(f"{DATASET}.Dataset") get_datasets() mock_run.assert_called_once_with("list", "-Hp", "-t", "filesystem", "-o", "name") calls = [call("pool/dataset"), call("pool/other")] mock_dataset.assert_has_calls(calls) def test_zpool_initialization(mocker: MockerFixture) -> None: """Test Zpool class initialization with mocked ZFS data.""" mocker.patch(f"{ZPOOL}._zpool_list", return_value=SAMPLE_ZPOOL_DATA) zpool = Zpool("testpool") assert zpool.__dict__ == { "name": "testpool", "allocated": 1000000, "altroot": "none", "ashift": 12, "autoexpand": "off", "autoreplace": "off", "autotrim": "off", "capacity": 50, "comment": "test pool", "dedupratio": "1.00x", "delegation": "on", "expandsize": "0", "failmode": "wait", "fragmentation": 20, "free": "1000000", "freeing": 0, "guid": 12345678, "health": "ONLINE", "leaked": 0, "readonly": "off", "size": 2000000, } def test_zpool_repr(mocker: MockerFixture) -> None: """Test Zpool string representation.""" mocker.patch(f"{ZPOOL}._zpool_list", return_value=SAMPLE_ZPOOL_DATA) zpool = Zpool("testpool") repr_string = repr(zpool) expected_attrs = [ "name", "allocated", "altroot", "ashift", "autoexpand", "autoreplace", "autotrim", "capacity", "comment", "dedupratio", "delegation", "expandsize", "failmode", "fragmentation", "freeing", "guid", "health", "leaked", "readonly", "size", ] for attr in expected_attrs: assert f"{attr}=" in repr_string def test_zpool_list(mocker: MockerFixture) -> None: """Test version validation in _zpool_list.""" mocker.patch( f"{ZPOOL}.run_zpool", return_value=CommandResult( args=(), stdout=json.dumps({"output_version": {"vers_major": 0, "vers_minor": 1, "command": "zpool list"}}), stderr="", return_code=0, ), ) result = _zpool_list("list", "invalid", "-pHj", "-o", "all") assert result == {"output_version": {"command": "zpool list", "vers_major": 0, "vers_minor": 1}} def test_zpool_list_version_check(mocker: MockerFixture) -> None: """Test version validation in _zpool_list.""" mocker.patch( f"{ZPOOL}.run_zpool", return_value=CommandResult( args=(), stdout=json.dumps({"output_version": {"vers_major": 1, "vers_minor": 0, "command": "zpool list"}}), stderr="", return_code=0, ), ) with pytest.raises(RuntimeError) as excinfo: _zpool_list("list", "invalid", "-pHj", "-o", "all") assert "Datasets are not in the correct format" in str(excinfo.value) # -- run_zfs, the subprocess boundary ----------------------------------------- def completed(returncode: int = 0, stdout: str = "", stderr: str = "") -> CompletedProcess: """Build a CompletedProcess the way subprocess.run would return one.""" return CompletedProcess(args=["zfs"], returncode=returncode, stdout=stdout, stderr=stderr) def test_run_zfs_passes_arguments_as_a_list(mocker: MockerFixture) -> None: """Arguments must never be joined into a string and re-split.""" mock_run = mocker.patch(f"{COMMAND}.subprocess.run", return_value=completed(stdout="ok\n")) run_zfs("get", "-Hp", "all", "media/temp") mock_run.assert_called_once_with( ["zfs", "get", "-Hp", "all", "media/temp"], capture_output=True, encoding="utf-8", check=False, ) def test_run_zfs_keeps_values_containing_spaces_intact(mocker: MockerFixture) -> None: """A property value with a space must reach zfs as one argument. bash_wrapper split on whitespace, so this silently became several arguments and zfs was handed something it could not parse. """ mock_run = mocker.patch(f"{COMMAND}.subprocess.run", return_value=completed()) run_zfs("set", "mountpoint=/zfs/two words", "media/temp") assert mock_run.call_args.args[0] == ["zfs", "set", "mountpoint=/zfs/two words", "media/temp"] def test_run_zfs_reports_success(mocker: MockerFixture) -> None: mocker.patch(f"{COMMAND}.subprocess.run", return_value=completed(stdout="output\n", stderr="")) result = run_zfs("list") assert result.ok assert result.stdout == "output\n" assert result.return_code == 0 assert result.args == ("zfs", "list") def test_run_zfs_returns_failures_rather_than_raising(mocker: MockerFixture) -> None: """Callers decide what a failure means, so run_zfs never raises.""" mocker.patch(f"{COMMAND}.subprocess.run", return_value=completed(returncode=1, stderr="no such pool\n")) result = run_zfs("list") assert not result.ok assert result.stderr == "no such pool\n" assert result.message == "no such pool" def test_run_zfs_keeps_streams_separate(mocker: MockerFixture) -> None: """A warning on stderr must not contaminate stdout. bash_wrapper returned stderr in place of stdout whenever stderr was non-empty, even on success, which is how a warning could be parsed as a list of dataset names. """ mocker.patch( f"{COMMAND}.subprocess.run", return_value=completed(stdout="pool/one\n", stderr="warning: something\n"), ) result = run_zfs("list") assert result.ok assert result.stdout == "pool/one\n" assert result.stderr == "warning: something\n" def test_command_result_message_falls_back_to_stdout() -> None: """Some zfs errors land on stdout, so the message must not be empty.""" result = CommandResult(args=("list",), stdout=" something went wrong \n", stderr="", return_code=1) assert result.message == "something went wrong" # -- list_dataset_names ------------------------------------------------------- def patch_run(mocker: MockerFixture, **kwargs) -> object: """Patch run_zfs with a single canned result.""" return mocker.patch(f"{DATASET}.run_zfs", return_value=CommandResult(args=(), **kwargs)) def test_list_dataset_names_builds_the_right_command(mocker: MockerFixture) -> None: mock_run = patch_run(mocker, stdout="pool\n", stderr="", return_code=0) list_dataset_names() mock_run.assert_called_once_with("list", "-Hp", "-t", "filesystem", "-o", "name") def test_list_dataset_names_includes_pool_roots(mocker: MockerFixture) -> None: """Unlike get_datasets, nothing is filtered out, so existence checks work.""" patch_run(mocker, stdout="media\nmedia/temp\nmedia/secure/docker\n", stderr="", return_code=0) assert list_dataset_names() == ["media", "media/temp", "media/secure/docker"] def test_list_dataset_names_drops_blank_lines(mocker: MockerFixture) -> None: patch_run(mocker, stdout="media\n\nmedia/temp\n\n", stderr="", return_code=0) assert list_dataset_names() == ["media", "media/temp"] def test_list_dataset_names_raises_on_failure(mocker: MockerFixture) -> None: """Never return a partial list: the reconciler would create everything.""" patch_run(mocker, stdout="", stderr="cannot open 'media': no such pool\n", return_code=1) with pytest.raises(RuntimeError) as excinfo: list_dataset_names() assert "Failed to list ZFS datasets" in str(excinfo.value) assert "no such pool" in str(excinfo.value) def test_list_dataset_names_never_parses_stderr_as_names(mocker: MockerFixture) -> None: """The regression this guards: error text read as a dataset list.""" patch_run(mocker, stdout="", stderr="permission denied\n", return_code=1) with pytest.raises(RuntimeError): list_dataset_names() # -- get_properties ----------------------------------------------------------- def test_get_properties_builds_the_right_command(mocker: MockerFixture) -> None: mock_run = patch_run(mocker, stdout="", stderr="", return_code=0) get_properties("media/temp") mock_run.assert_called_once_with("get", "-Hp", "-o", "property,value,source", "all", "media/temp") def test_get_properties_parses_value_and_source(mocker: MockerFixture) -> None: patch_run( mocker, stdout="compression\tzstd-9\tlocal\natime\toff\tinherited from media\nexec\ton\tdefault\n", stderr="", return_code=0, ) assert get_properties("media/temp") == { "compression": ("zstd-9", "local"), "atime": ("off", "inherited from media"), "exec": ("on", "default"), } def test_get_properties_skips_blank_lines(mocker: MockerFixture) -> None: """A stray blank line in the middle must not raise on unpacking.""" patch_run(mocker, stdout="compression\tzstd\tlocal\n\natime\toff\tlocal\n", stderr="", return_code=0) assert get_properties("media/temp") == { "compression": ("zstd", "local"), "atime": ("off", "local"), } def test_get_properties_raises_on_failure(mocker: MockerFixture) -> None: patch_run(mocker, stdout="", stderr="dataset does not exist\n", return_code=1) with pytest.raises(RuntimeError) as excinfo: get_properties("media/gone") assert "Failed to get properties for media/gone" in str(excinfo.value) # -- create_dataset ----------------------------------------------------------- def test_create_dataset_builds_sorted_option_flags(mocker: MockerFixture) -> None: mock_run = patch_run(mocker, stdout="", stderr="", return_code=0) assert create_dataset("media/temp", {"sync": "disabled", "compression": "zstd-9"}) is None mock_run.assert_called_once_with( "create", "-o", "compression=zstd-9", "-o", "sync=disabled", "media/temp", ) def test_create_dataset_with_no_properties(mocker: MockerFixture) -> None: mock_run = patch_run(mocker, stdout="", stderr="", return_code=0) create_dataset("media/temp", {}) mock_run.assert_called_once_with("create", "media/temp") def test_create_dataset_keeps_a_value_with_spaces_together(mocker: MockerFixture) -> None: mock_run = patch_run(mocker, stdout="", stderr="", return_code=0) create_dataset("media/temp", {"mountpoint": "/zfs/two words"}) assert mock_run.call_args.args == ("create", "-o", "mountpoint=/zfs/two words", "media/temp") def test_create_dataset_returns_the_error_on_failure(mocker: MockerFixture) -> None: patch_run(mocker, stdout="", stderr="cannot create 'media/temp': out of space\n", return_code=1) error = create_dataset("media/temp", {}) assert error is not None assert "Failed to create media/temp" in error assert "out of space" in error # -- set_property ------------------------------------------------------------- def test_set_property_builds_the_right_command(mocker: MockerFixture) -> None: mock_run = patch_run(mocker, stdout="", stderr="", return_code=0) assert set_property("media/temp", "compression", "zstd-9") is None mock_run.assert_called_once_with("set", "compression=zstd-9", "media/temp") def test_set_property_keeps_a_value_with_spaces_together(mocker: MockerFixture) -> None: mock_run = patch_run(mocker, stdout="", stderr="", return_code=0) set_property("media/temp", "mountpoint", "/zfs/two words") assert mock_run.call_args.args == ("set", "mountpoint=/zfs/two words", "media/temp") def test_set_property_returns_the_error_on_failure(mocker: MockerFixture) -> None: patch_run(mocker, stdout="", stderr="permission denied\n", return_code=1) error = set_property("media/temp", "compression", "zstd-9") assert error is not None assert "Failed to set compression=zstd-9 on media/temp" in error assert "permission denied" in error # -- run_zpool, the same wrapper the zfs side uses ---------------------------- def test_run_zpool_prefixes_the_binary(mocker: MockerFixture) -> None: mock_run = mocker.patch(f"{COMMAND}.subprocess.run", return_value=completed(stdout="{}")) run_zpool("list", "media", "-pHj", "-o", "all") mock_run.assert_called_once_with( ["zpool", "list", "media", "-pHj", "-o", "all"], capture_output=True, encoding="utf-8", check=False, ) def test_run_zpool_reports_failures(mocker: MockerFixture) -> None: mocker.patch(f"{COMMAND}.subprocess.run", return_value=completed(returncode=1, stderr="no such pool\n")) result = run_zpool("list", "gone") assert not result.ok assert result.message == "no such pool" assert result.args == ("zpool", "list", "gone") def test_run_zpool_keeps_streams_separate(mocker: MockerFixture) -> None: """The same contamination bug the zfs side had must not exist here.""" mocker.patch( f"{COMMAND}.subprocess.run", return_value=completed(stdout="{}", stderr="warning: pool is degraded\n"), ) result = run_zpool("list") assert result.ok assert result.stdout == "{}" def test_run_zpool_keeps_values_containing_spaces_intact(mocker: MockerFixture) -> None: mock_run = mocker.patch(f"{COMMAND}.subprocess.run", return_value=completed()) run_zpool("set", "comment=two words", "media") assert mock_run.call_args.args[0] == ["zpool", "set", "comment=two words", "media"] def test_run_command_runs_what_it_is_given(mocker: MockerFixture) -> None: """zfs and zpool are the same function with a different first argument.""" mock_run = mocker.patch(f"{COMMAND}.subprocess.run", return_value=completed()) run_command("zpool", "status") assert mock_run.call_args.args[0] == ["zpool", "status"] def test_zpool_list_raises_when_the_command_fails(mocker: MockerFixture) -> None: """A failed zpool must never have its error text decoded as json.""" mocker.patch( f"{ZPOOL}.run_zpool", return_value=CommandResult(args=(), stdout="", stderr="no such pool: media\n", return_code=1), ) with pytest.raises(RuntimeError) as excinfo: _zpool_list("list", "media", "-pHj", "-o", "all") assert "Failed to run zpool" in str(excinfo.value) assert "no such pool" in str(excinfo.value) def test_zpool_builds_the_right_command(mocker: MockerFixture) -> None: """Zpool passes argv through rather than a formatted string.""" mock_run = mocker.patch( f"{ZPOOL}.run_zpool", return_value=CommandResult( args=(), stdout=json.dumps(SAMPLE_ZPOOL_DATA), stderr="", return_code=0, ), ) Zpool("testpool") mock_run.assert_called_once_with("list", "testpool", "-pHj", "-o", "all")