Upgrade smartswitch via gNOI testcases

tests/common/fixtures/grpc_fixtures.py

@@ -37,7 +37,7 @@ def test_grpc_services(ptf_grpc):
 
     # Auto-configure using GNMIEnvironment
     env = GNMIEnvironment(duthost, GNMIEnvironment.GNMI_MODE)
-    client = PtfGrpc(ptfhost, env, duthost=duthost)
+    client = PtfGrpc(ptfhost, env, duthost=duthost, insecure=True)  # Use insecure=True to skip TLS verification if TLS is detected but certs not configured
 
     logger.info(f"Created auto-configured gRPC client: {client}")
     return client
@@ -298,6 +298,7 @@ def _configure_gnoi_tls_server(duthost):
     duthost.shell('sonic-db-cli CONFIG_DB hset "GNMI|gnmi" port 50052')
     duthost.shell('sonic-db-cli CONFIG_DB hset "GNMI|gnmi" client_auth true')
     duthost.shell('sonic-db-cli CONFIG_DB hset "GNMI|gnmi" log_level 2')
+    duthost.shell('sonic-db-cli CONFIG_DB hset "GNMI|gnmi" user_auth cert')
 
     # Configure certificate paths using centralized config
     config_db_settings = grpc_config.get_config_db_cert_settings()

tests/common/helpers/upgrade_helpers.py

@@ -8,11 +8,12 @@
 import tests.common.fixtures.grpc_fixtures  # noqa: F401
 from tests.common.helpers.assertions import pytest_assert
 from tests.common import reboot
+from tests.common.helpers.multi_thread_utils import SafeThreadPoolExecutor
 from tests.common.reboot import get_reboot_cause, reboot_ctrl_dict
 from tests.common.reboot import REBOOT_TYPE_WARM, REBOOT_TYPE_COLD
 from tests.common.utilities import wait_until, setup_ferret
 from tests.common.platform.device_utils import check_neighbors
-from typing import Dict, Optional
+from typing import Optional, Sequence, Tuple, Union, Dict
 
 SYSTEM_STABILIZE_MAX_TIME = 300
 logger = logging.getLogger(__name__)
@@ -22,7 +23,10 @@
 TMP_PORTS_FILE = '/tmp/ports.json'
 TMP_PEER_INFO_FILE = "/tmp/peer_dev_info.json"
 TMP_PEER_PORT_INFO_FILE = "/tmp/neigh_port_info.json"
+SS_TARGET_TYPE_HDR = "x-sonic-ss-target-type"
+SS_TARGET_INDEX_HDR = "x-sonic-ss-target-index"
 
+GrpcMetadata = Union[Dict[str, str], Sequence[Tuple[str, str]]]
 
 @dataclass(frozen=True)
 class GnoiUpgradeConfig:
@@ -32,6 +36,17 @@ class GnoiUpgradeConfig:
     protocol: str = "HTTP"
     allow_fail: bool = False
     to_version: Optional[str] = None  # Optional expected version string to validate after upgrade
+    ss_target_type: Optional[str] = None   # e.g. "dpu"
+    ss_target_index: Optional[int] = None  # e.g. 3
+    metadata: Optional[GrpcMetadata] = None
+    ss_reboot_ready_timeout: int = 1200
+    ss_reboot_message: str = "Rebooting DPU for maintenance"
+
+def build_smartswitch_metadata(target_type: str, target_index: int):
+    return [
+        (SS_TARGET_TYPE_HDR, str(target_type)),
+        (SS_TARGET_INDEX_HDR, str(target_index)),
+    ]
 
 
 def pytest_runtest_setup(item):
@@ -395,3 +410,81 @@ def perform_gnoi_upgrade(
     )
 
     return {"transfer_resp": transfer_resp, "setpkg_resp": setpkg_resp}
+
+def _wait_gnoi_time_ready(ptf_gnoi, metadata, cfg: GnoiUpgradeConfig, timeout=None, interval: int = 10) -> bool:
+    timeout = timeout or cfg.ss_reboot_ready_timeout
+    return wait_until(timeout, interval, 0, ptf_gnoi.system_time, metadata=metadata)
+
+
+def _upgrade_one_dpu_via_gnoi(duthost, tbinfo, ptf_gnoi, cfg: GnoiUpgradeConfig) -> Dict:
+    if not cfg.metadata:
+        raise ValueError("cfg.metadata must be provided for SmartSwitch DPU upgrade")
+
+    md = cfg.metadata
+
+    ptf_gnoi.system_time(metadata=md)
+
+    transfer_resp = ptf_gnoi.file_transfer_to_remote(
+        url=cfg.to_image,
+        local_path=cfg.dut_image_path,
+        protocol=cfg.protocol,
+        metadata=md,
+    )
+
+    setpkg_resp = ptf_gnoi.system_set_package(
+        local_path=cfg.dut_image_path,
+        version=cfg.to_version,
+        activate=True,
+        metadata=md,
+    )
+
+    # method = str(cfg.upgrade_type).upper()
+    try:
+        ## TODO: switch to use the perform reboot helper once it supports gNOI.
+        reboot_resp = ptf_gnoi.system_reboot(
+            method=cfg.upgrade_type,
+            delay=0,
+            message=cfg.ss_reboot_message,
+            metadata=md,
+        )
+    except Exception as e:
+        logger.info("Reboot raised (often expected): %s", e)
+        reboot_resp = None
+
+    if cfg.allow_fail:
+        return {"transfer_resp": transfer_resp, "setpkg_resp": setpkg_resp, "reboot_resp": reboot_resp}
+
+    ok = _wait_gnoi_time_ready(ptf_gnoi, md, cfg)
+    pytest_assert(ok, f"gNOI Time not reachable within {cfg.ss_reboot_ready_timeout}s after reboot")
+    check_services(duthost, tbinfo)
+    check_neighbors(duthost, tbinfo)
+    check_copp_config(duthost)
+
+    return {"transfer_resp": transfer_resp, "setpkg_resp": setpkg_resp, "reboot_resp": reboot_resp}
+
+
+def perform_gnoi_upgrade_smartswitch_dpu(duthost, tbinfo, ptf_gnoi, cfg: GnoiUpgradeConfig) -> Dict:
+    return _upgrade_one_dpu_via_gnoi(duthost, tbinfo, ptf_gnoi, cfg)
+
+
+def perform_gnoi_upgrade_smartswitch_dpus_parallel(
+    duthost, tbinfo,
+    ptf_gnoi,
+    cfgs: Sequence[GnoiUpgradeConfig],
+    max_workers: Optional[int] = None,
+) -> Dict[int, Dict]:
+    if not cfgs:
+        raise ValueError("cfgs is empty")
+
+    workers = max_workers or len(cfgs)
+    results: Dict[int, Dict] = {}
+
+    with SafeThreadPoolExecutor(max_workers=workers) as executor:
+        futs = {}
+        for i, cfg in enumerate(cfgs):
+            futs[i] = executor.submit(_upgrade_one_dpu_via_gnoi, duthost, tbinfo, ptf_gnoi, cfg)
+
+        for i, fut in futs.items():
+            results[i] = fut.result()
+
+    return results

tests/common/ptf_gnoi.py

@@ -31,7 +31,7 @@ def __init__(self, grpc_client):
         self.grpc_client = grpc_client
         logger.info(f"Initialized PtfGnoi wrapper with {grpc_client}")
 
-    def system_time(self) -> Dict:
+    def system_time(self, metadata=None) -> Dict:
         """
         Get the current system time from the device.
 
@@ -47,7 +47,7 @@ def system_time(self) -> Dict:
         logger.debug("Getting system time via gNOI System.Time")
 
         # Make the low-level gRPC call
-        response = self.grpc_client.call_unary("gnoi.system.System", "Time")
+        response = self.grpc_client.call_unary("gnoi.system.System", "Time", metadata=metadata)
 
         # Convert time string to int for consistency
         if "time" in response:
@@ -63,7 +63,7 @@ def system_time(self) -> Dict:
     # TODO: Add file_get(), file_put(), file_remove() methods
     # These are left for future implementation when gNOI File service is stable
 
-    def file_stat(self, remote_file: str) -> Dict:
+    def file_stat(self, remote_file: str, metadata=None) -> Dict:
         """
         Get file statistics from the device.
 
@@ -84,7 +84,7 @@ def file_stat(self, remote_file: str) -> Dict:
         request = {"path": remote_file}
 
         try:
-            response = self.grpc_client.call_unary("gnoi.file.File", "Stat", request)
+            response = self.grpc_client.call_unary("gnoi.file.File", "Stat", request, metadata=metadata)
 
             # Convert numeric strings to proper types for consistency
             if "stats" in response and isinstance(response["stats"], list):
@@ -117,6 +117,7 @@ def file_transfer_to_remote(
         local_path: str,
         protocol: Optional[str] = None,
         credentials: Optional[Dict[str, str]] = None,
+        metadata = None
     ) -> Dict:
         """
         Download a remote artifact to the DUT using gNOI File.TransferToRemote.
@@ -182,11 +183,11 @@ def file_transfer_to_remote(
             remote_download["credentials"] = credentials
 
         request = {
-            "localPath": local_path,
-            "remoteDownload": remote_download,
+            "local_path": local_path,
+            "remote_download": remote_download,
         }
 
-        response = self.grpc_client.call_unary("gnoi.file.File", "TransferToRemote", request)
+        response = self.grpc_client.call_unary("gnoi.file.File", "TransferToRemote", request, metadata=metadata)
         logger.info("TransferToRemote completed: %s -> %s", url, local_path)
         return response
 
@@ -195,6 +196,7 @@ def system_set_package(
         local_path: str,
         version: Optional[str] = None,
         activate: bool = True,
+        metadata = None,
     ) -> Dict:
         """
         Set the upgrade package on the DUT using gNOI System.SetPackage (client-streaming RPC).
@@ -233,6 +235,7 @@ def system_set_package(
             "gnoi.system.System",
             "SetPackage",
             [{"package": pkg}],
+            metadata=metadata,
         )
 
         logger.info("SetPackage completed: %s", local_path)
@@ -244,6 +247,7 @@ def system_reboot(
         delay: Optional[int] = None,
         message: Optional[str] = None,
         force: bool = False,
+        metadata=None,
     ) -> Dict:
         """
         Reboot the DUT using gNOI System.Reboot.
@@ -290,6 +294,6 @@ def system_reboot(
 
         logger.debug("Reboot via gNOI System.Reboot: %s", request)
 
-        response = self.grpc_client.call_unary("gnoi.system.System", "Reboot", request)
+        response = self.grpc_client.call_unary("gnoi.system.System", "Reboot", request, metadata=metadata)
         logger.info("Reboot request sent: method=%s delay=%s force=%s", method, delay, force)
         return response