Fix bug introduced which breaks pacman hook if internal_preupdate = 0

Please see issue #25

src/code/snap_operator.py

@@ -58,6 +58,9 @@ def find_target(config: configs.Config, suffix: str) -> Optional[snap_holder.Sna
 
 def _all_but_last_k(array: list[_GenericT], k: int) -> Iterator[_GenericT]:
     """All but at most k last elements."""
+    # Edge cases -
+    #   k > len(array): Returns empty array.
+    #   k < 0: Error.
     if k < 0:
         raise ValueError(f"k = {k} < 0")
     yield from array[: len(array) - k]
@@ -111,15 +114,22 @@ def _create_and_maintain_n_backups(
         ]
 
         if count > 0:
+            # From previously existing snaps, leave count - 1 snaps (since we
+            # will create one more).
+            n_snaps_to_leave = count - 1
+            # Create a new snap.
             snapshot = snap_holder.Snapshot(self._config.dest_prefix + self._now_str)
             snapshot.metadata.trigger = trigger
             if comment:
                 snapshot.metadata.comment = comment
             snapshot.create_from(self._config.source)
             self.snaps_created = True
+        else:
+            # From existing snaps, delete all.
+            n_snaps_to_leave = 0
 
-        # Clean up old snaps; leave count-1 previous snaps (plus the one now created).
-        for expired in _all_but_last_k(previous_snaps, count - 1):
+        # Clean up old snaps.
+        for expired in _all_but_last_k(previous_snaps, n_snaps_to_leave):
             expired.delete()
             self.snaps_deleted = True
 

src/code/snap_operator_test.py

@@ -111,6 +111,58 @@ def test_scheduled_triggered(self):
         self._mock_delete.assert_called_once_with()
         self._mock_create_from.assert_called_with("snap_source")
 
+    def test_pachook(self):
+        def setup_n_snaps(n: int):
+            self._old_snaps = [
+                snap_holder.Snapshot(f"/tmp/nodir/@home-202311150{k}0000")
+                for k in range(n)
+            ]
+            for snap in self._old_snaps:
+                snap.metadata.trigger = "I"
+            trigger_interval = datetime.timedelta(hours=12).total_seconds()
+            snapper = snap_operator.SnapOperator(
+                config=configs.Config(
+                    config_file="config_file",
+                    source="snap_source",
+                    dest_prefix="dest_prefix",
+                    trigger_interval=trigger_interval,
+                ),
+                now=_utc_to_local("20230213" "130000"),
+            )
+            self._mock_delete.reset_mock()
+            self._mock_create_from.reset_mock()
+            return snapper
+
+        # Have 0, need 0 => Create 0, delete 0.
+        setup_n_snaps(0)._create_and_maintain_n_backups(0, "I", None)
+        self._mock_create_from.assert_not_called()
+        self._mock_delete.assert_not_called()
+
+        # Have 0, need 3 => Create 1, delete 0.
+        setup_n_snaps(0)._create_and_maintain_n_backups(3, "I", None)
+        self._mock_create_from.assert_called_once_with("snap_source")
+        self._mock_delete.assert_not_called()
+
+        # Have 3, need 4 => Create 1, delete 0.
+        setup_n_snaps(3)._create_and_maintain_n_backups(4, "I", None)
+        self._mock_create_from.assert_called_once_with("snap_source")
+        self._mock_delete.assert_not_called()
+
+        # Have 3, need 3 => Create 1, delete 1.
+        setup_n_snaps(3)._create_and_maintain_n_backups(3, "I", None)
+        self._mock_create_from.assert_called_once_with("snap_source")
+        self._mock_delete.assert_called_once_with()
+
+        # Have 3, need 2 => Create 1, delete 2.
+        setup_n_snaps(3)._create_and_maintain_n_backups(2, "I", None)
+        self._mock_create_from.assert_called_once_with("snap_source")
+        self.assertEqual(self._mock_delete.call_count, 2)
+
+        # Have 3, need 0 => Create 0, delete 3.
+        setup_n_snaps(3)._create_and_maintain_n_backups(0, "I", None)
+        self._mock_create_from.assert_not_called()
+        self.assertEqual(self._mock_delete.call_count, 3)
+
     def test_list_json(self):
         self._old_snaps = [snap_holder.Snapshot("/tmp/nodir/@home-20230213001000")]
         self._old_snaps[-1].metadata.trigger = "S"