Richie
e3602feb5a
- Introduced a new `datasets.nix` file to define ZFS datasets and their properties, allowing for centralized management of dataset configurations. - Updated `default.nix` to import and enable the `zfs_manager` service, integrating it into the system configuration. - Refactored `zfs.sh` to remove dataset creation commands, delegating dataset management to the `zfs_manager` service. - Removed the legacy `snapshot_config.toml` file, as snapshot configurations are now handled within `datasets.nix`. - Modified `vars.nix` to derive mountpoint paths from `datasets.nix`, ensuring consistency across the configuration. - Created a new `zfs.nix` file to define the `zfs_manager` service and its dependencies. - Added comprehensive tests for the `zfs_manager` functionality, covering dataset creation, property management, and error handling.
210 lines
8.6 KiB
Python
210 lines
8.6 KiB
Python
"""test_snapshot_manager."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING
|
|
|
|
import pytest
|
|
|
|
from python.tools.snapshot_manager import get_snapshots_to_delete, get_time_stamp, load_config_data, main
|
|
from python.zfs.dataset import Dataset, Snapshot
|
|
|
|
if TYPE_CHECKING:
|
|
from pyfakefs.fake_filesystem import FakeFilesystem
|
|
from pytest_mock import MockerFixture
|
|
|
|
SNAPSHOT_MANAGER = "python.tools.snapshot_manager"
|
|
|
|
|
|
def patch_utcnow(mocker: MockerFixture, datetime_value: datetime) -> None:
|
|
"""patch_utcnow."""
|
|
mocker.patch("python.tools.snapshot_manager.utcnow", return_value=datetime_value)
|
|
|
|
|
|
def create_mock_snapshot(mocker: MockerFixture, name: str) -> Snapshot:
|
|
"""create_mock_snapshot."""
|
|
mock_snapshot = mocker.MagicMock(spec=Snapshot)
|
|
mock_snapshot.name = name
|
|
|
|
return mock_snapshot
|
|
|
|
|
|
def test_main(mocker: MockerFixture, fs: FakeFilesystem) -> None:
|
|
"""Test main."""
|
|
load_config_data.cache_clear()
|
|
|
|
mocker.patch(f"{SNAPSHOT_MANAGER}.get_time_stamp", return_value="2023-01-01T00:00:00")
|
|
|
|
mock_dataset = mocker.MagicMock(spec=Dataset)
|
|
mock_dataset.name = "test_dataset"
|
|
mock_dataset.create_snapshot.return_value = "snapshot created"
|
|
mock_get_datasets = mocker.patch(f"{SNAPSHOT_MANAGER}.get_datasets", return_value=(mock_dataset,))
|
|
|
|
mock_get_snapshots_to_delete = mocker.patch(f"{SNAPSHOT_MANAGER}.get_snapshots_to_delete", return_value=[])
|
|
mock_signal_alert = mocker.patch(f"{SNAPSHOT_MANAGER}.signal_alert")
|
|
mock_snapshot_config_toml = '["default"]\n15_min = 8\nhourly = 24\ndaily = 0\nmonthly = 0\n'
|
|
fs.create_file("/mock_snapshot_config.toml", contents=mock_snapshot_config_toml)
|
|
main(Path("/mock_snapshot_config.toml"))
|
|
|
|
mock_signal_alert.assert_not_called()
|
|
mock_get_datasets.assert_called_once()
|
|
mock_get_snapshots_to_delete.assert_called_once_with(
|
|
mock_dataset,
|
|
{
|
|
"15_min": 8,
|
|
"hourly": 24,
|
|
"daily": 0,
|
|
"monthly": 0,
|
|
},
|
|
)
|
|
|
|
|
|
def test_main_create_snapshot_failure(mocker: MockerFixture, fs: FakeFilesystem) -> None:
|
|
"""Test main."""
|
|
load_config_data.cache_clear()
|
|
|
|
mocker.patch(f"{SNAPSHOT_MANAGER}.get_time_stamp", return_value="2023-01-01T00:00:00")
|
|
|
|
mock_dataset = mocker.MagicMock(spec=Dataset)
|
|
mock_dataset.name = "test_dataset"
|
|
mock_dataset.create_snapshot.return_value = "snapshot not created"
|
|
mock_get_datasets = mocker.patch(f"{SNAPSHOT_MANAGER}.get_datasets", return_value=(mock_dataset,))
|
|
|
|
mock_get_snapshots_to_delete = mocker.patch(f"{SNAPSHOT_MANAGER}.get_snapshots_to_delete")
|
|
mock_signal_alert = mocker.patch(f"{SNAPSHOT_MANAGER}.signal_alert")
|
|
mock_snapshot_config_toml = '["default"]\n15_min = 8\nhourly = 24\ndaily = 0\nmonthly = 0\n'
|
|
fs.create_file("/mock_snapshot_config.toml", contents=mock_snapshot_config_toml)
|
|
with pytest.raises(SystemExit) as exit_info:
|
|
main(Path("/mock_snapshot_config.toml"))
|
|
|
|
assert exit_info.value.code == 1
|
|
mock_signal_alert.assert_called_once_with("test_dataset failed to create snapshot 2023-01-01T00:00:00")
|
|
mock_get_datasets.assert_called_once()
|
|
mock_get_snapshots_to_delete.assert_not_called()
|
|
|
|
|
|
def test_main_delete_snapshot_failure(mocker: MockerFixture, fs: FakeFilesystem) -> None:
|
|
"""Deletion failures make the service fail after processing the dataset."""
|
|
load_config_data.cache_clear()
|
|
|
|
mocker.patch(f"{SNAPSHOT_MANAGER}.get_time_stamp", return_value="2023-01-01T00:00:00")
|
|
|
|
mock_dataset = mocker.MagicMock(spec=Dataset)
|
|
mock_dataset.name = "test_dataset"
|
|
mock_dataset.create_snapshot.return_value = "snapshot created"
|
|
mocker.patch(f"{SNAPSHOT_MANAGER}.get_datasets", return_value=(mock_dataset,))
|
|
mocker.patch(
|
|
f"{SNAPSHOT_MANAGER}.get_snapshots_to_delete",
|
|
return_value=["test_dataset@auto_202301010000 failed to delete: busy"],
|
|
)
|
|
mocker.patch(f"{SNAPSHOT_MANAGER}.signal_alert")
|
|
mock_snapshot_config_toml = '["default"]\n15_min = 8\nhourly = 24\ndaily = 0\nmonthly = 0\n'
|
|
fs.create_file("/mock_snapshot_config.toml", contents=mock_snapshot_config_toml)
|
|
|
|
with pytest.raises(SystemExit) as exit_info:
|
|
main(Path("/mock_snapshot_config.toml"))
|
|
|
|
assert exit_info.value.code == 1
|
|
|
|
|
|
def test_main_exception(mocker: MockerFixture, fs: FakeFilesystem) -> None:
|
|
"""Test main."""
|
|
load_config_data.cache_clear()
|
|
|
|
mocker.patch(f"{SNAPSHOT_MANAGER}.get_time_stamp", return_value="2023-01-01T00:00:00")
|
|
|
|
mock_dataset = mocker.MagicMock(spec=Dataset)
|
|
mock_dataset.name = "test_dataset"
|
|
mock_dataset.create_snapshot.return_value = "snapshot created"
|
|
mock_get_datasets = mocker.patch(f"{SNAPSHOT_MANAGER}.get_datasets", side_effect=Exception("test"))
|
|
|
|
mock_get_snapshots_to_delete = mocker.patch(f"{SNAPSHOT_MANAGER}.get_snapshots_to_delete")
|
|
mock_signal_alert = mocker.patch(f"{SNAPSHOT_MANAGER}.signal_alert")
|
|
mock_snapshot_config_toml = '["default"]\n15_min = 8\nhourly = 24\ndaily = 0\nmonthly = 0\n'
|
|
fs.create_file("/mock_snapshot_config.toml", contents=mock_snapshot_config_toml)
|
|
with pytest.raises(SystemExit) as pytest_wrapped_e:
|
|
main(Path("/mock_snapshot_config.toml"))
|
|
|
|
assert isinstance(pytest_wrapped_e.value, SystemExit)
|
|
assert pytest_wrapped_e.value.code == 1
|
|
mock_signal_alert.assert_called_once_with("snapshot_manager failed")
|
|
mock_get_datasets.assert_called_once()
|
|
mock_get_snapshots_to_delete.assert_not_called()
|
|
|
|
|
|
def test_get_snapshots_to_delete(mocker: MockerFixture) -> None:
|
|
"""test_get_snapshots_to_delete."""
|
|
mock_snapshot_0 = create_mock_snapshot(mocker, "auto_202509150415")
|
|
mock_snapshot_1 = create_mock_snapshot(mocker, "auto_202509150415")
|
|
|
|
mock_dataset = mocker.MagicMock(spec=Dataset)
|
|
mock_dataset.name = "test_dataset"
|
|
mock_dataset.get_snapshots.return_value = (mock_snapshot_0, mock_snapshot_1)
|
|
mock_dataset.delete_snapshot.return_value = None
|
|
|
|
mock_signal_alert = mocker.patch(f"{SNAPSHOT_MANAGER}.signal_alert")
|
|
|
|
get_snapshots_to_delete(mock_dataset, {"15_min": 1, "hourly": 0, "daily": 0, "monthly": 0})
|
|
|
|
mock_signal_alert.assert_not_called()
|
|
mock_dataset.delete_snapshot.assert_called_once_with("auto_202509150415")
|
|
|
|
|
|
def test_get_snapshots_to_delete_no_snapshot(mocker: MockerFixture) -> None:
|
|
"""test_get_snapshots_to_delete_no_snapshot."""
|
|
mock_dataset = mocker.MagicMock(spec=Dataset)
|
|
mock_dataset.name = "test_dataset"
|
|
mock_dataset.get_snapshots.return_value = ()
|
|
mock_dataset.delete_snapshot.return_value = None
|
|
|
|
mock_signal_alert = mocker.patch(f"{SNAPSHOT_MANAGER}.signal_alert")
|
|
|
|
get_snapshots_to_delete(mock_dataset, {"15_min": 1, "hourly": 0, "daily": 0, "monthly": 0})
|
|
|
|
mock_signal_alert.assert_not_called()
|
|
mock_dataset.delete_snapshot.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize("invalid_count", [-1, "1", None, True])
|
|
def test_invalid_retention_is_rejected_before_reading_snapshots(mocker: MockerFixture, invalid_count: object) -> None:
|
|
"""Invalid standalone TOML values must never reach deletion logic."""
|
|
mock_dataset = mocker.MagicMock(spec=Dataset)
|
|
count_lookup = {"15_min": invalid_count, "hourly": 0, "daily": 0, "monthly": 0}
|
|
|
|
with pytest.raises(ValueError, match="15_min retention must be a non-negative integer"):
|
|
get_snapshots_to_delete(mock_dataset, count_lookup) # type: ignore[arg-type]
|
|
|
|
mock_dataset.get_snapshots.assert_not_called()
|
|
|
|
|
|
def test_get_snapshots_to_delete_errored(mocker: MockerFixture) -> None:
|
|
"""test_get_snapshots_to_delete_errored."""
|
|
mock_snapshot_0 = create_mock_snapshot(mocker, "auto_202509150415")
|
|
mock_snapshot_1 = create_mock_snapshot(mocker, "auto_202509150415")
|
|
|
|
mock_dataset = mocker.MagicMock(spec=Dataset)
|
|
mock_dataset.name = "test_dataset"
|
|
mock_dataset.get_snapshots.return_value = (mock_snapshot_0, mock_snapshot_1)
|
|
mock_dataset.delete_snapshot.return_value = "snapshot has dependent clones"
|
|
|
|
mock_signal_alert = mocker.patch(f"{SNAPSHOT_MANAGER}.signal_alert")
|
|
|
|
failures = get_snapshots_to_delete(
|
|
mock_dataset,
|
|
{"15_min": 1, "hourly": 0, "daily": 0, "monthly": 0},
|
|
)
|
|
|
|
assert failures == ["test_dataset@auto_202509150415 failed to delete: snapshot has dependent clones"]
|
|
mock_signal_alert.assert_called_once_with(
|
|
"test_dataset@auto_202509150415 failed to delete: snapshot has dependent clones"
|
|
)
|
|
mock_dataset.delete_snapshot.assert_called_once_with("auto_202509150415")
|
|
|
|
|
|
def test_get_time_stamp(mocker: MockerFixture) -> None:
|
|
"""Test get_time_stamp."""
|
|
patch_utcnow(mocker, datetime(2023, 1, 1, 0, 0, 0, tzinfo=UTC))
|
|
assert get_time_stamp() == "auto_202301010000"
|