feat: Create all result files as part of prepare (#283)

Author: takishida

aci-preupgrade-validation-script.py

@@ -929,6 +929,11 @@ class AciResult:
         "docUrl", "severity", "ruleStatus", "showValidation", "failureDetails",
     )
 
+    # ruleStatus
+    IN_PROGRESS = "in-progress"
+    PASS = "passed"
+    FAIL = "failed"
+
     def __init__(self, func_name, name, description):
         self.ruleId = func_name
         self.name = name
@@ -938,14 +943,18 @@ def __init__(self, func_name, name, description):
         self.recommended_action = ""
         self.docUrl = ""
         self.severity = "informational"
-        self.ruleStatus = "passed"  # passed|failed
+        self.ruleStatus = AciResult.IN_PROGRESS
         self.showValidation = True
         self.failureDetails = {
             "failType": "",
             "data": [],
             "unformatted_data": [],
         }
 
+    @property
+    def filename(self):
+        return re.sub(r'[^a-zA-Z0-9_]+|\s+', '_', self.ruleId) + '.json'
+
     @staticmethod
     def craftData(column, rows):
         if not (isinstance(rows, list) and isinstance(column, list)):
@@ -979,8 +988,9 @@ def updateWithResults(self, result, recommended_action, msg, doc_url, headers, d
         elif result in [MANUAL]:
             self.severity = "warning"
 
+        self.ruleStatus = AciResult.PASS
         if result not in [NA, PASS]:
-            self.ruleStatus = "failed"
+            self.ruleStatus = AciResult.FAIL
             self.failureDetails["failType"] = result
             self.failureDetails["data"] = self.craftData(headers, data)
             if unformatted_headers and unformatted_data:
@@ -996,12 +1006,11 @@ def buildResult(self):
         return {slot: getattr(self, slot) for slot in self.__slots__}
 
     def writeResult(self, path=JSON_DIR):
-        filename = re.sub(r'[^a-zA-Z0-9_]+|\s+', '_', self.ruleId) + '.json'
         if not os.path.isdir(path):
             os.mkdir(path)
-        with open(os.path.join(path, filename), "w") as f:
+        with open(os.path.join(path, self.filename), "w") as f:
             json.dump(self.buildResult(), f, indent=2)
-        return "{}/{}".format(path, filename)
+        return "{}/{}".format(path, self.filename)
 
 
 class Result:
@@ -1034,6 +1043,12 @@ def check_wrapper(check_title):
     def decorator(check_func):
         @functools.wraps(check_func)
         def wrapper(index, total_checks, *args, **kwargs):
+            # When init is True, we just initialize the result file and return
+            if kwargs.get("init") is True:
+                synth = AciResult(wrapper.__name__, check_title, "")
+                synth.writeResult()
+                return None
+
             # Print `[Check  1/81] <title>...`
             print_title(check_title, index, total_checks)
 
@@ -5302,6 +5317,7 @@ def apic_database_size_check(cversion, **kwargs):
 
 # ---- Script Execution ----
 
+
 def parse_args(args):
     parser = ArgumentParser(description="ACI Pre-Upgrade Validation Script - %s" % SCRIPT_VERSION)
     parser.add_argument("-t", "--tversion", action="store", type=str, help="Upgrade Target Version. Ex. 6.2(1a)")
@@ -5330,10 +5346,14 @@ def initialize():
     logging.basicConfig(level=logging.DEBUG, filename=LOG_FILE, format=fmt, datefmt='%Y-%m-%d %H:%M:%S')
 
 
-def prepare(api_only, arg_tversion, arg_cversion, total_checks):
+def prepare(api_only, arg_tversion, arg_cversion, checks):
     prints('    ==== %s%s, Script Version %s  ====\n' % (ts, tz, SCRIPT_VERSION))
     prints('!!!! Check https://github.com/datacenter/ACI-Pre-Upgrade-Validation-Script for Latest Release !!!!\n')
 
+    # Create empty result files for all checks
+    for idx, check in enumerate(checks):
+        check(idx + 1, len(checks), init=True)
+
     username = password = None
     if not api_only:
         username, password = get_credentials()
@@ -5359,7 +5379,7 @@ def prepare(api_only, arg_tversion, arg_cversion, total_checks):
         "tversion": str(tversion),
         "sw_cversion": str(sw_cversion),
         "api_only": api_only,
-        "total_checks": total_checks,
+        "total_checks": len(checks),
     }
     with open(META_FILE, "w") as f:
         json.dump(metadata, f, indent=2)
@@ -5530,7 +5550,7 @@ def main(_args=None):
         return
 
     initialize()
-    inputs = prepare(args.api_only, args.tversion, args.cversion, len(checks))
+    inputs = prepare(args.api_only, args.tversion, args.cversion, checks)
     run_checks(checks, inputs)
     wrapup(args.no_cleanup)
 

tests/test_prepare.py

@@ -2,9 +2,11 @@
 import importlib
 import logging
 import json
+import os
 
 script = importlib.import_module("aci-preupgrade-validation-script")
 AciVersion = script.AciVersion
+AciResult = script.AciResult
 
 
 @pytest.fixture(autouse=True)
@@ -133,7 +135,7 @@ def _mock_get_target_version(arg_tversion):
             "ave_eol_check",
             {"username": "admin", "password": "mypassword", "cversion": AciVersion("6.0(4d)"), "tversion": AciVersion("6.1(4a)"), "sw_cversion": AciVersion("6.0(9d)"), "vpc_node_ids": ["101", "102"]},
         ),
-        # veresions are switch syntax
+        # versions are switch syntax
         # The version `get_target_version()` is ignored.
         (
             {
@@ -147,7 +149,7 @@ def _mock_get_target_version(arg_tversion):
             "ave_eol_check",
             {"username": "admin", "password": "mypassword", "cversion": AciVersion("6.0(4d)"), "tversion": AciVersion("6.1(4a)"), "sw_cversion": AciVersion("6.0(9d)"), "vpc_node_ids": ["101", "102"]},
         ),
-        # veresions are switch or APIC syntax
+        # versions are switch or APIC syntax
         # The version `get_target_version()` is ignored.
         (
             {
@@ -164,15 +166,31 @@ def _mock_get_target_version(arg_tversion):
     ],
 )
 def test_prepare(mock_icurl, api_only, arg_tversion, arg_cversion, debug_function, expected_result):
+    script.initialize()
     checks = script.get_checks(api_only, debug_function)
-    inputs = script.prepare(api_only, arg_tversion, arg_cversion, len(checks))
+    inputs = script.prepare(api_only, arg_tversion, arg_cversion, checks)
     for key, value in expected_result.items():
-        if "version" in key:
+        if "version" in key:  # cversion or tversion
             assert isinstance(inputs[key], AciVersion)
             assert str(inputs[key]) == str(value)
         else:
             assert inputs[key] == value
 
+    result_files = os.listdir(script.JSON_DIR)
+    # Result files should be created for all checks
+    assert len(result_files) == len(checks)
+    for check in checks:
+        # Rule name is known only through the wrapper `check_wrapper`.
+        # Rule name content should be checked via another unit test.
+        # Use AciResult class here just to get the filename from `check.__name__`.
+        ar = AciResult(check.__name__, "unknown_name", "")
+        file_path = os.path.join(script.JSON_DIR, ar.filename)
+        assert os.path.exists(file_path), "Missing result file: {}".format(file_path)
+        with open(file_path, "r") as f:
+            result = json.load(f)
+        assert result["ruleId"] == check.__name__
+        assert result["ruleStatus"] == AciResult.IN_PROGRESS
+
     with open(script.META_FILE, "r") as f:
         meta = json.load(f)
         assert meta["name"] == "PreupgradeCheck"
@@ -191,13 +209,13 @@ def test_prepare(mock_icurl, api_only, arg_tversion, arg_cversion, debug_functio
 def test_tversion_invald():
     with pytest.raises(SystemExit):
         with pytest.raises(ValueError):
-            script.prepare(False, "invalid_version", "6.0(1a)", 1)
+            script.prepare(False, "invalid_version", "6.0(1a)", [])
 
 
 def test_cversion_invald():
     with pytest.raises(SystemExit):
         with pytest.raises(ValueError):
-            script.prepare(False, "6.0(1a)", "invalid_version", 1)
+            script.prepare(False, "6.0(1a)", "invalid_version", [])
 
 
 @pytest.mark.parametrize(
@@ -264,7 +282,46 @@ def test_prepare_exception(capsys, caplog, mock_icurl, api_only, arg_tversion, a
     with pytest.raises(SystemExit):
         with pytest.raises(Exception):
             checks = script.get_checks(api_only, debug_function)
-            script.prepare(api_only, arg_tversion, arg_cversion, len(checks))
+            script.prepare(api_only, arg_tversion, arg_cversion, checks)
     captured = capsys.readouterr()
     print(captured.out)
     assert captured.out.endswith(expected_result)
+
+
+# Unit test focusing only on the result file creation
+def test_prepare_initial_result_files(mock_icurl, icurl_outputs):
+    # Provide required API outputs used inside prepare()
+    icurl_outputs.update({
+        "firmwareCtrlrRunning.json": outputs["cversion"],
+        "firmwareRunning.json": outputs["switch_version"],
+        "fabricNodePEp.json": outputs["vpc_nodes"],
+    })
+
+    # Create two simple checks with known titles
+    @script.check_wrapper(check_title="Prepare Check A")
+    def prep_check_a(**kwargs):
+        return script.Result(result=script.PASS)
+
+    @script.check_wrapper(check_title="Prepare Check B")
+    def prep_check_b(**kwargs):
+        return script.Result(result=script.PASS)
+
+    checks = [prep_check_a, prep_check_b]
+
+    # Run prepare which should only initialize result files
+    script.prepare(api_only=False, arg_tversion=None, arg_cversion=None, checks=checks)
+
+    # Verify result files and contents
+    expected = {
+        "prep_check_a": "Prepare Check A",
+        "prep_check_b": "Prepare Check B",
+    }
+    for func_name, title in expected.items():
+        ar = AciResult(func_name, title, "")
+        file_path = os.path.join(script.JSON_DIR, ar.filename)
+        assert os.path.exists(file_path), "Missing result file: {}".format(file_path)
+        with open(file_path, "r") as f:
+            data = json.load(f)
+        assert data["ruleId"] == func_name
+        assert data["name"] == title
+        assert data["ruleStatus"] == AciResult.IN_PROGRESS