20 bootflash fail over 50 if node already staged

aci-preupgrade-validation-script.py

@@ -56,6 +56,11 @@
 warnings.simplefilter(action='ignore', category=FutureWarning)
 
 
+class OldVerPropNotFound(Exception):
+    """ Later versions of ACI can have class properties not found in older versions """
+    pass
+
+
 class Connection(object):
     """
     Object built primarily for executing commands on Cisco IOS/NXOS devices.  The following
@@ -613,7 +618,10 @@ def icurl(apitype, query):
     logging.debug('response: ' + str(response))
     imdata = json.loads(response)['imdata']
     if imdata and "error" in imdata[0].keys():
-        raise Exception('API call failed! Check debug log')
+        if imdata and "not found in class" in imdata[0]['error']['attributes']['text']:
+            raise OldVerPropNotFound('cversion does not have requested property')
+        else:
+            raise Exception('API call failed! Check debug log')
     else:
         return imdata
 
@@ -984,33 +992,52 @@ def switch_group_guideline_check(index, total_checks, **kwargs):
     return result
 
 
-def switch_bootflash_usage_check(index, total_checks, **kwargs):
+def switch_bootflash_usage_check(index, total_checks, tversion, **kwargs):
     title = 'Switch Node /bootflash usage'
     result = FAIL_UF
     msg = ''
     headers = ["Pod-ID", "Node-ID", "Utilization", "Alert"]
     data = []
     print_title(title, index, total_checks)
 
-    response_json = icurl('class',
-                          'eqptcapacityFSPartition.json?query-target-filter=eq(eqptcapacityFSPartition.path,"/bootflash")')
-    if not response_json:
+    partitions_api =  'eqptcapacityFSPartition.json'
+    partitions_api += '?query-target-filter=eq(eqptcapacityFSPartition.path,"/bootflash")'
+
+    download_sts_api =  'maintUpgJob.json'
+    download_sts_api += '?query-target-filter=and(eq(maintUpgJob.dnldStatus,"downloaded")' 
+    download_sts_api += ',eq(maintUpgJob.desiredVersion,"n9000-1{}"))'.format(tversion)
+
+    partitions = icurl('class', partitions_api)
+    if not partitions:
         result = ERROR
         msg = 'bootflash objects not found'
 
-    for eqptcapacityFSPartition in response_json:
+    predownloaded_nodes = []
+    try:
+        download_sts = icurl('class', download_sts_api)
+    except OldVerPropNotFound:
+        # Older versions don't have 'dnldStatus' param
+        download_sts = []
+
+    for maintUpgJob in download_sts:
+        dn = re.search(node_regex, maintUpgJob['maintUpgJob']['attributes']['dn'])
+        node = dn.group("node")
+        predownloaded_nodes.append(node)
+
+    for eqptcapacityFSPartition in partitions:
         dn = re.search(node_regex, eqptcapacityFSPartition['eqptcapacityFSPartition']['attributes']['dn'])
         pod = dn.group("pod")
         node = dn.group("node")
         avail = int(eqptcapacityFSPartition['eqptcapacityFSPartition']['attributes']['avail'])
         used = int(eqptcapacityFSPartition['eqptcapacityFSPartition']['attributes']['used'])
 
         usage = (used / (avail + used)) * 100
-        if usage >= 50:
+        if (usage >= 50) and (node not in predownloaded_nodes):
             data.append([pod, node, usage, "Over 50% usage! Contact Cisco TAC for Support"])
+
     if not data:
         result = PASS
-        msg = 'all below 50%'
+        msg = 'All below 50% or pre-downloaded'
     print_result(title, result, msg, headers, data)
     return result
 

tests/conftest.py

@@ -41,6 +41,14 @@ def mock_icurl(monkeypatch, icurl_outputs):
     def _mock_icurl(apitype, query):
         if icurl_outputs.get(query) is None:
             log.error("Query `%s` not found in test data", query)
-        return icurl_outputs.get(query, [])
+
+        imdata = icurl_outputs.get(query, [])
+        if imdata and "error" in imdata[0].keys():
+            if "not found in class"  in imdata[0]['error']['attributes']['text']:
+                raise script.OldVerPropNotFound('cversion does not have requested property')
+            else:
+                raise Exception('API call failed! Check debug log')
+        else:
+            return imdata
 
     monkeypatch.setattr(script, "icurl", _mock_icurl)

tests/switch_bootflash_usage_check/maintUpgJob_not_downloaded.json

@@ -0,0 +1 @@
+[]

tests/switch_bootflash_usage_check/maintUpgJob_old_ver_no_prop.json

@@ -0,0 +1,10 @@
+[
+  {
+      "error": {
+          "attributes": {
+              "code": "121",
+              "text": "Prop 'dnldStatus' not found in class 'maintUpgJob' property table"
+          }
+      }
+  }
+]

tests/switch_bootflash_usage_check/maintUpgJob_pre_downloaded.json

@@ -0,0 +1,16 @@
+[
+  {
+    "maintUpgJob": {
+      "attributes": {
+        "desiredVersion": "n9000-16.0(2h)",
+        "dn": "topology/pod-1/node-101/sys/fwstatuscont/upgjob",
+        "dnldPercent": "100",
+        "dnldStatus": "downloaded",
+        "startDate": "2023-11-16T10:15:22.894-08:00",
+        "status": "",
+        "upgradeStatus": "scheduled",
+        "upgradeStatusStr": "Scheduled"
+      }
+    }
+  }
+]

tests/switch_bootflash_usage_check/test_switch_bootflash_usage_check.py

@@ -11,18 +11,36 @@
 
 
 # icurl queries
-partitions = 'eqptcapacityFSPartition.json?query-target-filter=eq(eqptcapacityFSPartition.path,"/bootflash")'
+partitions =  'eqptcapacityFSPartition.json' 
+partitions += '?query-target-filter=eq(eqptcapacityFSPartition.path,"/bootflash")'
 
+download_sts =  'maintUpgJob.json'
+download_sts += '?query-target-filter=and(eq(maintUpgJob.dnldStatus,"downloaded")' 
+download_sts += ',eq(maintUpgJob.desiredVersion,"n9000-16.0(2h)"))'
 
 @pytest.mark.parametrize(
-    "icurl_outputs, expected_result",
+    "icurl_outputs, tversion, expected_result",
     [
         (
-            {partitions: read_data(dir, "eqptcapacityFSPartition_pos.json")},
+            {partitions: read_data(dir, "eqptcapacityFSPartition.json"),
+            download_sts: read_data(dir, "maintUpgJob_not_downloaded.json")},
+            "6.0(2h)",
+            script.FAIL_UF,
+        ),
+        (
+            {partitions: read_data(dir, "eqptcapacityFSPartition.json"),
+            download_sts: read_data(dir, "maintUpgJob_pre_downloaded.json")},
+            "6.0(2h)",
+            script.PASS,
+        ),
+        (
+            {partitions: read_data(dir, "eqptcapacityFSPartition.json"),
+            download_sts: read_data(dir, "maintUpgJob_old_ver_no_prop.json")},
+            "6.0(2h)",
             script.FAIL_UF,
         ),
     ],
 )
-def test_logic(mock_icurl, expected_result):
-    result = script.switch_bootflash_usage_check(1, 1)
+def test_logic(mock_icurl, tversion, expected_result):
+    result = script.switch_bootflash_usage_check(1, 1, script.AciVersion(tversion))
     assert result == expected_result