"""test_zfs_manager.""" from __future__ import annotations import json from pathlib import Path from typing import TYPE_CHECKING import pytest from python.tools.zfs_manager import main, parse_size, values_match if TYPE_CHECKING: from pyfakefs.fake_filesystem import FakeFilesystem from pytest_mock import MockerFixture ZFS_MANAGER = "python.tools.zfs_manager" CONFIG_PATH = "/mock_zfs_datasets.json" def write_config( fs: FakeFilesystem, datasets: dict[str, dict[str, str]], unmanaged: list[str] | None = None, ) -> Path: """Write a dataset declaration to the fake filesystem.""" contents = { "datasets": {name: {"manageProperties": True, "properties": props} for name, props in datasets.items()} | {name: {"manageProperties": False, "properties": {}} for name in unmanaged or []}, } fs.create_file(CONFIG_PATH, contents=json.dumps(contents)) return Path(CONFIG_PATH) def patch_zfs( mocker: MockerFixture, existing: list[str], properties: dict[str, dict[str, tuple[str, str]]] | None = None, ) -> dict[str, object]: """Patch every zfs call zfs_manager makes.""" return { "list": mocker.patch(f"{ZFS_MANAGER}.list_dataset_names", return_value=existing), "get": mocker.patch(f"{ZFS_MANAGER}.get_properties", side_effect=lambda name: (properties or {}).get(name, {})), "create": mocker.patch(f"{ZFS_MANAGER}.create_dataset", return_value=None), "set": mocker.patch(f"{ZFS_MANAGER}.set_property", return_value=None), "alert": mocker.patch(f"{ZFS_MANAGER}.signal_alert"), } def test_creates_missing_dataset(mocker: MockerFixture, fs: FakeFilesystem) -> None: zfs = patch_zfs(mocker, existing=["media", "media/secure"]) config = write_config(fs, {"media/secure/new": {"compression": "zstd-9"}}) main(config) zfs["create"].assert_called_once_with("media/secure/new", {"compression": "zstd-9"}) zfs["set"].assert_not_called() zfs["alert"].assert_not_called() def test_sets_drifted_property(mocker: MockerFixture, fs: FakeFilesystem) -> None: zfs = patch_zfs( mocker, existing=["media", "media/temp"], properties={"media/temp": {"compression": ("zstd", "inherited from media")}}, ) config = write_config(fs, {"media/temp": {"compression": "zstd-9"}}) main(config) zfs["set"].assert_called_once_with("media/temp", "compression", "zstd-9") zfs["create"].assert_not_called() def test_no_op_when_in_sync(mocker: MockerFixture, fs: FakeFilesystem) -> None: zfs = patch_zfs( mocker, existing=["media", "media/temp"], properties={"media/temp": {"sync": ("disabled", "local")}}, ) config = write_config(fs, {"media/temp": {"sync": "disabled"}}) main(config) zfs["set"].assert_not_called() zfs["create"].assert_not_called() zfs["alert"].assert_not_called() def test_size_property_does_not_churn(mocker: MockerFixture, fs: FakeFilesystem) -> None: """zfs get -p reports recordsize in bytes, the declaration uses a suffix.""" zfs = patch_zfs( mocker, existing=["media", "media/db"], properties={"media/db": {"recordsize": ("16384", "local"), "special_small_blocks": ("32768", "local")}}, ) config = write_config(fs, {"media/db": {"recordsize": "16k", "special_small_blocks": "32K"}}) main(config) zfs["set"].assert_not_called() def test_create_only_property_alerts_instead_of_setting(mocker: MockerFixture, fs: FakeFilesystem) -> None: zfs = patch_zfs( mocker, existing=["media", "media/secure"], properties={"media/secure": {"encryption": ("aes-256-gcm", "local")}}, ) config = write_config(fs, {"media/secure": {"encryption": "off"}}) main(config) zfs["set"].assert_not_called() assert zfs["alert"].call_count == 1 assert "can only be set when the dataset is created" in zfs["alert"].call_args.args[0] def test_undeclared_local_property_warns(mocker: MockerFixture, fs: FakeFilesystem) -> None: zfs = patch_zfs( mocker, existing=["media", "media/temp"], properties={"media/temp": {"exec": ("off", "local")}}, ) config = write_config(fs, {"media/temp": {}}) main(config) zfs["alert"].assert_called_once_with("media/temp has exec=off set outside of nix") def test_undeclared_inherited_property_is_silent(mocker: MockerFixture, fs: FakeFilesystem) -> None: zfs = patch_zfs( mocker, existing=["media", "media/temp"], properties={ "media/temp": { "compression": ("zstd", "inherited from media"), "exec": ("on", "default"), "nixos:shutdown-time": ("whenever", "local"), }, }, ) config = write_config(fs, {"media/temp": {}}) main(config) zfs["alert"].assert_not_called() def test_dry_run_makes_no_changes(mocker: MockerFixture, fs: FakeFilesystem) -> None: zfs = patch_zfs( mocker, existing=["media", "media/temp"], properties={"media/temp": {"compression": ("zstd", "local")}}, ) config = write_config(fs, {"media/temp": {"compression": "zstd-9"}, "media/new": {}}) main(config, dry_run=True) zfs["set"].assert_not_called() zfs["create"].assert_not_called() def test_pool_root_is_never_created(mocker: MockerFixture, fs: FakeFilesystem) -> None: zfs = patch_zfs(mocker, existing=[]) config = write_config(fs, {"media": {"atime": "off"}}) main(config) zfs["create"].assert_not_called() assert "does not create pools" in zfs["alert"].call_args.args[0] def test_children_skipped_when_parent_creation_fails(mocker: MockerFixture, fs: FakeFilesystem) -> None: zfs = patch_zfs(mocker, existing=["media"]) zfs["create"].return_value = "Failed to create media/secure: key not loaded" config = write_config(fs, {"media/secure": {}, "media/secure/child": {}}) main(config) zfs["create"].assert_called_once_with("media/secure", {}) def test_unmanaged_dataset_properties_are_untouched(mocker: MockerFixture, fs: FakeFilesystem) -> None: """A snapshots-only dataset is neither reconciled nor reported as unknown.""" zfs = patch_zfs( mocker, existing=["root_pool", "root_pool/var"], properties={"root_pool/var": {"compression": ("lz4", "local")}}, ) config = write_config(fs, {}, unmanaged=["root_pool", "root_pool/var"]) main(config) zfs["get"].assert_not_called() zfs["set"].assert_not_called() zfs["alert"].assert_not_called() def test_undeclared_dataset_is_left_alone(mocker: MockerFixture, fs: FakeFilesystem) -> None: zfs = patch_zfs(mocker, existing=["media", "media/undeclared"]) config = write_config(fs, {}) main(config) zfs["create"].assert_not_called() zfs["set"].assert_not_called() zfs["alert"].assert_not_called() def test_main_exception(mocker: MockerFixture, fs: FakeFilesystem) -> None: zfs = patch_zfs(mocker, existing=[]) zfs["list"].side_effect = Exception("test") config = write_config(fs, {}) with pytest.raises(SystemExit) as pytest_wrapped_e: main(config) assert pytest_wrapped_e.value.code == 1 zfs["alert"].assert_called_once_with("zfs_manager failed") @pytest.mark.parametrize( ("value", "expected"), [ ("16384", 16384), ("16k", 16384), ("16K", 16384), ("1M", 1048576), ("none", None), ("", None), ], ) def test_parse_size(value, expected) -> None: assert parse_size(value) == expected def test_values_match_falls_back_to_string_for_unparsable_sizes() -> None: assert not values_match("recordsize", "none", "16384") assert values_match("recordsize", "none", "none") assert not values_match("compression", "zstd", "zstd-9")