diff --git a/aci-preupgrade-validation-script.py b/aci-preupgrade-validation-script.py index 0c6de21a..8d3b3f3e 100644 --- a/aci-preupgrade-validation-script.py +++ b/aci-preupgrade-validation-script.py @@ -5084,21 +5084,34 @@ def post_upgrade_cb_check(cversion, tversion, **kwargs): if skip_current_mo: continue created_by_mo = new_mo_dict[new_mo]['CreatedBy'] - api = "{}.json?rsp-subtree-include=count" - if new_mo == "compatSwitchHw": - # Expected to see suppBit in 32 or 64. Zero 32 means a failed postUpgradeCb. - api += '&query-target-filter=eq(compatSwitchHw.suppBit,"32")' + if new_mo == "fvSlaDef": + # Duplicate IPSLA DNs can inflate both moCount and plain class responses. + # icurl must finish pagination before counting unique DNs (issue #344). + new_mo_count = len(set( + mo[new_mo]['attributes']['dn'] + for mo in icurl('class', '{}.json'.format(new_mo)) + )) + created_by_mo_count = len(set( + mo[created_by_mo]['attributes']['dn'] + for mo in icurl('class', '{}.json'.format(created_by_mo)) + )) + else: + api = "{}.json?rsp-subtree-include=count" + if new_mo == "compatSwitchHw": + # Expected to see suppBit in 32 or 64. Zero 32 means a failed postUpgradeCb. + api += '&query-target-filter=eq(compatSwitchHw.suppBit,"32")' + + temp_new_mo_count = icurl("class", api.format(new_mo)) + new_mo_count = int(temp_new_mo_count[0]['moCount']['attributes']['count']) + if created_by_mo: + temp_createdby_mo_count = icurl('class', api.format(created_by_mo)) + created_by_mo_count = int(temp_createdby_mo_count[0]['moCount']['attributes']['count']) - temp_new_mo_count = icurl("class", api.format(new_mo)) - new_mo_count = int(temp_new_mo_count[0]['moCount']['attributes']['count']) if created_by_mo == "": if new_mo_count == 0: data.append([new_mo, new_mo_dict[new_mo]["Impact"]]) - else: - temp_createdby_mo_count = icurl('class', api.format(created_by_mo)) - created_by_mo_count = int(temp_createdby_mo_count[0]['moCount']['attributes']['count']) - if created_by_mo_count != new_mo_count: - data.append([new_mo, new_mo_dict[new_mo]["Impact"]]) + elif created_by_mo_count != new_mo_count: + data.append([new_mo, new_mo_dict[new_mo]["Impact"]]) if data: result = FAIL_O diff --git a/tests/checks/post_upgrade_cb_check/fvIPSLAMonitoringPol.json b/tests/checks/post_upgrade_cb_check/fvIPSLAMonitoringPol.json new file mode 100644 index 00000000..863b7d0a --- /dev/null +++ b/tests/checks/post_upgrade_cb_check/fvIPSLAMonitoringPol.json @@ -0,0 +1,26 @@ +[ + { + "fvIPSLAMonitoringPol": { + "attributes": { + "dn": "uni/tn-common/ipslaMonitoringPol-default", + "name": "default" + } + } + }, + { + "fvIPSLAMonitoringPol": { + "attributes": { + "dn": "uni/tn-example-a/ipslaMonitoringPol-monitor", + "name": "monitor" + } + } + }, + { + "fvIPSLAMonitoringPol": { + "attributes": { + "dn": "uni/tn-example-b/ipslaMonitoringPol-monitor", + "name": "monitor" + } + } + } +] diff --git a/tests/checks/post_upgrade_cb_check/fvSlaDef.json b/tests/checks/post_upgrade_cb_check/fvSlaDef.json new file mode 100644 index 00000000..f6bed46d --- /dev/null +++ b/tests/checks/post_upgrade_cb_check/fvSlaDef.json @@ -0,0 +1,26 @@ +[ + { + "fvSlaDef": { + "attributes": { + "dn": "uni/tn-common/slaDefPol-default", + "name": "default" + } + } + }, + { + "fvSlaDef": { + "attributes": { + "dn": "uni/tn-example-a/slaDefPol-monitor", + "name": "monitor" + } + } + }, + { + "fvSlaDef": { + "attributes": { + "dn": "uni/tn-example-b/slaDefPol-monitor", + "name": "monitor" + } + } + } +] diff --git a/tests/checks/post_upgrade_cb_check/test_post_upgrade_cb_check.py b/tests/checks/post_upgrade_cb_check/test_post_upgrade_cb_check.py index 79c26fa9..eeb1f4c0 100644 --- a/tests/checks/post_upgrade_cb_check/test_post_upgrade_cb_check.py +++ b/tests/checks/post_upgrade_cb_check/test_post_upgrade_cb_check.py @@ -16,8 +16,8 @@ mo1_new = "infraRsToImplicitSetPol.json?rsp-subtree-include=count" mo1_old = "infraImplicitSetPol.json?rsp-subtree-include=count" -mo2_new = "fvSlaDef.json?rsp-subtree-include=count" -mo2_old = "fvIPSLAMonitoringPol.json?rsp-subtree-include=count" +mo2_new = "fvSlaDef.json" +mo2_old = "fvIPSLAMonitoringPol.json" mo3_new = "infraRsConnectivityProfileOpt.json?rsp-subtree-include=count" mo3_old = "infraRsConnectivityProfile.json?rsp-subtree-include=count" @@ -32,11 +32,14 @@ # icurl output sets +ipsla_new = read_data(dir, "fvSlaDef.json") +ipsla_old = read_data(dir, "fvIPSLAMonitoringPol.json") + mo_count_pass = { mo1_new: read_data(dir, "moCount_10.json"), mo1_old: read_data(dir, "moCount_10.json"), - mo2_new: read_data(dir, "moCount_10.json"), - mo2_old: read_data(dir, "moCount_10.json"), + mo2_new: ipsla_new, + mo2_old: ipsla_old, mo3_new: read_data(dir, "moCount_10.json"), mo3_old: read_data(dir, "moCount_10.json"), mo4_new: read_data(dir, "moCount_10.json"), @@ -52,8 +55,8 @@ mo1_new: read_data(dir, "moCount_0.json"), mo1_old: read_data(dir, "moCount_0.json"), # Others are number mismatch - mo2_new: read_data(dir, "moCount_8.json"), - mo2_old: read_data(dir, "moCount_10.json"), + mo2_new: ipsla_new[:-1], + mo2_old: ipsla_old, mo3_new: read_data(dir, "moCount_8.json"), mo3_old: read_data(dir, "moCount_10.json"), mo4_new: read_data(dir, "moCount_8.json"), @@ -105,3 +108,102 @@ def test_logic(run_check, mock_icurl, cversion, tversion, expected_result): tversion=script.AciVersion(tversion) if tversion else None, ) assert result.result == expected_result + + +def assert_ipsla_result(run_check, expected_result): + result = run_check( + cversion=script.AciVersion("4.2(7v)"), + tversion=script.AciVersion("4.2(7v)"), + ) + assert result.result == expected_result + expected_headers = [] if expected_result == script.ERROR else ["Missed Objects", "Impact"] + assert result.headers == expected_headers + expected_data = [] + if expected_result == script.FAIL_O: + expected_data = [["fvSlaDef", "IPSLA monitor policy will not be deployed"]] + assert result.data == expected_data + return result + + +@pytest.mark.parametrize( + "new_objects, old_objects, expected_result", + [ + pytest.param(ipsla_new[:1] * 2, ipsla_old[:1], script.PASS, + id="issue-344-duplicate-default"), + pytest.param(ipsla_new + [ipsla_new[1]], ipsla_old, script.PASS, + id="duplicate-non-default"), + pytest.param(ipsla_new + [ipsla_new[0]], ipsla_old + [ipsla_old[2]], script.PASS, + id="duplicates-in-both-classes"), + pytest.param([ipsla_new[0], ipsla_new[0], ipsla_new[1]], ipsla_old, script.FAIL_O, + id="duplicate-masks-missing-object"), + pytest.param(ipsla_new[:-1], ipsla_old, script.FAIL_O, id="missing-new-object"), + pytest.param(ipsla_new, ipsla_old[:-1], script.FAIL_O, id="extra-new-object"), + pytest.param([], [], script.PASS, id="both-classes-empty"), + pytest.param([], ipsla_old[:1], script.FAIL_O, id="new-class-empty"), + pytest.param(ipsla_new[:1], [], script.FAIL_O, id="old-class-empty"), + pytest.param(ipsla_new[1:], ipsla_old[1:2], script.FAIL_O, + id="same-name-in-distinct-tenants"), + ], +) +def test_ipsla_unique_counts(run_check, mock_icurl, icurl_outputs, + new_objects, old_objects, expected_result): + icurl_outputs.update(mo_count_pass) + icurl_outputs.update({mo2_new: new_objects, mo2_old: old_objects}) + assert_ipsla_result(run_check, expected_result) + + +@pytest.mark.parametrize("missing_object", [False, True]) +def test_ipsla_pagination(run_check, mock_icurl, icurl_outputs, monkeypatch, missing_object): + # The default DN is repeated across pages. Raw lengths must drive pagination. + first_new_page = ipsla_new[:2] + second_new_page = [ipsla_new[0]] + ([] if missing_object else ipsla_new[2:]) + new_total = str(len(first_new_page) + len(second_new_page)) + icurl_outputs.update(mo_count_pass) + icurl_outputs.update({ + mo2_new: [ + {"totalCount": new_total, "imdata": first_new_page}, + {"totalCount": new_total, "imdata": second_new_page}, + ], + mo2_old: [ + {"totalCount": "3", "imdata": ipsla_old[:1]}, + {"totalCount": "3", "imdata": ipsla_old[1:]}, + ], + }) + mocked_query = script._icurl + queried_pages = [] + + def track_pages(apitype, query, page=0, page_size=100000): + queried_pages.append((query, page)) + return mocked_query(apitype, query, page, page_size) + + monkeypatch.setattr(script, "_icurl", track_pages) + expected_result = script.FAIL_O if missing_object else script.PASS + assert_ipsla_result(run_check, expected_result) + for query in (mo2_new, mo2_old): + assert (query, 0) in queried_pages + assert (query, 1) in queried_pages + + +@pytest.mark.parametrize("query, classname", [(mo2_new, "fvSlaDef"), (mo2_old, "fvIPSLAMonitoringPol")]) +@pytest.mark.parametrize("failure", ["missing-dn", "api-error", "incomplete-response"]) +def test_ipsla_query_errors(run_check, mock_icurl, icurl_outputs, query, classname, failure): + icurl_outputs.update(mo_count_pass) + if failure == "missing-dn": + icurl_outputs[query] = [{classname: {"attributes": {"name": "default"}}}] + elif failure == "api-error": + icurl_outputs[query] = [{"error": {"attributes": { + "code": "503", "text": "Unable to deliver the message, Resolve timeout", + }}}] + else: + icurl_outputs[query] = [ + {"totalCount": "2", "imdata": icurl_outputs[query][:1]}, + {"totalCount": "2", "imdata": []}, + ] + result = assert_ipsla_result(run_check, script.ERROR) + assert result.msg.startswith("Unexpected Error:") + if failure == "missing-dn": + assert "dn" in result.msg + elif failure == "api-error": + assert "API Timeout" in result.msg + else: + assert "API response empty with totalCount:2" in result.msg