diff --git a/tests/transceiver/cdb_firmware_upgrade/conftest.py b/tests/transceiver/cdb_firmware_upgrade/conftest.py index 57da63b7034..7084844a2b5 100644 --- a/tests/transceiver/cdb_firmware_upgrade/conftest.py +++ b/tests/transceiver/cdb_firmware_upgrade/conftest.py @@ -3,6 +3,7 @@ import pytest +from tests.common.platform.interface_utils import get_physical_port_indices from tests.transceiver.attribute_parser.attribute_keys import ( CDB_FIRMWARE_UPGRADE_ATTRIBUTES_KEY, EEPROM_ATTRIBUTES_KEY, @@ -24,6 +25,10 @@ resolve_ports_under_test, select_attribute_ports, ) +from tests.transceiver.cdb_firmware_upgrade.firmware_operations import ( + restore_module_to_original, + run_firmware_op_on_ports, +) CMIS_CDB_FIRMWARE_BASE_PATH_ON_DUT = "/tmp/cmis_cdb_firmware" CMIS_CDB_FIRMWARE_PRESTAGED_PATH_ON_DUT = "/host/cmis_cdb_firmware" @@ -58,18 +63,18 @@ def transceiver_firmware_info_parser(): @pytest.fixture(scope="session") def required_firmware_metadata_for_all_transceivers( - get_dev_transceiver_details, + port_attributes_dict, transceiver_firmware_info_parser, - get_transceiver_common_attributes + cdb_firmware_qualifying_ports, ): return get_required_firmware_metadata_for_all_transceivers( - get_dev_transceiver_details, + port_attributes_dict, transceiver_firmware_info_parser.transceiver_firmware_info, - transceiver_common_attributes=get_transceiver_common_attributes + cdb_firmware_qualifying_ports, ) -@pytest.fixture(scope="module", autouse=True) +@pytest.fixture(scope="session", autouse=True) def stage_latest_firmware_binaries_on_dut( duthost, transceiver_firmware_info_parser, @@ -106,17 +111,16 @@ def stage_latest_firmware_binaries_on_dut( logger.info("All latest firmware staged to {}".format(CMIS_CDB_FIRMWARE_BASE_PATH_ON_DUT)) -@pytest.fixture(scope="module") -def cdb_firmware_qualifying_ports( - port_attributes_dict, lport_to_first_subport_mapping, get_lport_to_pport_mapping -): +@pytest.fixture(scope="session") +def cdb_firmware_qualifying_ports(duthost, port_attributes_dict, lport_to_first_subport_mapping): """CMIS active-optical first-subport ports the CDB firmware tests run on. Selection is the CDB attribute category gated by the EEPROM ``cmis_active_optical`` flag and restricted to any configured ``ports_under_test``. """ + lport_to_pport = get_physical_port_indices(duthost) explicit_ports = resolve_ports_under_test( - get_lport_to_pport_mapping, port_attributes_dict, CDB_FIRMWARE_UPGRADE_ATTRIBUTES_KEY + lport_to_pport, port_attributes_dict, CDB_FIRMWARE_UPGRADE_ATTRIBUTES_KEY ) qualifying_ports = select_attribute_ports( port_attributes_dict, @@ -139,10 +143,8 @@ def cdb_firmware_qualifying_ports( def dom_polling_disabled(duthost, port_attributes_dict, cdb_firmware_qualifying_ports): """Disable DOM polling on the ports under test, restoring it on teardown. - Scope is intentionally function (the default): upcoming firmware-operation - tests re-validate DOM values after each test, which requires DOM to be - re-enabled between tests. Function scope gives a per-test - disable -> yield -> re-enable cycle. + Firmware operation tests re-validate DOM values after each test, which + requires DOM to be re-enabled between tests. """ sleep_sec = 0 disabled_ports = [] @@ -176,12 +178,38 @@ def dom_polling_disabled(duthost, port_attributes_dict, cdb_firmware_qualifying_ logger.info("Re-enabled DOM polling on %d port(s)", len(disabled_ports)) -@pytest.fixture(scope="module", autouse=True) +@pytest.fixture(scope="session", autouse=True) +def restore_original_firmware_baseline( + stage_latest_firmware_binaries_on_dut, firmware_files_cleanup, duthost, port_attributes_dict, + cdb_firmware_qualifying_ports, required_firmware_metadata_for_all_transceivers, +): + """Restore every qualifying module to its original state before and + after the session. Depends on ``firmware_files_cleanup`` so the + post-session restore runs before the staged binaries are removed. + """ + lport_to_pport = get_physical_port_indices(duthost) + + def _restore(phase): + failures, ports = run_firmware_op_on_ports( + duthost, port_attributes_dict, cdb_firmware_qualifying_ports, + lport_to_pport, required_firmware_metadata_for_all_transceivers, + restore_module_to_original, + ) + logger.info("%s original firmware baseline on %d port(s)", phase, ports) + if failures: + pytest.fail(f"{phase} firmware restore failures:\n" + "\n".join(failures)) + + _restore("Pre-session") + yield + _restore("Post-session") + + +@pytest.fixture(scope="session", autouse=True) def firmware_files_cleanup( duthost ): """ - Module-scoped cleanup fixture that removes firmware files after all tests in the module complete. + Session-scoped cleanup fixture that removes firmware files after all tests in the session complete. """ yield # This is where all tests run diff --git a/tests/transceiver/cdb_firmware_upgrade/firmware_operations.py b/tests/transceiver/cdb_firmware_upgrade/firmware_operations.py new file mode 100644 index 00000000000..d934d190eb7 --- /dev/null +++ b/tests/transceiver/cdb_firmware_upgrade/firmware_operations.py @@ -0,0 +1,374 @@ +import logging + +from tests.common.platform.interface_utils import ( + get_physical_to_logical_port_mapping, + wait_ports_oper_status, +) +from tests.transceiver.attribute_parser.attribute_keys import ( + BASE_ATTRIBUTES_KEY, + CDB_FIRMWARE_UPGRADE_ATTRIBUTES_KEY, + SYSTEM_ATTRIBUTES_KEY, +) +from tests.transceiver.common import cli_helpers, dmesg_helpers, scenario_ops +from tests.transceiver.common.cli_parser_helper import ( + FW_ACTIVE, + FW_COMMITTED_IMAGE, + FW_INACTIVE, + FW_RUNNING_IMAGE, +) + +logger = logging.getLogger(__name__) + +I2C_ERROR_PATTERN = r"i2c.*(error|fail|timeout|nack)|(error|fail).*i2c" +THERMALCTLD = "thermalctld" + + +def select_target_version(firmware_versions, banks): + """Select the next version, after the active one, that is in neither bank.""" + active = banks.get(FW_ACTIVE, "") + inactive = banks.get(FW_INACTIVE, "") + start = firmware_versions.index(active) + 1 if active in firmware_versions else 0 + for offset in range(len(firmware_versions)): + next_version = firmware_versions[(start + offset) % len(firmware_versions)] + if next_version not in (active, inactive): + return next_version + return firmware_versions[0] + + +def _verify_bank_image_fields(after_banks, before_banks=None): + if before_banks is not None: + failures = [] + for field in (FW_RUNNING_IMAGE, FW_COMMITTED_IMAGE): + if after_banks.get(field) != before_banks.get(field): + failures.append( + f"{field} changed from {before_banks.get(field)} to " + f"{after_banks.get(field)} after download" + ) + return failures + running = after_banks.get(FW_RUNNING_IMAGE) + committed = after_banks.get(FW_COMMITTED_IMAGE) + if running != committed: + return [f"Committed Image {committed} != Running Image {running} after activation"] + return [] + + +def _stop_thermalctld(duthost): + status, _ = duthost.get_pmon_daemon_status(THERMALCTLD) + if status is None: + return False, "could not determine thermalctld status" + if status != "RUNNING": + return False, None + duthost.stop_pmon_daemon_service(THERMALCTLD) + status, _ = duthost.get_pmon_daemon_status(THERMALCTLD) + if status == "RUNNING": + return False, "thermalctld remained running after stop" + return True, None + + +def _start_thermalctld(duthost, was_stopped): + if not was_stopped: + return None + duthost.start_pmon_daemon(THERMALCTLD) + status, _ = duthost.get_pmon_daemon_status(THERMALCTLD) + if status != "RUNNING": + return f"thermalctld status after start is {status or 'unknown'}" + return None + + +def _scan_i2c_errors(duthost, dmesg_start_uptime, operation): + """Return per-port failures for new I2C dmesg errors.""" + i2c_errors, dmesg_err = dmesg_helpers.scan_new_dmesg_errors( + duthost, dmesg_start_uptime, set(), I2C_ERROR_PATTERN + ) + if dmesg_err: + return [dmesg_err] + if i2c_errors: + return [f"I2C error(s) in dmesg during {operation}: {'; '.join(i2c_errors[:3])}"] + return [] + + +def resolve_binary_path(metadata_map, vendor, pn, version): + """Return the staged on-DUT path for ``(vendor, pn, version)``.""" + for entry in metadata_map[(vendor, pn)]: + if entry["version"] == version: + return entry["dut_path"] + return None + + +def verify_firmware_downloaded(duthost, port, before_banks, target_version, download_err): + """Active/Running/Committed banks unchanged, the inactive bank has ``target_version``.""" + if download_err: + return [f"download failed: {download_err}"] + + after_banks, err = cli_helpers.sfputil_show_fwversion(duthost, port) + if err: + return [err] + + failures = [] + if after_banks.get(FW_ACTIVE) != before_banks.get(FW_ACTIVE): + failures.append( + f"active firmware changed from {before_banks.get(FW_ACTIVE)} to " + f"{after_banks.get(FW_ACTIVE)} after download" + ) + if after_banks.get(FW_INACTIVE) != target_version: + failures.append( + f"inactive firmware {after_banks.get(FW_INACTIVE) or 'N/A'} != " + f"downloaded {target_version}" + ) + failures += _verify_bank_image_fields(after_banks, before_banks=before_banks) + return failures + + +def perform_firmware_download(duthost, port, port_context, metadata_map, + target_version=None, expect_link_up=True): + """Download firmware to ``port`` and verify the firmware downloaded successfully. + + Returns a list of per-port failure strings (empty on success). + """ + cdb_attrs = port_context["cdb_attrs"] + system_attrs = port_context["system_attrs"] + vendor, pn = port_context["vendor"], port_context["pn"] + physical_index = port_context["physical_index"] + + before_banks, err = cli_helpers.sfputil_show_fwversion(duthost, port) + if err: + return [err] + + if target_version is None: + target_version = select_target_version( + cdb_attrs.get("firmware_versions"), before_banks, + ) + + fwfile = resolve_binary_path(metadata_map, vendor, pn, target_version) + + startup_wait = system_attrs.get("port_startup_wait_sec", 60) + if expect_link_up: + link_failures = wait_ports_oper_status(duthost, [port], "up", startup_wait) + if link_failures: + return ["port must be operationally up before download"] + link_failures + + if cdb_attrs.get("firmware_download_cdb_abort_support", True): + status, abort_err = cli_helpers.issue_cdb_fw_abort(duthost, physical_index) + if abort_err: + logger.warning("Port %s: pre-download CDB abort failed (proceeding): %s", port, abort_err) + else: + logger.info("Port %s: pre-download CDB abort status=%s", port, status) + + thermalctld_stopped = False + if cdb_attrs.get("thermalctld_disabling_required", False): + thermalctld_stopped, thermal_err = _stop_thermalctld(duthost) + if thermal_err: + return [f"failed to stop thermalctld: {thermal_err}"] + + failures = [] + try: + dmesg_start_uptime, dmesg_start_err = dmesg_helpers.capture_dmesg_uptime_watermark(duthost) + if dmesg_start_err: + failures.append(dmesg_start_err) + else: + timeout_sec = cdb_attrs.get("firmware_download_timeout_minutes", 30) * 60 + elapsed, dl_err = cli_helpers.sfputil_firmware_download(duthost, port, fwfile, timeout_sec) + logger.info("Port %s: firmware download %s took %ss", port, target_version, elapsed) + + failures += verify_firmware_downloaded( + duthost, port, before_banks, target_version, dl_err, + ) + if expect_link_up and not dl_err: + failures += wait_ports_oper_status(duthost, [port], "up", startup_wait) + + failures += _scan_i2c_errors(duthost, dmesg_start_uptime, "download") + finally: + thermal_err = _start_thermalctld(duthost, thermalctld_stopped) + if thermal_err: + failures.append(f"failed to restart thermalctld: {thermal_err}") + return failures + + +def verify_firmware_activation(duthost, port, before_banks, dual_bank_supported, + activated_version=None): + """Verifies the bank swap took effect.""" + after_banks, err = cli_helpers.sfputil_show_fwversion(duthost, port) + if err: + return [err] + + failures = [] + if dual_bank_supported: + if after_banks.get(FW_ACTIVE) != before_banks.get(FW_INACTIVE): + failures.append("active firmware not updated after activation") + if after_banks.get(FW_INACTIVE) != before_banks.get(FW_ACTIVE): + failures.append("previous active firmware not preserved in inactive bank after swap") + if after_banks.get(FW_RUNNING_IMAGE) == before_banks.get(FW_RUNNING_IMAGE): + failures.append("Running Image did not change after activation") + elif activated_version is None: + failures.append("activated_version is required for a single-bank module") + elif after_banks.get(FW_ACTIVE) != activated_version: + failures.append(f"active firmware {after_banks.get(FW_ACTIVE)} != activated {activated_version}") + + failures += _verify_bank_image_fields(after_banks) + + return failures + + +def perform_firmware_activation(duthost, port, port_context, + before_banks=None, activated_version=None): + """Activate the inactive-bank firmware on ``port``. + + Returns a list of per-port failure strings (empty on success). + """ + cdb_attrs = port_context["cdb_attrs"] + system_attrs = port_context["system_attrs"] + subports = port_context["subports"] + dual_bank = cdb_attrs.get("dual_bank_supported", True) + run_timeout = cdb_attrs.get("firmware_run_timeout_sec", 20) + commit_timeout = cdb_attrs.get("firmware_commit_timeout_sec", 10) + recover_sec = system_attrs.get("transceiver_reset_i2c_recover_sec", 5) + startup_wait = system_attrs.get("port_startup_wait_sec", 60) + shutdown_wait = system_attrs.get("port_shutdown_wait_sec", 5) + + if before_banks is None: + before_banks, err = cli_helpers.sfputil_show_fwversion(duthost, port) + if err: + return [err] + + failures = scenario_ops.perform_ports_shutdown(duthost, subports, shutdown_wait) + thermalctld_stopped = False + dmesg_start_uptime = None + try: + if not failures and cdb_attrs.get("thermalctld_disabling_required", False): + thermalctld_stopped, thermal_err = _stop_thermalctld(duthost) + if thermal_err: + failures.append(f"failed to stop thermalctld: {thermal_err}") + + if not failures: + dmesg_start_uptime, dmesg_start_err = dmesg_helpers.capture_dmesg_uptime_watermark(duthost) + if dmesg_start_err: + failures.append(dmesg_start_err) + + if not failures: + run_elapsed, run_err = cli_helpers.sfputil_firmware_run(duthost, port, run_timeout) + logger.info("Port %s: firmware run took %ss", port, run_elapsed) + if run_err: + failures.append(f"firmware run failed: {run_err}") + + if not failures: + commit_elapsed, commit_err = cli_helpers.sfputil_firmware_commit(duthost, port, commit_timeout) + logger.info("Port %s: firmware commit took %ss", port, commit_elapsed) + if commit_err: + failures.append(f"firmware commit failed: {commit_err}") + + if not failures: + failures += _scan_i2c_errors(duthost, dmesg_start_uptime, "activation") + + thermal_err = _start_thermalctld(duthost, thermalctld_stopped) + if thermal_err: + failures.append(f"failed to restart thermalctld: {thermal_err}") + else: + thermalctld_stopped = False + + if not failures: + failures += scenario_ops.perform_sfputil_reset( + duthost, port, recover_with_port_toggle=False, i2c_recover_sec=recover_sec, + ) + finally: + thermal_err = _start_thermalctld(duthost, thermalctld_stopped) + if thermal_err: + failures.append(f"failed to restart thermalctld: {thermal_err}") + startup_failures = scenario_ops.perform_ports_startup(duthost, subports, startup_wait) + failures += startup_failures + + if not failures: + failures += verify_firmware_activation( + duthost, port, before_banks, dual_bank, activated_version=activated_version, + ) + return failures + + +def activation_op(duthost, port, port_context, metadata_map): + """``run_firmware_op_on_ports`` per-port op: activate selected firmware.""" + cdb_attrs = port_context["cdb_attrs"] + if not cdb_attrs.get("dual_bank_supported", True): + banks, err = cli_helpers.sfputil_show_fwversion(duthost, port) + if err: + return [err] + target_version = select_target_version( + cdb_attrs.get("firmware_versions"), banks, + ) + failures = perform_firmware_download( + duthost, port, port_context, metadata_map, target_version=target_version, + ) + if failures: + return failures + return perform_firmware_activation( + duthost, port, port_context, activated_version=target_version, + ) + return perform_firmware_activation(duthost, port, port_context) + + +def restore_module_to_original(duthost, port, port_context, metadata_map): + """Restore a module to its original state.""" + cdb_attrs = port_context["cdb_attrs"] + system_attrs = port_context["system_attrs"] + gold_firmware = cdb_attrs.get("gold_firmware_version") + dual_bank = cdb_attrs.get("dual_bank_supported", True) + + failures = [] + banks, err = cli_helpers.sfputil_show_fwversion(duthost, port) + if err: + return [err] + + if banks.get(FW_ACTIVE) != gold_firmware: + if not dual_bank or banks.get(FW_INACTIVE) != gold_firmware: + failures += perform_firmware_download( + duthost, port, port_context, metadata_map, + target_version=gold_firmware, expect_link_up=False, + ) + if not failures: + failures += perform_firmware_activation( + duthost, port, port_context, activated_version=gold_firmware, + ) + if not failures and dual_bank: + banks, err = cli_helpers.sfputil_show_fwversion(duthost, port) + if err: + failures.append(err) + + if not failures and dual_bank: + orig_inactive = cdb_attrs.get("inactive_firmware_version") + if not orig_inactive: + failures.append("inactive_firmware_version is not defined for dual-bank module") + elif banks.get(FW_INACTIVE) != orig_inactive: + failures += perform_firmware_download( + duthost, port, port_context, metadata_map, + target_version=orig_inactive, expect_link_up=False, + ) + + failures += scenario_ops.perform_ports_startup( + duthost, port_context["subports"], + system_attrs.get("port_startup_wait_sec", 60), + ) + return failures + + +def run_firmware_op_on_ports(duthost, port_attributes_dict, qualifying_ports, lport_to_pport, + metadata_map, per_port_op): + """Run ``per_port_op`` on every qualifying CDB firmware port and aggregate failures. + + Returns ``(all_failures, num_ports)``, the caller ``pytest.fail``s or logs. + """ + pport_to_lport = get_physical_to_logical_port_mapping(lport_to_pport) + all_failures = [] + for port in qualifying_ports: + base_attrs = port_attributes_dict[port].get(BASE_ATTRIBUTES_KEY, {}) + vendor = base_attrs.get("normalized_vendor_name") + pn = base_attrs.get("normalized_vendor_pn") + physical_index = lport_to_pport.get(port) + if physical_index is None: + all_failures.append(f"{port}: could not resolve physical port index") + continue + port_context = { + "cdb_attrs": port_attributes_dict[port].get(CDB_FIRMWARE_UPGRADE_ATTRIBUTES_KEY, {}), + "system_attrs": port_attributes_dict[port].get(SYSTEM_ATTRIBUTES_KEY, {}), + "vendor": vendor, "pn": pn, "physical_index": physical_index, + "subports": pport_to_lport.get(physical_index, [port]), + } + all_failures += [f"{port}: {f}" for f in per_port_op(duthost, port, port_context, metadata_map)] + return all_failures, len(qualifying_ports) diff --git a/tests/transceiver/cdb_firmware_upgrade/test_firmware_prechecks.py b/tests/transceiver/cdb_firmware_upgrade/test_firmware_abort_versions.py similarity index 100% rename from tests/transceiver/cdb_firmware_upgrade/test_firmware_prechecks.py rename to tests/transceiver/cdb_firmware_upgrade/test_firmware_abort_versions.py diff --git a/tests/transceiver/cdb_firmware_upgrade/test_firmware_activation.py b/tests/transceiver/cdb_firmware_upgrade/test_firmware_activation.py new file mode 100644 index 00000000000..b396763398d --- /dev/null +++ b/tests/transceiver/cdb_firmware_upgrade/test_firmware_activation.py @@ -0,0 +1,26 @@ +"""TC4: CDB firmware activation validation. + +Activates firmware and verifies the final firmware state and recovery. +""" +import logging +import pytest + +from tests.transceiver.cdb_firmware_upgrade import firmware_operations + +logger = logging.getLogger(__name__) + + +def test_firmware_activation( + duthost, port_attributes_dict, cdb_firmware_qualifying_ports, get_lport_to_pport_mapping, + required_firmware_metadata_for_all_transceivers, + dom_polling_disabled, +): + """Activate selected firmware and verify every qualifying module recovers.""" + all_failures, num_ports = firmware_operations.run_firmware_op_on_ports( + duthost, port_attributes_dict, cdb_firmware_qualifying_ports, + get_lport_to_pport_mapping, + required_firmware_metadata_for_all_transceivers, firmware_operations.activation_op, + ) + logger.info("Firmware activation exercised %d port(s)", num_ports) + if all_failures: + pytest.fail("Firmware activation failures:\n" + "\n".join(all_failures)) diff --git a/tests/transceiver/cdb_firmware_upgrade/test_firmware_download.py b/tests/transceiver/cdb_firmware_upgrade/test_firmware_download.py new file mode 100644 index 00000000000..ec5609d89f5 --- /dev/null +++ b/tests/transceiver/cdb_firmware_upgrade/test_firmware_download.py @@ -0,0 +1,27 @@ +"""TC3: CDB firmware download validation. + +Downloads the next firmware in ``firmware_versions`` to the target bank and +verifies the firmware-downloaded state. +""" +import logging +import pytest + +from tests.transceiver.cdb_firmware_upgrade import firmware_operations + +logger = logging.getLogger(__name__) + + +def test_firmware_download( + duthost, port_attributes_dict, cdb_firmware_qualifying_ports, get_lport_to_pport_mapping, + required_firmware_metadata_for_all_transceivers, + dom_polling_disabled, +): + """Download firmware to the target bank and verify every qualifying module.""" + all_failures, num_ports = firmware_operations.run_firmware_op_on_ports( + duthost, port_attributes_dict, cdb_firmware_qualifying_ports, + get_lport_to_pport_mapping, + required_firmware_metadata_for_all_transceivers, firmware_operations.perform_firmware_download, + ) + logger.info("Firmware download exercised %d port(s)", num_ports) + if all_failures: + pytest.fail("Firmware download failures:\n" + "\n".join(all_failures)) diff --git a/tests/transceiver/cdb_firmware_upgrade/utils/firmware_utils.py b/tests/transceiver/cdb_firmware_upgrade/utils/firmware_utils.py index 42e65bc37e8..d28cf9417a1 100644 --- a/tests/transceiver/cdb_firmware_upgrade/utils/firmware_utils.py +++ b/tests/transceiver/cdb_firmware_upgrade/utils/firmware_utils.py @@ -1,229 +1,69 @@ -import os import logging -from packaging.version import parse as parse_version +import os import pytest -logger = logging.getLogger(__name__) - -NUM_LATEST_FIRMWARE_VERSIONS = 2 # Number of latest firmware versions to retrieve (gold firmware is added on top) - - -def get_transceiver_gold_firmware_version(normalized_vendor_pn, transceiver_common_attributes): - """ - Returns the gold firmware version for the given transceiver - normalized vendor part number. - The active_firmware field in transceiver_common_attributes corresponds to the - gold firmware version. - - @param normalized_vendor_pn: Normalized vendor part number of the transceiver. - @param transceiver_common_attributes: Dictionary containing common attributes of transceivers. - @return: Returns the gold firmware version as a string if found, otherwise None. - """ - if not normalized_vendor_pn: - pytest.fail("Normalized vendor part number is required to retrieve gold firmware version.") - - transceiver_metadata = transceiver_common_attributes.get(normalized_vendor_pn) - if transceiver_metadata is None: - logger.warning(f"No transceiver metadata found for {normalized_vendor_pn}") - return None - - logger.info(f"Retrieving gold firmware version for {normalized_vendor_pn}") - return transceiver_metadata.get("active_firmware") - - -def get_transceiver_gold_firmware_metadata( - normalized_vendor_name, - normalized_vendor_pn, - transceiver_firmware_info, - transceiver_common_attributes -): - """ - Returns the gold firmware metadata for a transceiver type based on its normalized vendor name - and normalized vendor part number. - @param normalized_vendor_name: Normalized vendor name of the transceiver. - @param normalized_vendor_pn: Normalized vendor part number of the transceiver. - @param transceiver_firmware_info: Dictionary that contains transceiver firmware metadata and attributes. - @param transceiver_common_attributes: Dictionary containing common attributes of transceivers. - @return: Returns the gold firmware metadata dictionary if found, otherwise an empty dictionary. - """ - firmware_metadata_list = get_firmware_metadata_list_by_transceiver_type( - normalized_vendor_name, - normalized_vendor_pn, - transceiver_firmware_info, - ) - - if not firmware_metadata_list: - logger.warning( - f"No firmware metadata available for transceiver type {normalized_vendor_name} {normalized_vendor_pn}" - ) - return {} - - transceiver_gold_firmware_version = get_transceiver_gold_firmware_version( - normalized_vendor_pn, - transceiver_common_attributes, - ) - if not transceiver_gold_firmware_version: - logger.warning( - f"No gold firmware version found for transceiver type {normalized_vendor_name} {normalized_vendor_pn}" - ) - return {} - - for firmware_metadata in firmware_metadata_list: - if firmware_metadata.get("version") == transceiver_gold_firmware_version: - return firmware_metadata - - logger.error( - f"No gold firmware metadata found for version {transceiver_gold_firmware_version} " - f"in {normalized_vendor_name} {normalized_vendor_pn}" - ) - return {} - - -def get_firmware_metadata_list_by_transceiver_type( - normalized_vendor_name, - normalized_vendor_pn, - transceiver_firmware_info, -): - """ - Returns all firmware metadata for a transceiver type based on its normalized vendor name - and normalized vendor part number. - @param normalized_vendor_name: Normalized vendor name of the transceiver. - @param normalized_vendor_pn: Normalized vendor part number of the transceiver. - @param transceiver_firmware_info: Dictionary that contains transceiver firmware metadata. - @return: Returns a list of firmware metadata dictionaries if found, otherwise an empty list. - """ - if not normalized_vendor_name or not normalized_vendor_pn: - pytest.fail("Normalized vendor name and part number are required to retrieve firmware metadata.") +from tests.transceiver.attribute_parser.attribute_keys import ( + BASE_ATTRIBUTES_KEY, + CDB_FIRMWARE_UPGRADE_ATTRIBUTES_KEY, +) - key = (normalized_vendor_name, normalized_vendor_pn) - firmware_metadata_list = transceiver_firmware_info.get(key) - if firmware_metadata_list is None: - logger.warning(f"No firmware metadata found for transceiver type {key}") - return [] - - logger.info(f"Found firmware metadata for transceiver type {key}") - return firmware_metadata_list +logger = logging.getLogger(__name__) def get_required_firmware_metadata_for_all_transceivers( - get_dev_transceiver_details, + port_attributes_dict, transceiver_firmware_info, - transceiver_common_attributes, + qualifying_ports, ): - """ - Finds all types of transceivers installed on the DUT and returns the - required firmware versions (the latest NUM_LATEST_FIRMWARE_VERSIONS plus the - gold firmware) for each type of transceiver. - - @param get_dev_transceiver_details: Dictionary of port transceiver details - @param transceiver_firmware_info: Dictionary containing transceiver firmware information - @param transceiver_common_attributes: Dictionary containing common attributes of transceivers - @return: Dictionary of transceiver types with (normalized vendor name, part number) as keys, - and a list of the required firmware metadata as values. - @raises: pytest.skip if no transceiver details or firmware versions found - @raises: pytest.fail if transceiver_common_attributes not provided or if the required - number of firmware versions is not available for a transceiver type. - """ - if not get_dev_transceiver_details: - pytest.skip("No transceiver details available, skipping test.") - - if not transceiver_common_attributes: - pytest.fail("Transceiver common attributes are required to determine the mandatory gold firmware.") + """Return exact manifest metadata for each qualifying port's firmware_versions.""" + if not port_attributes_dict: + pytest.skip("No port attributes available, skipping test.") + if not qualifying_ports: + pytest.skip("No qualifying CDB firmware ports found, skipping test.") firmware_metadata_by_transceiver_type = {} - for port, port_transceiver_info in get_dev_transceiver_details.items(): - if not port_transceiver_info: - logger.warning(f"No transceiver info found for port {port}") - continue - - # Extract normalized vendor name and part number - normalized_vendor_name = port_transceiver_info.get('normalized_vendor_name') - normalized_vendor_pn = port_transceiver_info.get('normalized_vendor_pn') - - # Validate required fields + for port in qualifying_ports: + port_attrs = port_attributes_dict[port] + base_attrs = port_attrs.get(BASE_ATTRIBUTES_KEY, {}) + cdb_attrs = port_attrs.get(CDB_FIRMWARE_UPGRADE_ATTRIBUTES_KEY, {}) + normalized_vendor_name = base_attrs.get("normalized_vendor_name") + normalized_vendor_pn = base_attrs.get("normalized_vendor_pn") if not normalized_vendor_name or not normalized_vendor_pn: - logger.warning(f"Missing normalized vendor name or part number for port {port}") - continue + pytest.fail(f"{port}: normalized vendor name or part number is missing") transceiver_key = (normalized_vendor_name, normalized_vendor_pn) - - # Skip if we've already processed this transceiver type + firmware_versions = cdb_attrs.get("firmware_versions") + if not firmware_versions: + pytest.fail(f"{port}: firmware_versions is missing or empty") if transceiver_key in firmware_metadata_by_transceiver_type: continue - # Get firmware metadata for this transceiver type - firmware_metadata_list = get_firmware_metadata_list_by_transceiver_type( - normalized_vendor_name, - normalized_vendor_pn, - transceiver_firmware_info - ) + firmware_metadata_list = transceiver_firmware_info.get(transceiver_key) if not firmware_metadata_list: - logger.info(f"No firmware metadata found for transceiver type {transceiver_key}") - continue - - # Sort firmware versions in descending order (newest first) - try: - sorted_firmware = sorted( - firmware_metadata_list, - key=lambda firmware: parse_version(firmware.get('version')), - reverse=True - ) - except Exception as e: - logger.error(f"Error sorting firmware versions for {transceiver_key}: {e}") - continue + pytest.fail(f"No firmware manifest metadata found for transceiver type {transceiver_key}") - # Select the required number of latest firmware versions - num_available = len(sorted_firmware) - if num_available < NUM_LATEST_FIRMWARE_VERSIONS: - pytest.fail( - f"Only {num_available} firmware versions available for transceiver " - f"type {transceiver_key}, but {NUM_LATEST_FIRMWARE_VERSIONS} required. " - f"Available versions: {[fw.get('version') for fw in sorted_firmware]}" - ) - else: - selected_firmware = sorted_firmware[:NUM_LATEST_FIRMWARE_VERSIONS] - - num_required_firmware = NUM_LATEST_FIRMWARE_VERSIONS + 1 # latest versions + gold - - # Add gold firmware - gold_firmware_metadata = get_transceiver_gold_firmware_metadata( - normalized_vendor_name, - normalized_vendor_pn, - transceiver_firmware_info, - transceiver_common_attributes - ) - if gold_firmware_metadata: - # Add gold firmware and remove duplicates while preserving order - all_firmware = selected_firmware + [gold_firmware_metadata] - seen_versions = set() - unique_firmware = [] - for firmware in all_firmware: - version = firmware.get('version') - if version and version not in seen_versions: - seen_versions.add(version) - unique_firmware.append(firmware) - selected_firmware = unique_firmware - - if len(selected_firmware) < num_required_firmware: - for firmware in sorted_firmware[NUM_LATEST_FIRMWARE_VERSIONS:]: - version = firmware.get('version') - if version and version not in seen_versions: - seen_versions.add(version) - selected_firmware.append(firmware) - if len(selected_firmware) >= num_required_firmware: - break - else: - logger.error(f"No gold firmware metadata found for transceiver type {transceiver_key}") - - if len(selected_firmware) != num_required_firmware: + metadata_by_version = {} + for firmware_metadata in firmware_metadata_list: + version = firmware_metadata.get("version") + if version: + metadata_by_version[version] = firmware_metadata + + missing_versions = [ + version for version in firmware_versions if version not in metadata_by_version + ] + if missing_versions: pytest.fail( - f"Expected exactly {num_required_firmware} firmware versions " - f"({NUM_LATEST_FIRMWARE_VERSIONS} latest + gold) for transceiver type {transceiver_key}, " - f"found {len(selected_firmware)}: " - f"{[fw.get('version') for fw in selected_firmware]}" + f"Firmware version(s) {missing_versions} for {transceiver_key} are missing from the manifest" ) + selected_firmware = [metadata_by_version[version] for version in firmware_versions] + for firmware_metadata in selected_firmware: + if not firmware_metadata.get("binary") or not firmware_metadata.get("md5sum"): + pytest.fail( + f"Incomplete firmware metadata for {transceiver_key}/{firmware_metadata.get('version')}" + ) + firmware_metadata_by_transceiver_type[transceiver_key] = selected_firmware if not firmware_metadata_by_transceiver_type: @@ -328,6 +168,7 @@ def download_and_validate_firmware_binaries(duthost, firmware_base_url, firmware firmware_metadata['version'], firmware_metadata['binary'] ) + firmware_metadata['dut_path'] = fw_binary_path_on_dut download_firmware_binary(duthost, fw_binary_path_on_server, fw_binary_path_on_dut) verify_firmware_checksum(duthost, fw_binary_path_on_dut, firmware_metadata['md5sum']) @@ -383,6 +224,7 @@ def stage_prestaged_firmware_binaries(duthost, firmware_host_path, firmware_meta firmware_metadata['version'], firmware_metadata['binary'] ) + firmware_metadata['dut_path'] = fw_binary_path_on_dut copy_firmware_binary(duthost, fw_binary_path_on_host, fw_binary_path_on_dut) verify_firmware_checksum(duthost, fw_binary_path_on_dut, firmware_metadata['md5sum']) diff --git a/tests/transceiver/common/cli_helpers.py b/tests/transceiver/common/cli_helpers.py index 398309e74c0..599dff199d0 100644 --- a/tests/transceiver/common/cli_helpers.py +++ b/tests/transceiver/common/cli_helpers.py @@ -33,6 +33,8 @@ ``pytest.fail`` at the end, and an exception inside the loop would short-circuit that pattern. """ +import time + from tests.transceiver.common.cli_parser_helper import ( parse_fwversion, parse_hexdump, @@ -65,6 +67,10 @@ SFPUTIL_SHOW_PRESENCE = "sfputil show presence" SHOW_TRANSCEIVER_INFO = "show interfaces transceiver info" SHOW_TRANSCEIVER_PRESENCE = "show interfaces transceiver presence" +SFPUTIL_FIRMWARE_DOWNLOAD = "sfputil firmware download" +SFPUTIL_FIRMWARE_RUN = "sfputil firmware run" +SFPUTIL_FIRMWARE_COMMIT = "sfputil firmware commit" +SFPUTIL_RESET = "sfputil reset" # Max characters of stdout/stderr echoed into a failure message. Some sfputil # errors dump the full 500+ port list, which would bury the failure summary in @@ -72,6 +78,13 @@ # "Error: invalid port ..." / "Root privileges are required") while keeping the # aggregated per-port failure report readable. CLI_ERROR_DETAIL_MAX_CHARS = 200 +TIMEOUT_RC = 124 + +# Success markers for the firmware sfputil commands +FW_DOWNLOAD_SUCCESS_MARKER = "Firmware download complete success" +FW_RUN_SUCCESS_MARKER = "Firmware run in mode=0 success" +FW_COMMIT_SUCCESS_MARKER = "Firmware commit successful" +SFPUTIL_RESET_SUCCESS_MARKER = "OK" # ────────────────────────────────────────────────────────────────────── @@ -361,7 +374,7 @@ def set_dom_polling(duthost, port, enable, namespace=None): """ action = "enable" if enable else "disable" cmd = f"config interface{_ns_flag(namespace)} transceiver dom {port} {action}" - result = duthost.shell(cmd, module_ignore_errors=True) + result = duthost.command(cmd, module_ignore_errors=True) if result.get("rc", RC_FAILURE) != 0: return f"{cmd} failed with rc={result.get('rc')} ({_error_detail(result)})" return None @@ -371,3 +384,70 @@ def show_interfaces_transceiver_info(duthost, port=None, namespace=None): """Run ``show interfaces transceiver info [-n ] []`` → ``({port: {field: value}}, err)``.""" cmd = show_interfaces_transceiver_info_cmd(port, namespace=namespace) return _run_and_parse(duthost, cmd, parse_eeprom) + + +_CDB_FW_ABORT_PYCODE = ( + "import sonic_platform.platform as P\n" + "api = P.Platform().get_chassis().get_sfp({idx}).get_xcvr_api()\n" + "cdb = getattr(api, 'cdb', None)\n" + "print(cdb.abort_fw_download() if cdb is not None else 'NO_CDB')\n" +) + + +def _run_firmware_cmd(duthost, cmd, timeout_sec, success_marker): + """Run ``cmd`` under a hard ``timeout``; returns ``(elapsed_sec, err)``. + + ``err`` is ``None`` only when rc is 0 AND ``success_marker`` is in stdout. + """ + wrapped = f"timeout {timeout_sec} {cmd}" + start = time.time() + result = duthost.command(wrapped, module_ignore_errors=True) + elapsed = round(time.time() - start, 1) + rc = result.get("rc", RC_FAILURE) + if rc == TIMEOUT_RC: + return elapsed, f"{cmd} timed out after {timeout_sec}s" + if rc != 0: + return elapsed, f"{cmd} failed with rc={rc} ({_error_detail(result)})" + stdout = "\n".join(result.get("stdout_lines") or []) + if success_marker and success_marker not in stdout: + return elapsed, f"{cmd} did not report success ('{success_marker}' absent)" + return elapsed, None + + +def sfputil_firmware_download(duthost, port, fwfile, timeout_sec): + """Run ``sfputil firmware download ""`` + + Returns ``(elapsed_sec, err)``. + """ + cmd = f'{SFPUTIL_FIRMWARE_DOWNLOAD} {port} "{fwfile}"' + return _run_firmware_cmd(duthost, cmd, timeout_sec, FW_DOWNLOAD_SUCCESS_MARKER) + + +def sfputil_firmware_run(duthost, port, timeout_sec): + """Run ``sfputil firmware run `` returns ``(elapsed_sec, err)``.""" + cmd = f"{SFPUTIL_FIRMWARE_RUN} {port}" + return _run_firmware_cmd(duthost, cmd, timeout_sec, FW_RUN_SUCCESS_MARKER) + + +def sfputil_firmware_commit(duthost, port, timeout_sec): + """Run ``sfputil firmware commit `` returns ``(elapsed_sec, err)``.""" + cmd = f"{SFPUTIL_FIRMWARE_COMMIT} {port}" + return _run_firmware_cmd(duthost, cmd, timeout_sec, FW_COMMIT_SUCCESS_MARKER) + + +def sfputil_reset(duthost, port, timeout_sec=60): + """Run ``sfputil reset `` returns ``(elapsed_sec, err)``.""" + cmd = f"{SFPUTIL_RESET} {port}" + return _run_firmware_cmd(duthost, cmd, timeout_sec, SFPUTIL_RESET_SUCCESS_MARKER) + + +def issue_cdb_fw_abort(duthost, physical_index): + """``status`` is the raw ``api.cdb.abort_fw_download()`` reply.""" + pycode = _CDB_FW_ABORT_PYCODE.format(idx=int(physical_index)) + result = duthost.shell('python3 -c "{}"'.format(pycode), module_ignore_errors=True) + if result.get("rc", RC_FAILURE) != 0: + return None, f"CDB abort failed with rc={result.get('rc')} ({_error_detail(result)})" + status = " ".join(result.get("stdout_lines") or []).strip() + if status == "NO_CDB": + return None, "CDB not supported on module" + return status, None diff --git a/tests/transceiver/common/cli_parser_helper.py b/tests/transceiver/common/cli_parser_helper.py index fb5551b5e99..76f433a7c81 100644 --- a/tests/transceiver/common/cli_parser_helper.py +++ b/tests/transceiver/common/cli_parser_helper.py @@ -17,6 +17,12 @@ # ── General shell / command-result constants ──────────────────────────── "RC_FAILURE", + # ── sfputil show fwversion field labels ───────────────────────────────── + "FW_ACTIVE", + "FW_INACTIVE", + "FW_RUNNING_IMAGE", + "FW_COMMITTED_IMAGE", + # ── Public parsers ────────────────────────────────────────────────────── "parse_fwversion", "parse_hexdump", @@ -205,6 +211,12 @@ def parse_presence(output_lines): return res +FW_ACTIVE = "Active Firmware" +FW_INACTIVE = "Inactive Firmware" +FW_RUNNING_IMAGE = "Running Image" +FW_COMMITTED_IMAGE = "Committed Image" + + def parse_fwversion(output_lines): """Parse ``sfputil show fwversion `` output into a ``{field: value}`` map. diff --git a/tests/transceiver/common/dmesg_helpers.py b/tests/transceiver/common/dmesg_helpers.py new file mode 100644 index 00000000000..4c60cd094bb --- /dev/null +++ b/tests/transceiver/common/dmesg_helpers.py @@ -0,0 +1,40 @@ +"""Scan operation-scoped dmesg errors using monotonic timestamps.""" +import re + +_DMESG_MONOTONIC_TS_RE = re.compile(r'^\[\s*(\d+(?:\.\d+)?)\]') +_DMESG_ERROR_LEVELS = "emerg,alert,crit,err,warn" + + +def capture_dmesg_uptime_watermark(duthost): + """Return ``(seconds_since_boot, err)`` for an operation-window watermark.""" + result = duthost.shell("cat /proc/uptime", module_ignore_errors=True) + if result.get("rc", 1) != 0: + return None, "failed to capture dmesg uptime watermark" + try: + return float(result.get("stdout_lines", [])[0].split()[0]), None + except (ValueError, IndexError): + return None, "could not parse /proc/uptime for dmesg watermark" + + +def scan_new_dmesg_errors(duthost, start_uptime, seen_errors, grep_pattern): + """Return ``(new_matching_lines, err)`` and update ``seen_errors``.""" + result = duthost.shell( + "sudo dmesg --level=" + _DMESG_ERROR_LEVELS, + module_ignore_errors=True, + ) + if result.get("rc", 1) != 0: + return [], "failed to read dmesg for operation-window errors" + + pattern = re.compile(grep_pattern, re.IGNORECASE) + truly_new = [] + for line in result.get("stdout_lines", []): + line = line.strip() + if not line or pattern.search(line) is None: + continue + m = _DMESG_MONOTONIC_TS_RE.match(line) + if m is not None and float(m.group(1)) < start_uptime: + continue + if line not in seen_errors: + seen_errors.add(line) + truly_new.append(line) + return truly_new, None diff --git a/tests/transceiver/common/scenario_ops.py b/tests/transceiver/common/scenario_ops.py index bdfd0c0ed40..40e7b893296 100644 --- a/tests/transceiver/common/scenario_ops.py +++ b/tests/transceiver/common/scenario_ops.py @@ -10,8 +10,10 @@ """ import logging +import time from tests.common.platform.interface_utils import wait_ports_oper_status +from tests.transceiver.common import cli_helpers logger = logging.getLogger(__name__) @@ -80,3 +82,24 @@ def perform_ports_startup(duthost, ports, wait_sec): else: logger.info("All %d port(s) reached oper-up", len(ports)) return failures + + +def perform_sfputil_reset(duthost, port, recover_with_port_toggle=True, + i2c_recover_sec=5, shutdown_wait=5, startup_wait=60): + """``sfputil reset`` a transceiver, optionally bracketed by a port toggle. + + Some modules stay oper-down after a reset, so ``recover_with_port_toggle`` + helps bring the port back up. Sleeps ``i2c_recover_sec`` after the reset. + Returns a list of per-port failure strings. + """ + failures = [] + if recover_with_port_toggle: + failures += perform_ports_shutdown(duthost, [port], shutdown_wait) + reset_elapsed, reset_err = cli_helpers.sfputil_reset(duthost, port) + logger.info("Port %s: transceiver reset took %ss", port, reset_elapsed) + if reset_err: + failures.append(f"transceiver reset failed on {port}: {reset_err}") + time.sleep(i2c_recover_sec) + if recover_with_port_toggle: + failures += perform_ports_startup(duthost, [port], startup_wait) + return failures