fix overzealous snapshot removal bug

test_zfs_backup.py

@@ -207,7 +207,20 @@ class TestDataset:
 
                 counter = 0
                 ds._clean_old_snapshots_for_interval("daily", MAX_COUNT)
-                assert counter == MAX_COUNT
+                assert counter <= MAX_COUNT
+
+    @pytest.mark.parametrize("max_count", [10, 40])
+    def test__get_oldest_snapshots_exceeding_max_count(self, max_count):
+        snapshots = [
+            snap for snap in gen_snapshots_for_tank_ROOT() if "daily" in str(snap)
+        ]
+        ORIGINAL_COUNT = 31
+        assert len(snapshots) == ORIGINAL_COUNT
+        expected_result_len = ORIGINAL_COUNT - max_count
+        if expected_result_len < 0:
+            expected_result_len = 0
+        result = Dataset._get_oldest_snapshots_exceeding_max_count(snapshots, max_count)
+        assert len(result) == expected_result_len
 
 
 def gen_snapshots_for_tank_ROOT() -> list[Snapshot]:

zfs_backup.py

@@ -205,6 +205,7 @@ class ZFSPath:
         """Return the path in the same manner as `zfs`."""
         path = "/".join(self._elements)
         if self._is_snapshot:
+            assert self._snapshot_name is not None
             return "@".join([path, self._snapshot_name])
         return path
 
@@ -722,22 +723,25 @@ class Dataset:
         for interval, number in config.keep_snapshots_per_interval.items():
             self._clean_old_snapshots_for_interval(interval, number)
 
-    def _clean_old_snapshots_for_interval(
+    def _clean_old_snapshots_for_interval(self, interval: str, max_count: int) -> None:
-        self, interval: str, max_count: int
-    ) -> None:  # TODO I assume the regex matches across different datasets, which it shouldn't
         regex = _get_regex_matching_snapshots_with_tags([config.snapshot_tag, interval])
         snapshots = [
             snapshot for snapshot in self.snapshots if snapshot.matches_regex(regex)
         ]
-        old_snapshots = self._get_n_oldest_snapshots(snapshots, max_count)
+        old_snapshots = self._get_oldest_snapshots_exceeding_max_count(
+            snapshots, max_count
+        )
         for snapshot in old_snapshots:
             snapshot.destroy()
 
     @staticmethod
-    def _get_n_oldest_snapshots(
+    def _get_oldest_snapshots_exceeding_max_count(
-        snapshots: list[Snapshot], count: int
+        snapshots: list[Snapshot], max_count: int
     ) -> list[Snapshot]:
         snapshots.sort()
+        count = len(snapshots) - max_count
+        if count < 0:
+            count = 0
         return snapshots[:count]