ProxMenux/AppImage/scripts/tests/test_audit_backup_recovered_count.py

"""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()