Clean up implementation and update tests

ch_tools/monrun_checks/ch_backup.py

@@ -6,7 +6,7 @@
 import os.path
 from datetime import datetime, timedelta, timezone
 from os.path import exists
-from typing import Dict, List, Tuple
+from typing import Dict, List, Sequence
 
 from click import Context
 from cloup import command, option, pass_context
@@ -132,27 +132,25 @@ def _check_valid_backups_exist(backups: List[Dict]) -> Result:
     return Result(CRIT, "No valid backups found")
 
 
-def _count_failed_backups(backups: List[Dict]) -> Tuple[int, int]:
-    counter, userfault_counter = 0, 0
+def _count_failed_backups(backups: List[Dict]) -> int:
+    counter = 0
     for i, backup in enumerate(backups):
         state = backup["state"]
 
         if state == "created":
             break
 
         if (state == "failed") or (state == "creating" and i > 0):
-            if "exception" in backup and _is_userfault_exception(backup["exception"]):
-                userfault_counter += 1
             counter += 1
 
-    return counter, userfault_counter
+    return counter
 
 
 def _check_last_backup_not_failed(backups: List[Dict], crit_threshold: int) -> Result:
     """
     Check that the last backup is not failed. Its status must be 'created' or 'creating'.
     """
-    counter, _ = _count_failed_backups(backups)
+    counter = _count_failed_backups(backups)
 
     if counter == 0:
         return Result(OK)
@@ -292,7 +290,7 @@ def _merge_results(*results: Result) -> Result:
     return merged_result
 
 
-def _is_backup_failed_by_userfault_error(backups: List[Dict]) -> bool:
+def _is_backup_failed_by_userfault_error(backups: Sequence[Dict]) -> bool:
     failed_backup = None
     for i, backup in enumerate(backups):
         state = backup["state"]
@@ -305,17 +303,5 @@ def _is_backup_failed_by_userfault_error(backups: List[Dict]) -> bool:
     if not failed_backup:
         return False
 
-    exception = failed_backup.get("exception", "")
-    return any(value in exception for value in USERFAULT_ERRORS)
-
-
-def _is_userfault_exception(exception):
-    """
-    Check if exception was caused by user.
-    Current list:
-      * Disk quota exceeded
-    """
-    if not exception:
-        return False
-
-    return any(value in exception for value in USERFAULT_ERRORS)
+    exception_msg = failed_backup.get("exception") or ""
+    return any(value in exception_msg for value in USERFAULT_ERRORS)

tests/unit/monrun/test_last_backup.py → tests/unit/monrun/test_backup.py

@@ -1,63 +1,56 @@
 from typing import Dict, Sequence
 
-from hamcrest import assert_that, equal_to
 from pytest import mark
 
-from ch_tools.monrun_checks.ch_backup import _count_failed_backups
+from ch_tools.monrun_checks.ch_backup import _is_backup_failed_by_userfault_error
 
 
 @mark.parametrize(
-    ["backups", "userfault_expected"],
+    ["backups", "result"],
     [
-        (({"state": "created"},), 0),
-        (
-            (
-                {"state": "failed", "exception": None},
-                {"state": "created"},
-            ),
-            0,
-        ),
+        (({"state": "created"},), False),
         (
             (
                 {"state": "failed"},
                 {"state": "created"},
             ),
-            0,
+            False,
         ),
         (
             (
                 {"state": "failed"},
                 {"state": "failed"},
                 {"state": "created"},
             ),
-            0,
+            False,
         ),
         (
             (
-                {"state": "failed", "exception": "Disk quota exceeded"},
-                {"state": "failed"},
-                {"state": "failed"},
+                {"state": "failed", "exception": None},
                 {"state": "created"},
             ),
-            1,
+            False,
         ),
         (
             (
-                {"state": "failed", "exception": "God's will"},
+                {
+                    "state": "failed",
+                    "exception": "ClickhouseError: Code: 243. DB::Exception: Cannot reserve 1.00 MiB, not enough space. (NOT_ENOUGH_SPACE) (version 23.8.14.6 (official build))",
+                },
                 {"state": "failed"},
                 {"state": "failed"},
                 {"state": "created"},
             ),
-            0,
+            True,
         ),
         (
             (
-                {"state": "failed", "exception": None},
+                {"state": "failed", "exception": "God's will"},
                 {"state": "failed"},
                 {"state": "failed"},
                 {"state": "created"},
             ),
-            0,
+            False,
         ),
         (
             (
@@ -66,12 +59,11 @@
                 {"state": "failed", "exception": "Disk quota exceeded"},
                 {"state": "created"},
             ),
-            3,
+            True,
         ),
     ],
 )
-def test_last_backup_not_failed(
-    backups: Sequence[Dict], userfault_expected: Sequence[int]
+def test_is_backup_failed_by_userfault_error(
+    backups: Sequence[Dict], result: bool
 ) -> None:
-    _, userfault = _count_failed_backups(list(backups))
-    assert_that(userfault, equal_to(userfault_expected))
+    assert _is_backup_failed_by_userfault_error(backups) == result