NetworkPlugin update: adding url collector arg and test enhancements

nodescraper/cli/dynamicparserbuilder.py

@@ -24,7 +24,7 @@
 #
 ###############################################################################
 import argparse
-from typing import Optional, Type
+from typing import Literal, Optional, Type, get_args, get_origin
 
 from pydantic import BaseModel
 
@@ -96,19 +96,47 @@ def get_model_arg(cls, type_class_map: dict) -> Optional[Type[BaseModel]]:
             None,
         )
 
+    @classmethod
+    def get_literal_choices(cls, type_class_map: dict) -> Optional[list]:
+        """Get the choices from a Literal type if present
+
+        Args:
+            type_class_map (dict): mapping of type classes
+
+        Returns:
+            Optional[list]: list of valid choices for the Literal type, or None if not a Literal
+        """
+        # Check if Literal is in the type_class_map
+        literal_type = type_class_map.get(Literal)
+        if literal_type and literal_type.inner_type is not None:
+            return None
+        return None
+
     def add_argument(
         self,
         type_class_map: dict,
         arg_name: str,
         required: bool,
+        annotation: Optional[Type] = None,
     ) -> None:
         """Add an argument to a parser with an appropriate type
 
         Args:
             type_class_map (dict): type classes for the arg
             arg_name (str): argument name
             required (bool): whether or not the arg is required
+            annotation (Optional[Type]): full type annotation for extracting Literal choices
         """
+        # Check for Literal types and extract choices
+        literal_choices = None
+        if Literal in type_class_map and annotation:
+            # Extract all arguments from the annotation
+            args = get_args(annotation)
+            for arg in args:
+                if get_origin(arg) is Literal:
+                    literal_choices = list(get_args(arg))
+                    break
+
         if list in type_class_map:
             type_class = type_class_map[list]
             self.parser.add_argument(
@@ -125,6 +153,15 @@ def add_argument(
                 required=required,
                 choices=[True, False],
             )
+        elif Literal in type_class_map and literal_choices:
+            # Add argument with choices for Literal types
+            self.parser.add_argument(
+                f"--{arg_name}",
+                type=str,
+                required=required,
+                choices=literal_choices,
+                metavar=f"{{{','.join(literal_choices)}}}",
+            )
         elif float in type_class_map:
             self.parser.add_argument(
                 f"--{arg_name}", type=float, required=required, metavar=META_VAR_MAP[float]
@@ -166,6 +203,10 @@ def build_model_arg_parser(self, model: type[BaseModel], required: bool) -> list
             if type(None) in type_class_map and len(attr_data.type_classes) == 1:
                 continue
 
-            self.add_argument(type_class_map, attr.replace("_", "-"), required)
+            # Get the full annotation from the model field
+            field = model.model_fields.get(attr)
+            annotation = field.annotation if field else None
+
+            self.add_argument(type_class_map, attr.replace("_", "-"), required, annotation)
 
         return list(type_map.keys())

nodescraper/plugins/inband/network/collector_args.py

@@ -0,0 +1,34 @@
+###############################################################################
+#
+# MIT License
+#
+# Copyright (c) 2025 Advanced Micro Devices, Inc.
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+#
+###############################################################################
+
+from typing import Literal, Optional
+
+from nodescraper.models import CollectorArgs
+
+
+class NetworkCollectorArgs(CollectorArgs):
+    url: Optional[str] = None
+    netprobe: Optional[Literal["ping", "wget", "curl"]] = None

nodescraper/plugins/inband/network/network_collector.py

@@ -27,9 +27,10 @@
 from typing import Dict, List, Optional, Tuple
 
 from nodescraper.base import InBandDataCollector
-from nodescraper.enums import EventCategory, EventPriority, ExecutionStatus
+from nodescraper.enums import EventCategory, EventPriority, ExecutionStatus, OSFamily
 from nodescraper.models import TaskResult
 
+from .collector_args import NetworkCollectorArgs
 from .networkdata import (
     BroadcomNicDevice,
     BroadcomNicQos,
@@ -55,7 +56,7 @@
 )
 
 
-class NetworkCollector(InBandDataCollector[NetworkDataModel, None]):
+class NetworkCollector(InBandDataCollector[NetworkDataModel, NetworkCollectorArgs]):
     """Collect network configuration details using ip command"""
 
     DATA_MODEL = NetworkDataModel
@@ -64,6 +65,9 @@ class NetworkCollector(InBandDataCollector[NetworkDataModel, None]):
     CMD_RULE = "ip rule show"
     CMD_NEIGHBOR = "ip neighbor show"
     CMD_ETHTOOL_TEMPLATE = "ethtool {interface}"
+    CMD_PING = "ping"
+    CMD_WGET = "wget"
+    CMD_CURL = "curl"
 
     # LLDP commands
     CMD_LLDPCLI_NEIGHBOR = "lldpcli show neighbor"
@@ -1669,12 +1673,59 @@ def _collect_pensando_nic_info(
             uncollected_commands,
         )
 
+    def _check_network_connectivity(self, cmd: str, url: str) -> bool:
+        """Check network connectivity using specified command.
+
+        Args:
+            cmd: Command to use for connectivity check (ping, wget, or curl)
+            url: URL or hostname to check
+
+        Returns:
+            bool: True if network is accessible, False otherwise
+        """
+        if cmd not in {"ping", "wget", "curl"}:
+            raise ValueError(
+                f"Invalid network probe command: '{cmd}'. "
+                f"Valid options are: 'ping', 'wget', 'curl'"
+            )
+
+        # Determine ping options based on OS
+        ping_option = "-c 1" if self.system_info.os_family == OSFamily.LINUX else "-n 1"
+
+        # Build command based on cmd parameter using class constants
+        if cmd == "ping":
+            result = self._run_sut_cmd(f"{self.CMD_PING} {url} {ping_option}")
+        elif cmd == "wget":
+            result = self._run_sut_cmd(f"{self.CMD_WGET} {url}")
+        else:  # curl
+            result = self._run_sut_cmd(f"{self.CMD_CURL} {url}")
+
+        if result.exit_code == 0:
+            self._log_event(
+                category=EventCategory.NETWORK,
+                description="System networking is up",
+                data={"url": url, "accessible": result.exit_code == 0},
+                priority=EventPriority.INFO,
+            )
+        else:
+            self._log_event(
+                category=EventCategory.NETWORK,
+                description=f"{cmd} to {url} failed!",
+                data={"url": url, "not accessible": result.exit_code == 0},
+                priority=EventPriority.ERROR,
+            )
+
+        return result.exit_code == 0
+
     def collect_data(
         self,
-        args=None,
+        args: Optional[NetworkCollectorArgs] = None,
     ) -> Tuple[TaskResult, Optional[NetworkDataModel]]:
         """Collect network configuration from the system.
 
+        Args:
+            args: Optional NetworkCollectorArgs with URL for network connectivity check
+
         Returns:
             Tuple[TaskResult, Optional[NetworkDataModel]]: tuple containing the task result
             and an instance of NetworkDataModel or None if collection failed.
@@ -1695,6 +1746,23 @@ def collect_data(
         pensando_rdma_statistics: List[PensandoNicRdmaStatistics] = []
         pensando_version_host_software: Optional[PensandoNicVersionHostSoftware] = None
         pensando_version_firmware: List[PensandoNicVersionFirmware] = []
+        network_accessible: Optional[bool] = None
+
+        # Check network connectivity if URL is provided
+        if args and args.url:
+            cmd = args.netprobe if args.netprobe else "ping"
+            try:
+                network_accessible = self._check_network_connectivity(cmd, args.url)
+            except ValueError as e:
+                self._log_event(
+                    category=EventCategory.NETWORK,
+                    description=str(e),
+                    data={"netprobe": cmd, "url": args.url},
+                    priority=EventPriority.ERROR,
+                    console_log=True,
+                )
+                # Set network_accessible to None since we couldn't check
+                network_accessible = None
 
         # Collect interface/address information
         res_addr = self._run_sut_cmd(self.CMD_ADDR)
@@ -1823,6 +1891,7 @@ def collect_data(
             pensando_nic_rdma_statistics=pensando_rdma_statistics,
             pensando_nic_version_host_software=pensando_version_host_software,
             pensando_nic_version_firmware=pensando_version_firmware,
+            accessible=network_accessible,
         )
         self.result.status = ExecutionStatus.OK
         return self.result, network_data

nodescraper/plugins/inband/network/network_plugin.py

@@ -25,13 +25,16 @@
 ###############################################################################
 from nodescraper.base import InBandDataPlugin
 
+from .collector_args import NetworkCollectorArgs
 from .network_collector import NetworkCollector
 from .networkdata import NetworkDataModel
 
 
-class NetworkPlugin(InBandDataPlugin[NetworkDataModel, None, None]):
+class NetworkPlugin(InBandDataPlugin[NetworkDataModel, NetworkCollectorArgs, None]):
     """Plugin for collection of network configuration data"""
 
     DATA_MODEL = NetworkDataModel
 
     COLLECTOR = NetworkCollector
+
+    COLLECTOR_ARGS = NetworkCollectorArgs

nodescraper/plugins/inband/network/networkdata.py

@@ -317,3 +317,4 @@ class NetworkDataModel(DataModel):
     pensando_nic_rdma_statistics: List[PensandoNicRdmaStatistics] = Field(default_factory=list)
     pensando_nic_version_host_software: Optional[PensandoNicVersionHostSoftware] = None
     pensando_nic_version_firmware: List[PensandoNicVersionFirmware] = Field(default_factory=list)
+    accessible: Optional[bool] = None  # Network accessibility check via ping

test/functional/fixtures/network_plugin_config.json

@@ -2,6 +2,10 @@
   "global_args": {},
   "plugins": {
     "NetworkPlugin": {
+      "collector_args": {
+        "url": "mock.example.com",
+        "netprobe": "ping"
+      },
       "analysis_args": {}
     }
   },