zfs-backup/test_zfs_backup.py

import pytest

from unittest import mock

from zfs_backup import ZFSPath, Dataset, Snapshot


class TestZFSPath:
    @pytest.mark.parametrize(
        "elements",
        [
            ["pool"],
            ["pool", "dataset1", "dataset2"],
        ],
    )
    def test__sanity_check_valid(self, elements):
        ZFSPath._sanity_check(elements)

    @pytest.mark.parametrize(
        "elements",
        [
            [""],
            ["", "dataset1"],
            ["pool", "dataset1", ""],
            ["pool", "", "dataset2"],
        ],
    )
    def test__sanity_check_invalid(self, elements):
        with pytest.raises(ValueError):
            ZFSPath._sanity_check(elements)

    @pytest.mark.parametrize(
        "elements, snapshot_name, result",
        [
            [["pool"], None, ""],
            [["pool", "dataset"], None, "dataset"],
            [["pool", "dataset"], "snapshot", "dataset@snapshot"],
            [
                ["pool", "dataset1", "dataset2"],
                "snapshot",
                "dataset1/dataset2@snapshot",
            ],
        ],
        ids=[
            "only pool",
            "pool with one dataset",
            "pool with dataset and snapshot",
            "pool with two nested datasets and snapshot",
        ],
    )
    def test_name_without_pool(self, elements, snapshot_name, result):
        instance = ZFSPath(elements, snapshot_name)
        assert instance.name_without_pool == result

    @pytest.mark.parametrize(
        "elements, snapshot_name, result",
        [
            [["pool"], None, "pool"],
            [["pool", "dataset"], None, "pool/dataset"],
            [["pool", "dataset"], "snapshot", "pool/dataset@snapshot"],
            [
                ["pool", "dataset1", "dataset2"],
                "snapshot",
                "pool/dataset1/dataset2@snapshot",
            ],
        ],
        ids=[
            "only pool",
            "pool with one dataset",
            "pool with dataset and snapshot",
            "pool with two nested datasets and snapshot",
        ],
    )
    def test___repr__(self, elements, snapshot_name, result):
        assert str(ZFSPath(elements, snapshot_name)) == result

    @pytest.mark.parametrize(
        "instance1, instance2, result",
        [
            [ZFSPath(["pool"]), ZFSPath(["pool"]), True],
            [ZFSPath(["pool"]), ZFSPath(["tank"]), False],
            [ZFSPath(["pool", "dataset"]), ZFSPath(["pool", "dataset"]), True],
            [ZFSPath(["pool", "dataset"]), ZFSPath(["pool", "bla"]), False],
            [
                ZFSPath(["pool", "dataset"], "snapshot"),
                ZFSPath(["pool", "dataset"], "snapshot"),
                True,
            ],
            [
                ZFSPath(["pool", "dataset"], "snapshot"),
                ZFSPath(["pool", "dataset"], "bla"),
                False,
            ],
        ],
        ids=[
            "only pool equal",
            "only pool not equal",
            "dataset equal",
            "dataset not equal",
            "snapshot equal",
            "snapshot not equal",
        ],
    )
    def test___eq__(self, instance1, instance2, result):
        assert (instance1 == instance2) == result

    @pytest.mark.parametrize(
        "string, elements, snapshot_name",
        [
            ["pool/dataset", ["pool", "dataset"], None],
            ["pool/dataset@snapshot", ["pool", "dataset"], "snapshot"],
            [
                "pool/dataset1/dataset2@snapshot",
                ["pool", "dataset1", "dataset2"],
                "snapshot",
            ],
        ],
        ids=[
            "pool/dataset",
            "pool/dataset@snapshot",
            "pool/dataset1/dataset2@snapshot",
        ],
    )
    def test_from_string(self, string, elements, snapshot_name):
        instance = ZFSPath.from_string(string)
        assert instance._elements == elements
        assert instance._snapshot_name == instance.snapshot_name == snapshot_name

    @pytest.mark.parametrize(
        "instance1, instance2",
        [
            [ZFSPath(["pool", "dataset"]), ZFSPath(["pool", "dataset"])],
            [ZFSPath(["pool", "dataset"], "snapshot"), ZFSPath(["pool", "dataset"])],
        ],
        ids=[
            "pool with dataset",
            "pool with dataset and snapshot",
        ],
    )
    def test_dataset(self, instance1, instance2):
        assert instance1.dataset == instance2

    @pytest.mark.parametrize(
        "instance1, instance2",
        [
            [ZFSPath(["pool"]), ZFSPath(["tank"])],
            [ZFSPath(["pool", "dataset"]), ZFSPath(["tank", "dataset"])],
            [
                ZFSPath(["pool", "dataset"], "snapshot"),
                ZFSPath(["tank", "dataset"], "snapshot"),
            ],
        ],
        ids=[
            "pool",
            "pool with dataset",
            "pool with dataset and snapshot",
        ],
    )
    def test_replace_pool(self, instance1, instance2):
        assert instance1.replace_pool(instance2.pool_name) == instance2


class TestZFS:
    @pytest.mark.parametrize("string", ["tank", "tank/ROOT", "tank/ROOT/default"])
    def test_get_snapshots(self, string):
        path = ZFSPath.from_string(string)
        ds = Dataset(path)
        for snapshot in ds.snapshots:
            assert snapshot.dataset == ds


class TestDataset:
    @pytest.mark.parametrize("zfspath", ["tank", "tank/ROOT", "tank/ROOT/default"])
    def test__clean_old_snapshots_for_interval(self, zfspath):
        """Check that only snapshots that are direct children of the dataset are found during cleanup."""

        def validate_snapshot_selection(snapshot: Snapshot):
            print(snapshot)  # show output with "pytest -s"
            str(snapshot).startswith(f"{zfspath}@")

        dataset = Dataset(ZFSPath.from_string(zfspath))
        with mock.patch.object(
            Snapshot, "destroy", lambda snapshot: validate_snapshot_selection(snapshot)
        ):
            dataset._clean_old_snapshots_for_interval("daily", 999)