fix: count distinct guests in backup recovery summary

AppImage/scripts/tests/test_audit_backup_recovered_count.py

@@ -0,0 +1,152 @@
+"""Offline backup-summary regressions; never import the administrative module.
+
+Run: python3 -m unittest discover -s AppImage/scripts/tests -p test_audit_backup_recovered_count.py -v
+"""
+import ast
+import json
+from pathlib import Path
+from types import SimpleNamespace
+import unittest
+
+
+SOURCE = Path(__file__).resolve().parents[1] / "audit_checks_pve.py"
+
+
+def load_check(source=SOURCE):
+    tree = ast.parse(source.read_text())
+    names = {"_backup_job_results", "_unverified"}
+    nodes = [n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name in names]
+    assert {n.name for n in nodes} == names
+    for node in nodes:
+        node.decorator_list = []
+    env = {"json": json, "CLASS_WARNING": "WARNING",
+           "CLASS_OBSERVATION": "OBSERVATION", "CLASS_CONFORMANT": "CONFORMANT",
+           "CLASS_UNVERIFIED": "UNVERIFIED"}
+    exec(compile(ast.Module(body=nodes, type_ignores=[]), str(source), "exec"), env)
+    return env["_backup_job_results"]
+
+
+def evaluate(tasks=None, *, response=None, check=None):
+    calls = []
+
+    def run(argv, **kwargs):
+        assert argv == ["pvesh", "get", "/nodes/fixture/tasks", "--typefilter",
+                        "vzdump", "--limit", "200", "--output-format", "json"], argv
+        assert kwargs == {"timeout": 30}, kwargs
+        calls.append(argv)
+        return response if response is not None else (0, json.dumps(tasks))
+
+    result = (check or load_check())(SimpleNamespace(node="fixture", run=run))
+    assert len(calls) == 1, calls
+    return result
+
+
+def runs(vmid, statuses, **extra):
+    return [{"type": "vzdump", "id": vmid, "status": status, "starttime": i,
+             "upid": f"UPID:fixture:0001:0001:{i:08x}:vzdump:{vmid}:root@pam:",
+             **extra} for i, status in enumerate(statuses.split(), 1)]
+
+
+class BackupRecoveredCountTests(unittest.TestCase):
+    def assert_summary(self, tasks, count, total, history, key="recovered", incomplete=False):
+        result = evaluate(tasks)
+        self.assertEqual(result["summary_key"], key)
+        self.assertEqual(result["summary_params"], {"count": str(count), "total": str(total)})
+        self.assertEqual(result["incomplete"], incomplete)
+        healed = [r for r in result["affected"] if r["reason_key"] == "backupRunRecovered"]
+        self.assertEqual([r["when"] for r in healed], history)
+        self.assertTrue(all(r["classification"] == "OBSERVATION" for r in healed))
+        self.assertIn(f"earlier failures a later run cleared: {len(history)}\n", result["evidence"])
+        return result
+
+    def test_repeated_recoveries_count_one_guest_keep_both_events(self):
+        tasks = runs(101, "ERROR OK ERROR OK")
+        result = self.assert_summary(tasks, 1, 1, [1, 3])
+        self.assertEqual([r["upid"] for r in result["affected"]], [tasks[0]["upid"], tasks[2]["upid"]])
+
+    def test_repeated_recovery_plus_clean_guest_is_not_clamped_event_count(self):
+        self.assert_summary(runs(101, "ERROR OK ERROR OK") + runs(102, "OK"), 1, 2, [1, 3])
+
+    def test_single_recovery(self):
+        self.assert_summary(runs(101, "ERROR OK"), 1, 1, [1])
+
+    def test_two_recovered_guests(self):
+        self.assert_summary(runs(101, "ERROR OK") + runs(102, "ERROR OK"), 2, 2, [1, 1])
+
+    def test_repeated_failures_record_only_failure_cleared_by_success(self):
+        self.assert_summary(runs(101, "ERROR ERROR OK ERROR ERROR OK"), 1, 1, [2, 5])
+
+    def test_final_failure_retains_precedence_and_history(self):
+        result = self.assert_summary(runs(101, "ERROR OK ERROR"), 1, 1, [1], key="someFailed")
+        self.assertEqual(result["affected"][0]["when"], 3)
+        self.assertEqual(result["affected"][0]["classification"], "WARNING")
+
+    def test_other_guest_failure_retains_precedence(self):
+        self.assert_summary(runs(101, "ERROR OK ERROR OK") + runs(102, "ERROR"),
+                            1, 2, [1, 3], key="someFailed")
+
+    def test_running_and_unknown_do_not_replace_completed_run(self):
+        for status in ("running", "unknown", "", None):
+            with self.subTest(status=status):
+                tasks = runs(101, "ERROR OK") + [{**runs(101, "OK")[0], "starttime": 3, "status": status}]
+                self.assert_summary(tasks, 1, 1, [1], incomplete=True)
+
+    def test_upid_fallback_repeated_recovery(self):
+        tasks = runs(101, "ERROR OK ERROR OK")
+        for task in tasks:
+            del task["id"]
+        result = self.assert_summary(tasks, 1, 1, [1, 3])
+        self.assertEqual([r["vmid"] for r in result["affected"]], [101, 101])
+
+    def test_reverse_input_is_sorted_chronologically(self):
+        self.assert_summary(list(reversed(runs(101, "ERROR OK ERROR OK"))), 1, 1, [1, 3])
+
+    def test_zero_runs_and_nonbackup_tasks(self):
+        self.assertIsNone(evaluate([]))
+        self.assertIsNone(evaluate(runs(101, "ERROR", type="qmstart")))
+
+    def test_all_successful_has_no_recovery(self):
+        result = evaluate(runs(101, "OK OK") + runs(102, "OK"))
+        self.assertEqual(result["summary_key"], "allSucceeded")
+        self.assertEqual(result["summary_params"], {"total": "2"})
+        self.assertEqual(result["classification"], "CONFORMANT")
+        self.assertNotIn("affected", result)
+
+    def test_unverified_collection_and_inventory(self):
+        for response in ((1, "mock error"), (0, "not json"), (0, "{}"), (0, "[1]")):
+            with self.subTest(response=response):
+                self.assertEqual(evaluate(response=response), {
+                    "classification": "UNVERIFIED", "summary_key": "evaluationFailed",
+                    "incomplete": True, "evidence": response[1]})
+        result = evaluate(runs(101, "unknown"))
+        self.assertEqual(result["summary_key"], "evaluationFailed")
+        self.assertTrue(result["incomplete"])
+
+    def test_integer_and_string_ids_remain_distinct(self):
+        result = self.assert_summary(runs(101, "ERROR OK") + runs("101", "ERROR OK"), 2, 2, [1, 1])
+        self.assertEqual([r["vmid"] for r in result["affected"]], [101, "101"])
+
+    def test_vm_and_ct_type_suffixes_do_not_change_existing_identity(self):
+        tasks = runs(101, "ERROR OK", type="vzdump-qemu") + runs(102, "ERROR OK", type="vzdump-lxc")
+        self.assert_summary(tasks, 2, 2, [1, 1])
+        tasks = runs(101, "ERROR", type="vzdump-qemu") + [
+            {**runs(101, "OK", type="vzdump-lxc")[0], "starttime": 2}]
+        self.assert_summary(tasks, 1, 1, [1])
+
+    def test_unknown_and_multiguest_identity_are_not_reinterpreted(self):
+        # Existing identity contract: no expansion of a multi-guest task,
+        # no inference of real guest counts from a missing/nonnumeric ID.
+        for vmid, upid in ((None, "UPID:fixture:1:1:1:vzdump::root@pam:"),
+                           (None, "malformed"), ("101,102", "batch")):
+            with self.subTest(vmid=vmid, upid=upid):
+                result = self.assert_summary(runs(vmid, "ERROR OK", upid=upid), 1, 1, [1])
+                self.assertEqual(result["affected"][0]["vmid"], vmid)
+
+    def test_explicit_id_takes_precedence_over_upid(self):
+        tasks = runs(101, "ERROR OK", upid="UPID:fixture:1:1:1:vzdump:102:root@pam:")
+        result = self.assert_summary(tasks, 1, 1, [1])
+        self.assertEqual(result["affected"][0]["vmid"], 101)
+
+
+if __name__ == "__main__":
+    unittest.main()