Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 1 addition & 3 deletions isvctl/configs/providers/nico/config/bare_metal.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -364,16 +364,14 @@ 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}}"
- "--site-id"
- "{{site_id}}"
- "--api-base"
- "{{nico_api_base}}"
- "--gap"
- "BFX03-02"
timeout: 120

- name: query_bmc_kernel_logs
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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".

Expand Down Expand Up @@ -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": []},
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,120 @@
#!/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())
202 changes: 202 additions & 0 deletions isvctl/tests/providers/nico/test_nico_provider.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -1095,6 +1103,200 @@ 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],
Expand Down
Loading