From c9bcc1753bae4eea4c1c42e9e4e66283951228e6 Mon Sep 17 00:00:00 2001 From: Hasan Khan Date: Sat, 8 Aug 2026 18:45:56 -0700 Subject: [PATCH 1/3] Add NICo NVSwitch firmware query Signed-off-by: Hasan Khan --- .../providers/nico/config/bare_metal.yaml | 4 +- .../nico/scripts/breakfix/gap_stub.py | 6 +- .../scripts/breakfix/query_switch_firmware.py | 110 ++++++++++++++++ .../providers/nico/test_nico_provider.py | 124 ++++++++++++++++++ 4 files changed, 236 insertions(+), 8 deletions(-) create mode 100755 isvctl/configs/providers/nico/scripts/breakfix/query_switch_firmware.py diff --git a/isvctl/configs/providers/nico/config/bare_metal.yaml b/isvctl/configs/providers/nico/config/bare_metal.yaml index 1b84b2baf..54882f16b 100644 --- a/isvctl/configs/providers/nico/config/bare_metal.yaml +++ b/isvctl/configs/providers/nico/config/bare_metal.yaml @@ -364,7 +364,7 @@ commands: - name: query_switch_firmware phase: test continue_on_failure: true - command: "python ../scripts/breakfix/gap_stub.py" + command: "python ../scripts/breakfix/query_switch_firmware.py" args: - "--org" - "{{org}}" @@ -372,8 +372,6 @@ commands: - "{{site_id}}" - "--api-base" - "{{nico_api_base}}" - - "--gap" - - "BFX03-02" timeout: 120 - name: query_bmc_kernel_logs diff --git a/isvctl/configs/providers/nico/scripts/breakfix/gap_stub.py b/isvctl/configs/providers/nico/scripts/breakfix/gap_stub.py index 55fdc447e..b5df988c8 100644 --- a/isvctl/configs/providers/nico/scripts/breakfix/gap_stub.py +++ b/isvctl/configs/providers/nico/scripts/breakfix/gap_stub.py @@ -6,7 +6,7 @@ Several break-fix requirements have no NICo tenant REST surface to exercise: the mutating BFX01 workflows run through Maestro/repair fixtures, and the -BFX02-02/BFX03-02/BFX04-01/BFX05/BFX06 signals are not exposed at all. Each of +BFX02-02/BFX04-01/BFX05/BFX06 signals are not exposed at all. Each of those steps emits a structured skip naming the gap rather than a hard failure, so the suite reports "not available on this platform" instead of "broken". @@ -44,10 +44,6 @@ "NICo has no retirement-notice query API (BFX02-02 gap)", {"notices_queryable": False, "notices": []}, ), - "BFX03-02": ( - "NV switch tray firmware is not queryable via NICo tenant REST API (BFX03-02 gap)", - {"trays": []}, - ), "BFX04-01": ( "GPUd/Sentinel/Maestro node health agents are not observable via NICo REST (BFX04-01 gap)", {"agents_observable": False, "agents": []}, diff --git a/isvctl/configs/providers/nico/scripts/breakfix/query_switch_firmware.py b/isvctl/configs/providers/nico/scripts/breakfix/query_switch_firmware.py new file mode 100755 index 000000000..bb6230173 --- /dev/null +++ b/isvctl/configs/providers/nico/scripts/breakfix/query_switch_firmware.py @@ -0,0 +1,110 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Query NVSwitch tray firmware versions from NICo racks (BFX03-02). + +NICo's read-only rack list endpoint returns rack components when +``includeComponents`` is enabled. NVSwitch components expose their installed +firmware through ``firmwareVersion``. +""" + +from __future__ import annotations + +import argparse +import re +import sys +from pathlib import Path +from typing import Any +from urllib.error import URLError + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from breakfix._common import emit +from common.nico_client import NicoAuthError, forge_get_all, resolve_auth + + +def _is_nvswitch(component: dict[str, Any]) -> bool: + """Return whether a rack component is explicitly typed as an NVSwitch.""" + component_type = re.sub(r"[^a-z0-9]", "", str(component.get("type") or "").lower()) + return component_type in {"nvswitch", "componenttypenvswitch"} + + +def _tray_id(component: dict[str, Any]) -> str: + """Choose the first stable, human-useful identifier NICo provides.""" + for field in ("componentId", "id", "serialNumber", "name"): + value = component.get(field) + if value is not None and str(value).strip(): + return str(value) + return "" + + +def main() -> int: + """List NVSwitch tray firmware versions as provider-neutral JSON.""" + parser = argparse.ArgumentParser(description="Query NICo NVSwitch tray firmware versions") + parser.add_argument("--org", required=True) + parser.add_argument("--site-id", required=True) + parser.add_argument("--api-base", required=True) + args = parser.parse_args() + + result: dict[str, Any] = { + "success": False, + "platform": "nico", + "site_id": args.site_id, + "trays": [], + } + try: + auth = resolve_auth() + racks = forge_get_all( + args.org, + "rack", + auth.token, + base_url=args.api_base, + params={"siteId": args.site_id, "includeComponents": "true"}, + result_key="racks", + ) + except NicoAuthError as exc: + result.update(error_type="auth", error=str(exc)) + return emit(result) + except (URLError, ValueError) as exc: + result["error"] = f"{type(exc).__name__}: {exc}" + return emit(result) + + seen: set[str] = set() + trays: list[dict[str, Any]] = [] + for rack in racks: + rack_id = str(rack.get("id") or "") + components = rack.get("components") or [] + if not isinstance(components, list): + continue + for component in components: + if not isinstance(component, dict) or not _is_nvswitch(component): + continue + tray_id = _tray_id(component) + dedupe_key = tray_id or f"{rack_id}:{component.get('slotId')}:{component.get('trayIdx')}" + if dedupe_key in seen: + continue + seen.add(dedupe_key) + trays.append( + { + "tray_id": tray_id, + "firmware_version": str(component.get("firmwareVersion") or ""), + "rack_id": rack_id, + "slot_id": component.get("slotId"), + "tray_index": component.get("trayIdx"), + } + ) + + if not trays: + result.update( + success=True, + skipped=True, + skip_reason="No NVSwitch tray components were returned for this NICo site", + ) + return emit(result) + + result.update(success=True, trays=trays) + return emit(result) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/isvctl/tests/providers/nico/test_nico_provider.py b/isvctl/tests/providers/nico/test_nico_provider.py index 40c1cc708..640f51bba 100644 --- a/isvctl/tests/providers/nico/test_nico_provider.py +++ b/isvctl/tests/providers/nico/test_nico_provider.py @@ -141,6 +141,14 @@ def _load_nico_script(relative_path: str, module_name: str) -> ModuleType: return module +def _load_switch_firmware_script() -> ModuleType: + """Load the BFX03-02 NVSwitch firmware query script.""" + return _load_nico_script( + "breakfix/query_switch_firmware.py", + "test_query_switch_firmware", + ) + + def _load_governance_metrics_script() -> ModuleType: """Load the query_metrics (governance) script as a module for direct unit testing.""" script_path = NICO_SCRIPTS / "governance" / "query_metrics.py" @@ -1095,6 +1103,122 @@ def test_nico_scripts_require_api_base( assert "--api-base" in captured.err +def test_switch_firmware_queries_racks_and_maps_only_nvswitch_components( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """BFX03-02 should use NICo rack inventory and preserve missing firmware for validation.""" + module = _load_switch_firmware_script() + observed: dict[str, Any] = {} + + def fake_get_all(*args: Any, **kwargs: Any) -> list[dict[str, Any]]: + observed["args"] = args + observed["kwargs"] = kwargs + return [ + { + "id": "rack-1", + "components": [ + { + "id": "internal-switch-1", + "componentId": "switch-1", + "type": "ComponentTypeNVSwitch", + "firmwareVersion": "1.2.3", + "slotId": 3, + "trayIdx": 0, + }, + { + "id": "switch-2", + "type": "nvswitch", + "firmwareVersion": None, + }, + { + "id": "compute-1", + "type": "ComponentTypeCompute", + "firmwareVersion": "9.9.9", + }, + ], + } + ] + + monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="test-token")) + monkeypatch.setattr(module, "forge_get_all", fake_get_all) + monkeypatch.setattr( + sys, + "argv", + [ + "query_switch_firmware.py", + "--org", + "test-org", + "--site-id", + "site-1", + "--api-base", + "https://nico.example/v2/org", + ], + ) + + assert module.main() == 0 + + payload = json.loads(capsys.readouterr().out) + assert payload["success"] is True + assert payload["trays"] == [ + { + "tray_id": "switch-1", + "firmware_version": "1.2.3", + "rack_id": "rack-1", + "slot_id": 3, + "tray_index": 0, + }, + { + "tray_id": "switch-2", + "firmware_version": "", + "rack_id": "rack-1", + "slot_id": None, + "tray_index": None, + }, + ] + assert observed["args"] == ("test-org", "rack", "test-token") + assert observed["kwargs"] == { + "base_url": "https://nico.example/v2/org", + "params": {"siteId": "site-1", "includeComponents": "true"}, + "result_key": "racks", + } + + +def test_switch_firmware_skips_when_site_has_no_nvswitch_components( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """A site without NVSwitch inventory is inapplicable, not a false pass.""" + module = _load_switch_firmware_script() + monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="test-token")) + monkeypatch.setattr( + module, + "forge_get_all", + lambda *args, **kwargs: [{"id": "rack-1", "components": [{"type": "ComponentTypeCompute"}]}], + ) + monkeypatch.setattr( + sys, + "argv", + [ + "query_switch_firmware.py", + "--org", + "test-org", + "--site-id", + "site-1", + "--api-base", + "https://nico.example/v2/org", + ], + ) + + assert module.main() == 0 + + payload = json.loads(capsys.readouterr().out) + assert payload["success"] is True + assert payload["skipped"] is True + assert payload["trays"] == [] + assert "No NVSwitch tray components" in payload["skip_reason"] + + def test_dpu_health_script_treats_nullable_machine_lists_as_empty( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], From d4d746196f684ed1227387f20323ff295033050a Mon Sep 17 00:00:00 2001 From: Hasan Khan Date: Sat, 8 Aug 2026 21:40:56 -0700 Subject: [PATCH 2/3] Handle Flow-disabled NVSwitch inventory Signed-off-by: Hasan Khan --- .../scripts/breakfix/query_switch_firmware.py | 76 +++++---- .../providers/nico/test_nico_provider.py | 146 ++++++++++++++---- isvtest/tests/test_breakfix.py | 43 ++++++ 3 files changed, 198 insertions(+), 67 deletions(-) diff --git a/isvctl/configs/providers/nico/scripts/breakfix/query_switch_firmware.py b/isvctl/configs/providers/nico/scripts/breakfix/query_switch_firmware.py index bb6230173..c40b01259 100755 --- a/isvctl/configs/providers/nico/scripts/breakfix/query_switch_firmware.py +++ b/isvctl/configs/providers/nico/scripts/breakfix/query_switch_firmware.py @@ -2,11 +2,11 @@ # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -"""Query NVSwitch tray firmware versions from NICo racks (BFX03-02). +"""Query NVSwitch tray firmware versions from NICo Flow (BFX03-02). -NICo's read-only rack list endpoint returns rack components when -``includeComponents`` is enabled. NVSwitch components expose their installed -firmware through ``firmwareVersion``. +NICo's read-only tray list endpoint returns every tray at a Flow-enabled site. +The provider filters the version-specific tray type values client-side, and +NVSwitch trays expose their installed firmware through ``firmwareVersion``. """ from __future__ import annotations @@ -16,17 +16,19 @@ import sys from pathlib import Path from typing import Any -from urllib.error import URLError +from urllib.error import HTTPError, URLError sys.path.insert(0, str(Path(__file__).resolve().parents[1])) -from breakfix._common import emit +from breakfix._common import emit, skip_result from common.nico_client import NicoAuthError, forge_get_all, resolve_auth +_FLOW_DISABLED_MESSAGE = "site does not have nico flow enabled" + def _is_nvswitch(component: dict[str, Any]) -> bool: - """Return whether a rack component is explicitly typed as an NVSwitch.""" + """Return whether a tray is explicitly typed as an NVSwitch.""" component_type = re.sub(r"[^a-z0-9]", "", str(component.get("type") or "").lower()) - return component_type in {"nvswitch", "componenttypenvswitch"} + return component_type in {"switch", "nvswitch", "componenttypenvswitch"} def _tray_id(component: dict[str, Any]) -> str: @@ -38,6 +40,11 @@ def _tray_id(component: dict[str, Any]) -> str: return "" +def _flow_disabled(exc: HTTPError) -> bool: + """Return whether NICo rejected the query because Flow is disabled.""" + return exc.code == 412 and _FLOW_DISABLED_MESSAGE in str(exc).lower() + + def main() -> int: """List NVSwitch tray firmware versions as provider-neutral JSON.""" parser = argparse.ArgumentParser(description="Query NICo NVSwitch tray firmware versions") @@ -54,45 +61,48 @@ def main() -> int: } try: auth = resolve_auth() - racks = forge_get_all( + components = forge_get_all( args.org, - "rack", + "tray", auth.token, base_url=args.api_base, - params={"siteId": args.site_id, "includeComponents": "true"}, - result_key="racks", + params={"siteId": args.site_id}, + result_key="trays", ) except NicoAuthError as exc: result.update(error_type="auth", error=str(exc)) return emit(result) + except HTTPError as exc: + if _flow_disabled(exc): + skip = skip_result( + args.site_id, + "NICo Flow is not enabled for this site; NVSwitch tray firmware is unavailable (BFX03-02 gap)", + gap="BFX03-02", + ) + skip["trays"] = [] + return emit(skip) + result.update(error_type="api", error=f"NICo tray query failed (HTTP {exc.code})") + return emit(result) except (URLError, ValueError) as exc: result["error"] = f"{type(exc).__name__}: {exc}" return emit(result) seen: set[str] = set() trays: list[dict[str, Any]] = [] - for rack in racks: - rack_id = str(rack.get("id") or "") - components = rack.get("components") or [] - if not isinstance(components, list): + for component in components: + if not isinstance(component, dict) or not _is_nvswitch(component): continue - for component in components: - if not isinstance(component, dict) or not _is_nvswitch(component): - continue - tray_id = _tray_id(component) - dedupe_key = tray_id or f"{rack_id}:{component.get('slotId')}:{component.get('trayIdx')}" - if dedupe_key in seen: - continue - seen.add(dedupe_key) - trays.append( - { - "tray_id": tray_id, - "firmware_version": str(component.get("firmwareVersion") or ""), - "rack_id": rack_id, - "slot_id": component.get("slotId"), - "tray_index": component.get("trayIdx"), - } - ) + tray_id = _tray_id(component) + if tray_id and tray_id in seen: + continue + if tray_id: + seen.add(tray_id) + trays.append( + { + "tray_id": tray_id, + "firmware_version": str(component.get("firmwareVersion") or ""), + } + ) if not trays: result.update( diff --git a/isvctl/tests/providers/nico/test_nico_provider.py b/isvctl/tests/providers/nico/test_nico_provider.py index 640f51bba..2c00560b1 100644 --- a/isvctl/tests/providers/nico/test_nico_provider.py +++ b/isvctl/tests/providers/nico/test_nico_provider.py @@ -1103,11 +1103,11 @@ def test_nico_scripts_require_api_base( assert "--api-base" in captured.err -def test_switch_firmware_queries_racks_and_maps_only_nvswitch_components( +def test_switch_firmware_queries_all_trays_and_filters_nvswitches( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: - """BFX03-02 should use NICo rack inventory and preserve missing firmware for validation.""" + """BFX03-02 should query all trays and accept each deployed NVSwitch type spelling.""" module = _load_switch_firmware_script() observed: dict[str, Any] = {} @@ -1116,28 +1116,32 @@ def fake_get_all(*args: Any, **kwargs: Any) -> list[dict[str, Any]]: observed["kwargs"] = kwargs return [ { - "id": "rack-1", - "components": [ - { - "id": "internal-switch-1", - "componentId": "switch-1", - "type": "ComponentTypeNVSwitch", - "firmwareVersion": "1.2.3", - "slotId": 3, - "trayIdx": 0, - }, - { - "id": "switch-2", - "type": "nvswitch", - "firmwareVersion": None, - }, - { - "id": "compute-1", - "type": "ComponentTypeCompute", - "firmwareVersion": "9.9.9", - }, - ], - } + "id": "internal-switch-1", + "componentId": "switch-1", + "type": "switch", + "firmwareVersion": "1.2.3", + "rackId": "rack-1", + "position": {"slotId": 3, "trayIdx": 0}, + }, + { + "id": "switch-2", + "type": "NVSwitch", + "firmwareVersion": None, + "rackId": "rack-2", + "position": {"slotId": 5, "trayIdx": 1}, + }, + { + "id": "switch-3", + "type": "ComponentTypeNVSwitch", + "firmwareVersion": "3.4.5", + "rackId": "rack-2", + }, + { + "id": "compute-1", + "type": "compute", + "firmwareVersion": "9.9.9", + "rackId": "rack-2", + }, ] monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="test-token")) @@ -1164,23 +1168,21 @@ def fake_get_all(*args: Any, **kwargs: Any) -> list[dict[str, Any]]: { "tray_id": "switch-1", "firmware_version": "1.2.3", - "rack_id": "rack-1", - "slot_id": 3, - "tray_index": 0, }, { "tray_id": "switch-2", "firmware_version": "", - "rack_id": "rack-1", - "slot_id": None, - "tray_index": None, + }, + { + "tray_id": "switch-3", + "firmware_version": "3.4.5", }, ] - assert observed["args"] == ("test-org", "rack", "test-token") + assert observed["args"] == ("test-org", "tray", "test-token") assert observed["kwargs"] == { "base_url": "https://nico.example/v2/org", - "params": {"siteId": "site-1", "includeComponents": "true"}, - "result_key": "racks", + "params": {"siteId": "site-1"}, + "result_key": "trays", } @@ -1194,7 +1196,7 @@ def test_switch_firmware_skips_when_site_has_no_nvswitch_components( monkeypatch.setattr( module, "forge_get_all", - lambda *args, **kwargs: [{"id": "rack-1", "components": [{"type": "ComponentTypeCompute"}]}], + lambda *args, **kwargs: [{"id": "compute-1", "type": "compute"}], ) monkeypatch.setattr( sys, @@ -1219,6 +1221,82 @@ def test_switch_firmware_skips_when_site_has_no_nvswitch_components( assert "No NVSwitch tray components" in payload["skip_reason"] +def test_switch_firmware_skips_when_nico_flow_is_disabled( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """NICo's Flow-disabled precondition is a documented runtime gap, not a pass.""" + module = _load_switch_firmware_script() + + def flow_disabled(*args: Any, **kwargs: Any) -> list[dict[str, Any]]: + raise HTTPError( + "https://nico.example/v2/org/test-org/nico/tray", + 412, + "Site does not have NICo Flow enabled", + None, + None, + ) + + monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="test-token")) + monkeypatch.setattr(module, "forge_get_all", flow_disabled) + monkeypatch.setattr( + sys, + "argv", + [ + "query_switch_firmware.py", + "--org", + "test-org", + "--site-id", + "site-1", + "--api-base", + "https://nico.example/v2/org", + ], + ) + + assert module.main() == 0 + + payload = json.loads(capsys.readouterr().out) + assert payload["success"] is True + assert payload["skipped"] is True + assert payload["gap"] == "BFX03-02" + assert payload["trays"] == [] + assert "Flow is not enabled" in payload["skip_reason"] + + +def test_switch_firmware_does_not_skip_other_http_errors( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """Authentication and API authorization failures must remain failures.""" + module = _load_switch_firmware_script() + + def forbidden(*args: Any, **kwargs: Any) -> list[dict[str, Any]]: + raise HTTPError("https://nico.example/tray", 403, "Forbidden", None, None) + + monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="test-token")) + monkeypatch.setattr(module, "forge_get_all", forbidden) + monkeypatch.setattr( + sys, + "argv", + [ + "query_switch_firmware.py", + "--org", + "test-org", + "--site-id", + "site-1", + "--api-base", + "https://nico.example/v2/org", + ], + ) + + assert module.main() == 1 + + payload = json.loads(capsys.readouterr().out) + assert payload["success"] is False + assert payload["error_type"] == "api" + assert "skipped" not in payload + + def test_dpu_health_script_treats_nullable_machine_lists_as_empty( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], diff --git a/isvtest/tests/test_breakfix.py b/isvtest/tests/test_breakfix.py index 485c8cbcb..c4465815c 100644 --- a/isvtest/tests/test_breakfix.py +++ b/isvtest/tests/test_breakfix.py @@ -17,6 +17,7 @@ HostReplacementCheck, MaintenanceEventsCheck, NodeHealthAgentCheck, + NvSwitchFirmwareCheck, PlannedMaintenanceNotificationCheck, RepairHistoryCheck, RetirementNoticesCheck, @@ -146,6 +147,48 @@ def test_node_maintenance_reports_mode(self) -> None: assert "maintenance_mode=hw" in check.message +class TestNvSwitchFirmwareCheck: + """Cover BFX03-02 firmware evidence for every returned switch tray.""" + + def test_propagates_flow_disabled_runtime_skip(self) -> None: + """A Flow-disabled site remains skipped instead of becoming a pass.""" + step_output = { + "success": True, + "skipped": True, + "skip_reason": "NICo Flow is not enabled for this site", + "gap": "BFX03-02", + "trays": [], + } + with pytest.raises(pytest.skip.Exception): + _run(NvSwitchFirmwareCheck, step_output) + + def test_passes_when_every_tray_has_firmware(self) -> None: + """Every discovered NVSwitch tray must report a non-empty version.""" + step_output = { + "success": True, + "trays": [ + {"tray_id": "switch-1", "firmware_version": "1.2.3"}, + {"tray_id": "switch-2", "firmware_version": "2.0.0"}, + ], + } + check = _run(NvSwitchFirmwareCheck, step_output) + assert check.passed + assert "2 NV switch tray(s)" in check.message + + def test_fails_when_any_tray_has_no_firmware(self) -> None: + """One missing version keeps partial inventory from passing BFX03-02.""" + step_output = { + "success": True, + "trays": [ + {"tray_id": "switch-1", "firmware_version": "1.2.3"}, + {"tray_id": "switch-2", "firmware_version": ""}, + ], + } + check = _run(NvSwitchFirmwareCheck, step_output) + assert not check.passed + assert "1 switch tray(s) missing firmware_version" in check.message + + class TestNodeHealthAgentCheck: """Cover the BFX04-01 GPUd/Sentinel health-agent check.""" From 9305b483bf56f37939c6987e1ce3f09a924359eb Mon Sep 17 00:00:00 2001 From: Hasan Khan Date: Tue, 18 Aug 2026 21:03:48 -0700 Subject: [PATCH 3/3] Add direct GB300 NVSwitch firmware inspection Signed-off-by: Hasan Khan --- .../providers/gb300/config/bare_metal.yaml | 44 ++++ .../scripts/breakfix/query_switch_firmware.py | 199 +++++++++++++++++ .../providers/nico/config/bare_metal.yaml | 4 +- .../nico/scripts/breakfix/gap_stub.py | 6 +- .../scripts/breakfix/query_switch_firmware.py | 120 ---------- .../providers/gb300/test_gb300_provider.py | 208 ++++++++++++++++++ .../providers/nico/test_nico_provider.py | 202 ----------------- isvtest/tests/test_breakfix.py | 16 +- 8 files changed, 464 insertions(+), 335 deletions(-) create mode 100644 isvctl/configs/providers/gb300/config/bare_metal.yaml create mode 100644 isvctl/configs/providers/gb300/scripts/breakfix/query_switch_firmware.py delete mode 100755 isvctl/configs/providers/nico/scripts/breakfix/query_switch_firmware.py create mode 100644 isvctl/tests/providers/gb300/test_gb300_provider.py diff --git a/isvctl/configs/providers/gb300/config/bare_metal.yaml b/isvctl/configs/providers/gb300/config/bare_metal.yaml new file mode 100644 index 000000000..9615c8b38 --- /dev/null +++ b/isvctl/configs/providers/gb300/config/bare_metal.yaml @@ -0,0 +1,44 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +# Direct GB300 NVSwitch firmware validation. +# +# This provider runs on a GB300 BCM head where Bright Cluster Manager and +# cm-nvfwupd are installed. It reads existing NVSwitch inventory and executes +# only `nvfwupd show_version -j`; it never performs a firmware update. +# +# Prerequisites: +# - passwordless sudo for the read-only cmsh and nvfwupd commands +# - GB300_RACK set to a rack prefix (for example, a05-p01), or +# GB300_SWITCH_HOSTS set to a comma-separated list of NVSwitch hostnames +# - GB300_SWITCH_LIMIT optionally controls the number inspected (default: 1) +# +# Usage: +# GB300_RACK=a05-p01 \ +# uv run isvctl test run \ +# -f isvctl/configs/providers/gb300/config/bare_metal.yaml \ +# -- -k NvSwitchFirmwareCheck + +import: + - ../../../suites/bare_metal.yaml + +version: "1.0" + +commands: + bare_metal: + phases: ["test"] + steps: + - name: query_switch_firmware + phase: test + continue_on_failure: true + command: "python ../scripts/breakfix/query_switch_firmware.py" + timeout: 180 + +tests: + cluster_name: "gb300-direct-firmware-validation" + description: "Inspect GB300 NVSwitch tray firmware directly with nvfwupd" + + settings: + region: "" + instance_type: "" + teardown_flag: "" diff --git a/isvctl/configs/providers/gb300/scripts/breakfix/query_switch_firmware.py b/isvctl/configs/providers/gb300/scripts/breakfix/query_switch_firmware.py new file mode 100644 index 000000000..57f27b4a5 --- /dev/null +++ b/isvctl/configs/providers/gb300/scripts/breakfix/query_switch_firmware.py @@ -0,0 +1,199 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Inspect GB300 NVSwitch tray firmware with read-only ``nvfwupd`` (BFX03-02). + +The script is intended to run on a GB300 BCM head. It discovers dedicated +``nvswitch`` devices through read-only ``cmsh`` inventory and invokes only +``nvfwupd show_version -j`` against the selected tray BMCs. Credentials remain +inside the privileged subprocess and are never included in the JSON result. +""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import subprocess +import sys +from collections.abc import Sequence +from typing import Any + +_HOST_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$") +_NVFWUPD_PATH = "/cm/local/apps/cm-nvfwupd/nvfwupd" +_SERVER_TYPE = "gb200switch" +_MODULE_SETUP = """\ +export MODULEPATH=/cm/local/modulefiles:/cm/shared/modulefiles +source /cm/local/apps/environment-modules/current/init/bash +module load shared cmsh cm-nvfwupd >/dev/null 2>&1 +""" +_DISCOVER_SCRIPT = ( + _MODULE_SETUP + + """\ +exec cmsh-lazy-load -c 'device; list -t switch -f hostname:64,category:32,status:32' +""" +) +_QUERY_SCRIPT = ( + _MODULE_SETUP + + f"""\ +switch_host="$1" +bmc_creds=$(cmsh-lazy-load -c "device; use $switch_host; get ip; accesssettings; get username; get password") +set -- $bmc_creds +if [ "$#" -lt 3 ]; then + exit 20 +fi +bmc_ip="$1" +bmc_user="$2" +bmc_pass="$3" +exec {_NVFWUPD_PATH} \\ + -t ip="$bmc_ip" user="$bmc_user" password="$bmc_pass" servertype={_SERVER_TYPE} \\ + show_version -j +""" +) + + +class InspectionError(RuntimeError): + """Raised when direct NVSwitch firmware evidence cannot be obtained.""" + + +def _emit(result: dict[str, Any]) -> int: + """Print the provider-neutral JSON result and return its exit status.""" + print(json.dumps(result, indent=2, sort_keys=True)) + return 0 if result.get("success") else 1 + + +def _run_privileged(script: str, *args: str, timeout: int) -> subprocess.CompletedProcess[str]: + """Run one fixed read-only helper as root without exposing BMC credentials.""" + return subprocess.run( + ["sudo", "-n", "bash", "-s", "--", *args], + input=script, + text=True, + capture_output=True, + check=False, + timeout=timeout, + ) + + +def _validate_host(host: str) -> str: + """Reject values that cannot be safely used as BCM device names.""" + value = host.strip() + if not _HOST_RE.fullmatch(value): + raise InspectionError("invalid NVSwitch hostname") + return value + + +def _discover_switches(rack: str) -> list[str]: + """Return dedicated NVSwitch hostnames from BCM's read-only inventory.""" + completed = _run_privileged(_DISCOVER_SCRIPT, timeout=30) + if completed.returncode != 0: + raise InspectionError("unable to read NVSwitch inventory from BCM") + + rack_prefix = rack.strip().lower() + switches: list[str] = [] + for line in completed.stdout.splitlines(): + fields = line.split() + if len(fields) < 2 or "nvswitch" not in fields[1].lower(): + continue + host = _validate_host(fields[0]) + if rack_prefix and not host.lower().startswith(f"{rack_prefix}-"): + continue + if host not in switches: + switches.append(host) + return switches + + +def _query_tray(host: str) -> dict[str, Any]: + """Run ``nvfwupd show_version -j`` against one dedicated switch tray.""" + completed = _run_privileged(_QUERY_SCRIPT, _validate_host(host), timeout=120) + if completed.returncode != 0: + raise InspectionError("nvfwupd show_version failed") + try: + inventory = json.loads(completed.stdout) + except json.JSONDecodeError as exc: + raise InspectionError("nvfwupd returned invalid JSON") from exc + if not isinstance(inventory, dict) or str(inventory.get("Error Code", 0)) != "0": + raise InspectionError("nvfwupd returned a firmware inventory error") + + devices = inventory.get("Firmware Devices") + if not isinstance(devices, list) or not devices: + raise InspectionError("nvfwupd returned no firmware devices") + + versions: dict[str, str] = {} + for device in devices: + if not isinstance(device, dict): + raise InspectionError("nvfwupd returned a malformed firmware device") + name = str(device.get("AP Name") or "").strip() + version = str(device.get("Sys Version") or "").strip() + if not name or not version: + raise InspectionError("nvfwupd returned an incomplete firmware device") + versions[name] = version + + primary_version = versions.get("BMC") or versions.get("ASIC") or next(iter(versions.values())) + return { + "tray_id": host, + "firmware_version": primary_version, + "firmware_versions": versions, + } + + +def _configured_switches(cli_switches: Sequence[str]) -> list[str]: + """Combine repeatable CLI targets with the comma-separated environment value.""" + configured = list(cli_switches) + configured.extend(os.environ.get("GB300_SWITCH_HOSTS", "").split(",")) + switches: list[str] = [] + for candidate in configured: + if not candidate.strip(): + continue + host = _validate_host(candidate) + if host not in switches: + switches.append(host) + return switches + + +def _positive_int(value: str) -> int: + """Parse a strictly positive tray limit for argparse.""" + parsed = int(value) + if parsed < 1: + raise argparse.ArgumentTypeError("must be at least 1") + return parsed + + +def main() -> int: + """Inspect selected GB300 NVSwitch trays and emit BFX03-02 evidence.""" + parser = argparse.ArgumentParser(description="Inspect GB300 NVSwitch tray firmware with nvfwupd") + parser.add_argument("--rack", default=os.environ.get("GB300_RACK", ""), help="BCM rack hostname prefix") + parser.add_argument("--switch-host", action="append", default=[], help="NVSwitch hostname; repeat as needed") + parser.add_argument( + "--limit", + type=_positive_int, + default=os.environ.get("GB300_SWITCH_LIMIT", "1"), + help="Maximum number of trays to inspect (default: 1)", + ) + args = parser.parse_args() + + result: dict[str, Any] = { + "success": False, + "platform": "gb300", + "source": "nvfwupd show_version -j", + "trays": [], + } + try: + switches = _configured_switches(args.switch_host) + if not switches: + switches = _discover_switches(args.rack) + if not switches: + raise InspectionError("no NVSwitch trays discovered on the GB300 system") + result["trays"] = [_query_tray(host) for host in switches[: args.limit]] + except (InspectionError, subprocess.TimeoutExpired) as exc: + result["error_type"] = "firmware_inspection" + result["error"] = str(exc) if isinstance(exc, InspectionError) else "NVSwitch firmware inspection timed out" + return _emit(result) + + result["success"] = True + return _emit(result) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/isvctl/configs/providers/nico/config/bare_metal.yaml b/isvctl/configs/providers/nico/config/bare_metal.yaml index 54882f16b..1b84b2baf 100644 --- a/isvctl/configs/providers/nico/config/bare_metal.yaml +++ b/isvctl/configs/providers/nico/config/bare_metal.yaml @@ -364,7 +364,7 @@ commands: - name: query_switch_firmware phase: test continue_on_failure: true - command: "python ../scripts/breakfix/query_switch_firmware.py" + command: "python ../scripts/breakfix/gap_stub.py" args: - "--org" - "{{org}}" @@ -372,6 +372,8 @@ commands: - "{{site_id}}" - "--api-base" - "{{nico_api_base}}" + - "--gap" + - "BFX03-02" timeout: 120 - name: query_bmc_kernel_logs diff --git a/isvctl/configs/providers/nico/scripts/breakfix/gap_stub.py b/isvctl/configs/providers/nico/scripts/breakfix/gap_stub.py index b5df988c8..55fdc447e 100644 --- a/isvctl/configs/providers/nico/scripts/breakfix/gap_stub.py +++ b/isvctl/configs/providers/nico/scripts/breakfix/gap_stub.py @@ -6,7 +6,7 @@ Several break-fix requirements have no NICo tenant REST surface to exercise: the mutating BFX01 workflows run through Maestro/repair fixtures, and the -BFX02-02/BFX04-01/BFX05/BFX06 signals are not exposed at all. Each of +BFX02-02/BFX03-02/BFX04-01/BFX05/BFX06 signals are not exposed at all. Each of those steps emits a structured skip naming the gap rather than a hard failure, so the suite reports "not available on this platform" instead of "broken". @@ -44,6 +44,10 @@ "NICo has no retirement-notice query API (BFX02-02 gap)", {"notices_queryable": False, "notices": []}, ), + "BFX03-02": ( + "NV switch tray firmware is not queryable via NICo tenant REST API (BFX03-02 gap)", + {"trays": []}, + ), "BFX04-01": ( "GPUd/Sentinel/Maestro node health agents are not observable via NICo REST (BFX04-01 gap)", {"agents_observable": False, "agents": []}, diff --git a/isvctl/configs/providers/nico/scripts/breakfix/query_switch_firmware.py b/isvctl/configs/providers/nico/scripts/breakfix/query_switch_firmware.py deleted file mode 100755 index c40b01259..000000000 --- a/isvctl/configs/providers/nico/scripts/breakfix/query_switch_firmware.py +++ /dev/null @@ -1,120 +0,0 @@ -#!/usr/bin/env python3 -# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -# SPDX-License-Identifier: Apache-2.0 - -"""Query NVSwitch tray firmware versions from NICo Flow (BFX03-02). - -NICo's read-only tray list endpoint returns every tray at a Flow-enabled site. -The provider filters the version-specific tray type values client-side, and -NVSwitch trays expose their installed firmware through ``firmwareVersion``. -""" - -from __future__ import annotations - -import argparse -import re -import sys -from pathlib import Path -from typing import Any -from urllib.error import HTTPError, URLError - -sys.path.insert(0, str(Path(__file__).resolve().parents[1])) -from breakfix._common import emit, skip_result -from common.nico_client import NicoAuthError, forge_get_all, resolve_auth - -_FLOW_DISABLED_MESSAGE = "site does not have nico flow enabled" - - -def _is_nvswitch(component: dict[str, Any]) -> bool: - """Return whether a tray is explicitly typed as an NVSwitch.""" - component_type = re.sub(r"[^a-z0-9]", "", str(component.get("type") or "").lower()) - return component_type in {"switch", "nvswitch", "componenttypenvswitch"} - - -def _tray_id(component: dict[str, Any]) -> str: - """Choose the first stable, human-useful identifier NICo provides.""" - for field in ("componentId", "id", "serialNumber", "name"): - value = component.get(field) - if value is not None and str(value).strip(): - return str(value) - return "" - - -def _flow_disabled(exc: HTTPError) -> bool: - """Return whether NICo rejected the query because Flow is disabled.""" - return exc.code == 412 and _FLOW_DISABLED_MESSAGE in str(exc).lower() - - -def main() -> int: - """List NVSwitch tray firmware versions as provider-neutral JSON.""" - parser = argparse.ArgumentParser(description="Query NICo NVSwitch tray firmware versions") - parser.add_argument("--org", required=True) - parser.add_argument("--site-id", required=True) - parser.add_argument("--api-base", required=True) - args = parser.parse_args() - - result: dict[str, Any] = { - "success": False, - "platform": "nico", - "site_id": args.site_id, - "trays": [], - } - try: - auth = resolve_auth() - components = forge_get_all( - args.org, - "tray", - auth.token, - base_url=args.api_base, - params={"siteId": args.site_id}, - result_key="trays", - ) - except NicoAuthError as exc: - result.update(error_type="auth", error=str(exc)) - return emit(result) - except HTTPError as exc: - if _flow_disabled(exc): - skip = skip_result( - args.site_id, - "NICo Flow is not enabled for this site; NVSwitch tray firmware is unavailable (BFX03-02 gap)", - gap="BFX03-02", - ) - skip["trays"] = [] - return emit(skip) - result.update(error_type="api", error=f"NICo tray query failed (HTTP {exc.code})") - return emit(result) - except (URLError, ValueError) as exc: - result["error"] = f"{type(exc).__name__}: {exc}" - return emit(result) - - seen: set[str] = set() - trays: list[dict[str, Any]] = [] - for component in components: - if not isinstance(component, dict) or not _is_nvswitch(component): - continue - tray_id = _tray_id(component) - if tray_id and tray_id in seen: - continue - if tray_id: - seen.add(tray_id) - trays.append( - { - "tray_id": tray_id, - "firmware_version": str(component.get("firmwareVersion") or ""), - } - ) - - if not trays: - result.update( - success=True, - skipped=True, - skip_reason="No NVSwitch tray components were returned for this NICo site", - ) - return emit(result) - - result.update(success=True, trays=trays) - return emit(result) - - -if __name__ == "__main__": - sys.exit(main()) diff --git a/isvctl/tests/providers/gb300/test_gb300_provider.py b/isvctl/tests/providers/gb300/test_gb300_provider.py new file mode 100644 index 000000000..e34a8410c --- /dev/null +++ b/isvctl/tests/providers/gb300/test_gb300_provider.py @@ -0,0 +1,208 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for direct GB300 NVSwitch tray firmware inspection.""" + +from __future__ import annotations + +import importlib.util +import json +import subprocess +import sys +from pathlib import Path +from types import ModuleType +from typing import Any + +import pytest +import yaml +from isvtest.validations.breakfix import NvSwitchFirmwareCheck + +ISVCTL_ROOT = Path(__file__).resolve().parents[3] +GB300_ROOT = ISVCTL_ROOT / "configs" / "providers" / "gb300" +SCRIPT_PATH = GB300_ROOT / "scripts" / "breakfix" / "query_switch_firmware.py" +CONFIG_PATH = GB300_ROOT / "config" / "bare_metal.yaml" + +LIVE_SHAPED_INVENTORY = { + "Error Code": 0, + "Firmware Devices": [ + {"AP Name": "ASIC", "Sys Version": "35.2014.4784"}, + {"AP Name": "BIOS", "Sys Version": "0ACTV_00.01.020"}, + {"AP Name": "BMC", "Sys Version": "88.0002.1961"}, + {"AP Name": "CPLD1", "Sys Version": "CPLD000420_REV0300"}, + {"AP Name": "CPLD2", "Sys Version": "CPLD000419_REV0301"}, + {"AP Name": "CPLD3", "Sys Version": "CPLD000418_REV0200"}, + {"AP Name": "EROT", "Sys Version": "01.04.0031.0000_n04"}, + {"AP Name": "FPGA", "Sys Version": "0.24"}, + {"AP Name": "SSD", "Sys Version": "CE00A450"}, + ], +} + + +def _load_script() -> ModuleType: + """Load the provider script as an isolated module.""" + spec = importlib.util.spec_from_file_location("test_gb300_switch_firmware", SCRIPT_PATH) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _completed(stdout: str, *, returncode: int = 0, stderr: str = "") -> subprocess.CompletedProcess[str]: + """Build a subprocess result for privileged-helper mocks.""" + return subprocess.CompletedProcess([], returncode, stdout=stdout, stderr=stderr) + + +def test_config_wires_only_read_only_firmware_query() -> None: + """The GB300 provider binds BFX03-02 to the direct query script.""" + config = yaml.safe_load(CONFIG_PATH.read_text(encoding="utf-8")) + steps = config["commands"]["bare_metal"]["steps"] + + assert steps == [ + { + "name": "query_switch_firmware", + "phase": "test", + "continue_on_failure": True, + "command": "python ../scripts/breakfix/query_switch_firmware.py", + "timeout": 180, + } + ] + + +def test_discovers_dedicated_nvswitches_in_requested_rack(monkeypatch: pytest.MonkeyPatch) -> None: + """BCM compute and other-rack devices never become firmware targets.""" + module = _load_script() + inventory = """\ +a05-p01-dgx-01-c01 compute [ UP ] +a05-p01-nvsw-01 nvswitch [ UP ] +a05-p01-nvsw-02 nvswitch [ UP ] +b04-p01-nvsw-01 nvswitch [ UP ] +""" + monkeypatch.setattr(module, "_run_privileged", lambda *args, **kwargs: _completed(inventory)) + + assert module._discover_switches("a05-p01") == ["a05-p01-nvsw-01", "a05-p01-nvsw-02"] + + +def test_queries_live_shaped_nvfwupd_inventory_read_only(monkeypatch: pytest.MonkeyPatch) -> None: + """The observed GB300 switch shape produces complete BFX03-02 evidence.""" + module = _load_script() + observed: dict[str, Any] = {} + + def fake_run(script: str, *args: str, timeout: int) -> subprocess.CompletedProcess[str]: + observed.update(script=script, args=args, timeout=timeout) + return _completed(json.dumps(LIVE_SHAPED_INVENTORY)) + + monkeypatch.setattr(module, "_run_privileged", fake_run) + + tray = module._query_tray("a05-p01-nvsw-01") + + assert tray["tray_id"] == "a05-p01-nvsw-01" + assert tray["firmware_version"] == "88.0002.1961" + assert tray["firmware_versions"]["ASIC"] == "35.2014.4784" + assert tray["firmware_versions"]["CPLD3"] == "CPLD000418_REV0200" + assert observed["args"] == ("a05-p01-nvsw-01",) + assert observed["timeout"] == 120 + assert "show_version -j" in observed["script"] + assert all(token not in observed["script"] for token in ("update_fw", "update_firmware", "activate_fw")) + + +def test_main_passes_with_direct_gb300_tray_evidence( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """One inspected GB300 NVSwitch tray satisfies the suite's minimum.""" + module = _load_script() + monkeypatch.delenv("GB300_SWITCH_HOSTS", raising=False) + monkeypatch.setattr(module, "_discover_switches", lambda rack: ["a05-p01-nvsw-01"]) + monkeypatch.setattr( + module, + "_query_tray", + lambda host: { + "tray_id": host, + "firmware_version": "88.0002.1961", + "firmware_versions": {"BMC": "88.0002.1961", "ASIC": "35.2014.4784"}, + }, + ) + monkeypatch.setattr(sys, "argv", ["query_switch_firmware.py", "--rack", "a05-p01"]) + + assert module.main() == 0 + + payload = json.loads(capsys.readouterr().out) + assert payload == { + "platform": "gb300", + "source": "nvfwupd show_version -j", + "success": True, + "trays": [ + { + "tray_id": "a05-p01-nvsw-01", + "firmware_version": "88.0002.1961", + "firmware_versions": {"BMC": "88.0002.1961", "ASIC": "35.2014.4784"}, + } + ], + } + + check = NvSwitchFirmwareCheck(config={"step_output": payload}) + check.run() + assert check.passed + assert "1 NV switch tray(s)" in check.message + + +def test_assumed_gb300_without_switch_inventory_fails( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """Applicable GB300 hardware cannot skip or pass without tray evidence.""" + module = _load_script() + monkeypatch.delenv("GB300_SWITCH_HOSTS", raising=False) + monkeypatch.setattr(module, "_discover_switches", lambda rack: []) + monkeypatch.setattr(sys, "argv", ["query_switch_firmware.py", "--rack", "a05-p01"]) + + assert module.main() == 1 + + payload = json.loads(capsys.readouterr().out) + assert payload["success"] is False + assert payload["trays"] == [] + assert payload["error_type"] == "firmware_inspection" + assert "no NVSwitch trays discovered" in payload["error"] + assert "skipped" not in payload + + +def test_incomplete_nvfwupd_device_is_rejected(monkeypatch: pytest.MonkeyPatch) -> None: + """A reported firmware component without a version is not a PASS.""" + module = _load_script() + incomplete = { + "Error Code": 0, + "Firmware Devices": [ + {"AP Name": "BMC", "Sys Version": "88.0002.1961"}, + {"AP Name": "ASIC", "Sys Version": ""}, + ], + } + monkeypatch.setattr( + module, + "_run_privileged", + lambda *args, **kwargs: _completed(json.dumps(incomplete)), + ) + + with pytest.raises(module.InspectionError, match="incomplete firmware device"): + module._query_tray("a05-p01-nvsw-01") + + +def test_privileged_failure_does_not_leak_stderr( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """BMC credentials or raw command diagnostics never enter provider JSON.""" + module = _load_script() + monkeypatch.delenv("GB300_SWITCH_HOSTS", raising=False) + monkeypatch.setattr(module, "_discover_switches", lambda rack: ["a05-p01-nvsw-01"]) + monkeypatch.setattr( + module, + "_run_privileged", + lambda *args, **kwargs: _completed("", returncode=1, stderr="password=do-not-print"), + ) + monkeypatch.setattr(sys, "argv", ["query_switch_firmware.py", "--rack", "a05-p01"]) + + assert module.main() == 1 + + output = capsys.readouterr().out + assert "do-not-print" not in output + assert "password=" not in output diff --git a/isvctl/tests/providers/nico/test_nico_provider.py b/isvctl/tests/providers/nico/test_nico_provider.py index 2c00560b1..40c1cc708 100644 --- a/isvctl/tests/providers/nico/test_nico_provider.py +++ b/isvctl/tests/providers/nico/test_nico_provider.py @@ -141,14 +141,6 @@ def _load_nico_script(relative_path: str, module_name: str) -> ModuleType: return module -def _load_switch_firmware_script() -> ModuleType: - """Load the BFX03-02 NVSwitch firmware query script.""" - return _load_nico_script( - "breakfix/query_switch_firmware.py", - "test_query_switch_firmware", - ) - - def _load_governance_metrics_script() -> ModuleType: """Load the query_metrics (governance) script as a module for direct unit testing.""" script_path = NICO_SCRIPTS / "governance" / "query_metrics.py" @@ -1103,200 +1095,6 @@ def test_nico_scripts_require_api_base( assert "--api-base" in captured.err -def test_switch_firmware_queries_all_trays_and_filters_nvswitches( - monkeypatch: pytest.MonkeyPatch, - capsys: pytest.CaptureFixture[str], -) -> None: - """BFX03-02 should query all trays and accept each deployed NVSwitch type spelling.""" - module = _load_switch_firmware_script() - observed: dict[str, Any] = {} - - def fake_get_all(*args: Any, **kwargs: Any) -> list[dict[str, Any]]: - observed["args"] = args - observed["kwargs"] = kwargs - return [ - { - "id": "internal-switch-1", - "componentId": "switch-1", - "type": "switch", - "firmwareVersion": "1.2.3", - "rackId": "rack-1", - "position": {"slotId": 3, "trayIdx": 0}, - }, - { - "id": "switch-2", - "type": "NVSwitch", - "firmwareVersion": None, - "rackId": "rack-2", - "position": {"slotId": 5, "trayIdx": 1}, - }, - { - "id": "switch-3", - "type": "ComponentTypeNVSwitch", - "firmwareVersion": "3.4.5", - "rackId": "rack-2", - }, - { - "id": "compute-1", - "type": "compute", - "firmwareVersion": "9.9.9", - "rackId": "rack-2", - }, - ] - - monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="test-token")) - monkeypatch.setattr(module, "forge_get_all", fake_get_all) - monkeypatch.setattr( - sys, - "argv", - [ - "query_switch_firmware.py", - "--org", - "test-org", - "--site-id", - "site-1", - "--api-base", - "https://nico.example/v2/org", - ], - ) - - assert module.main() == 0 - - payload = json.loads(capsys.readouterr().out) - assert payload["success"] is True - assert payload["trays"] == [ - { - "tray_id": "switch-1", - "firmware_version": "1.2.3", - }, - { - "tray_id": "switch-2", - "firmware_version": "", - }, - { - "tray_id": "switch-3", - "firmware_version": "3.4.5", - }, - ] - assert observed["args"] == ("test-org", "tray", "test-token") - assert observed["kwargs"] == { - "base_url": "https://nico.example/v2/org", - "params": {"siteId": "site-1"}, - "result_key": "trays", - } - - -def test_switch_firmware_skips_when_site_has_no_nvswitch_components( - monkeypatch: pytest.MonkeyPatch, - capsys: pytest.CaptureFixture[str], -) -> None: - """A site without NVSwitch inventory is inapplicable, not a false pass.""" - module = _load_switch_firmware_script() - monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="test-token")) - monkeypatch.setattr( - module, - "forge_get_all", - lambda *args, **kwargs: [{"id": "compute-1", "type": "compute"}], - ) - monkeypatch.setattr( - sys, - "argv", - [ - "query_switch_firmware.py", - "--org", - "test-org", - "--site-id", - "site-1", - "--api-base", - "https://nico.example/v2/org", - ], - ) - - assert module.main() == 0 - - payload = json.loads(capsys.readouterr().out) - assert payload["success"] is True - assert payload["skipped"] is True - assert payload["trays"] == [] - assert "No NVSwitch tray components" in payload["skip_reason"] - - -def test_switch_firmware_skips_when_nico_flow_is_disabled( - monkeypatch: pytest.MonkeyPatch, - capsys: pytest.CaptureFixture[str], -) -> None: - """NICo's Flow-disabled precondition is a documented runtime gap, not a pass.""" - module = _load_switch_firmware_script() - - def flow_disabled(*args: Any, **kwargs: Any) -> list[dict[str, Any]]: - raise HTTPError( - "https://nico.example/v2/org/test-org/nico/tray", - 412, - "Site does not have NICo Flow enabled", - None, - None, - ) - - monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="test-token")) - monkeypatch.setattr(module, "forge_get_all", flow_disabled) - monkeypatch.setattr( - sys, - "argv", - [ - "query_switch_firmware.py", - "--org", - "test-org", - "--site-id", - "site-1", - "--api-base", - "https://nico.example/v2/org", - ], - ) - - assert module.main() == 0 - - payload = json.loads(capsys.readouterr().out) - assert payload["success"] is True - assert payload["skipped"] is True - assert payload["gap"] == "BFX03-02" - assert payload["trays"] == [] - assert "Flow is not enabled" in payload["skip_reason"] - - -def test_switch_firmware_does_not_skip_other_http_errors( - monkeypatch: pytest.MonkeyPatch, - capsys: pytest.CaptureFixture[str], -) -> None: - """Authentication and API authorization failures must remain failures.""" - module = _load_switch_firmware_script() - - def forbidden(*args: Any, **kwargs: Any) -> list[dict[str, Any]]: - raise HTTPError("https://nico.example/tray", 403, "Forbidden", None, None) - - monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="test-token")) - monkeypatch.setattr(module, "forge_get_all", forbidden) - monkeypatch.setattr( - sys, - "argv", - [ - "query_switch_firmware.py", - "--org", - "test-org", - "--site-id", - "site-1", - "--api-base", - "https://nico.example/v2/org", - ], - ) - - assert module.main() == 1 - - payload = json.loads(capsys.readouterr().out) - assert payload["success"] is False - assert payload["error_type"] == "api" - assert "skipped" not in payload - - def test_dpu_health_script_treats_nullable_machine_lists_as_empty( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], diff --git a/isvtest/tests/test_breakfix.py b/isvtest/tests/test_breakfix.py index c4465815c..daf254604 100644 --- a/isvtest/tests/test_breakfix.py +++ b/isvtest/tests/test_breakfix.py @@ -150,17 +150,11 @@ def test_node_maintenance_reports_mode(self) -> None: class TestNvSwitchFirmwareCheck: """Cover BFX03-02 firmware evidence for every returned switch tray.""" - def test_propagates_flow_disabled_runtime_skip(self) -> None: - """A Flow-disabled site remains skipped instead of becoming a pass.""" - step_output = { - "success": True, - "skipped": True, - "skip_reason": "NICo Flow is not enabled for this site", - "gap": "BFX03-02", - "trays": [], - } - with pytest.raises(pytest.skip.Exception): - _run(NvSwitchFirmwareCheck, step_output) + def test_fails_when_no_trays_are_returned(self) -> None: + """An applicable GB300 system cannot pass without NVSwitch tray evidence.""" + check = _run(NvSwitchFirmwareCheck, {"success": True, "trays": []}) + assert not check.passed + assert "Expected at least 1 NV switch tray(s), got 0" in check.message def test_passes_when_every_tray_has_firmware(self) -> None: """Every discovered NVSwitch tray must report a non-empty version."""