fix(anta.tests): Correct inputs to outputs mapping for BGP tests

anta/cli/exec/utils.py

@@ -72,6 +72,9 @@ async def collect(dev: AntaDevice, command: str, outformat: Literal["json", "tex
         elif c.ofmt == "text":
             outfile = outdir / f"{safe_command}.log"
             content = c.text_output
+        else:
+            logger.error("Command outformat is not in ['json', 'text'] for command '%s'", command)
+            return
         with outfile.open(mode="w", encoding="UTF-8") as f:
             f.write(content)
         logger.info("Collected command '%s' from device %s (%s)", command, dev.name, dev.hw_model)

anta/tests/field_notices.py

@@ -103,6 +103,11 @@ def test(self) -> None:
         for component in command_output["details"]["components"]:
             if component["name"] == "Aboot":
                 aboot_version = component["version"].split("-")[2]
+                break
+        else:
+            self.result.is_failure("Aboot component not found")
+            return
+
         self.result.is_success()
         incorrect_aboot_version = (
             aboot_version.startswith("4.0.")

anta/tests/interfaces.py

@@ -14,6 +14,7 @@
 from pydantic import BaseModel, Field
 from pydantic_extra_types.mac_address import MacAddress
 
+from anta import GITHUB_SUGGESTION
 from anta.custom_types import EthernetInterface, Interface, Percent, PositiveInteger
 from anta.decorators import skip_on_platforms
 from anta.models import AntaCommand, AntaTemplate, AntaTest
@@ -702,6 +703,11 @@ def test(self) -> None:
             for interface in self.inputs.interfaces:
                 if interface.name == intf:
                     input_interface_detail = interface
+                    break
+            else:
+                self.result.is_error(f"Could not find `{intf}` in the input interfaces. {GITHUB_SUGGESTION}")
+                continue
+
             input_primary_ip = str(input_interface_detail.primary_ip)
             failed_messages = []
 

anta/tests/routing/bgp.py

@@ -256,20 +256,24 @@ def test(self) -> None:
 
         failures: dict[tuple[str, Any], dict[str, Any]] = {}
 
-        for command, af_input in zip(self.instance_commands, self.inputs.address_families):
+        for command in self.instance_commands:
             num_peers = None
             peer_count = 0
             command_output = command.json_output
 
-            afi = af_input.afi
-            safi = af_input.safi
-            afi_vrf = af_input.vrf
-            num_peers = af_input.num_peers
+            afi = command.params.afi
+            safi = command.params.safi if hasattr(command.params, "safi") else None
+            afi_vrf = command.params.vrf if hasattr(command.params, "vrf") else "default"
 
             # Swapping AFI and SAFI in case of SR-TE
             if afi == "sr-te":
                 afi, safi = safi, afi
 
+            for input_entry in self.inputs.address_families:
+                if input_entry.afi == afi and input_entry.safi == safi and input_entry.vrf == afi_vrf:
+                    num_peers = input_entry.num_peers
+                    break
+
             if not (vrfs := command_output.get("vrfs")):
                 _add_bgp_failures(failures=failures, afi=afi, safi=safi, vrf=afi_vrf, issue="Not Configured")
                 continue
@@ -396,8 +400,8 @@ def test(self) -> None:
             command_output = command.json_output
 
             afi = command.params.afi
-            safi = getattr(command.params, "safi", None)
-            afi_vrf = getattr(command.params, "vrf", "default")
+            safi = command.params.safi if hasattr(command.params, "safi") else None
+            afi_vrf = command.params.vrf if hasattr(command.params, "vrf") else "default"
 
             # Swapping AFI and SAFI in case of SR-TE
             if afi == "sr-te":
@@ -543,18 +547,22 @@ def test(self) -> None:
 
         failures: dict[tuple[str, Any], dict[str, Any]] = {}
 
-        for command, af_input in zip(self.instance_commands, self.inputs.address_families):
+        for command in self.instance_commands:
             command_output = command.json_output
 
-            afi = af_input.afi
-            safi = af_input.safi
-            afi_vrf = af_input.vrf
-            afi_peers = af_input.peers
+            afi = command.params.afi
+            safi = command.params.safi if hasattr(command.params, "safi") else None
+            afi_vrf = command.params.vrf if hasattr(command.params, "vrf") else "default"
 
             # Swapping AFI and SAFI in case of SR-TE
             if afi == "sr-te":
                 afi, safi = safi, afi
 
+            for input_entry in self.inputs.address_families:
+                if input_entry.afi == afi and input_entry.safi == safi and input_entry.vrf == afi_vrf:
+                    afi_peers = input_entry.peers
+                    break
+
             if not (vrfs := command_output.get("vrfs")):
                 _add_bgp_failures(failures=failures, afi=afi, safi=safi, vrf=afi_vrf, issue="Not Configured")
                 continue

anta/tests/stp.py

@@ -188,6 +188,7 @@ def test(self) -> None:
             if not (topologies := get_value(command.json_output, "topologies")):
                 not_configured.append(vlan_id)
             else:
+                interfaces_not_forwarding = []
                 for value in topologies.values():
                     if vlan_id and int(vlan_id) in value["vlans"]:
                         interfaces_not_forwarding = [interface for interface, state in value["interfaces"].items() if state["state"] != "forwarding"]

docs/scripts/generate_svg.py

@@ -72,6 +72,8 @@ def custom_progress_bar() -> None:
         print("Usage: python generate_svg.py anta <options>")
         sys.exit(1)
 
+    # possibly-used-before-assignment - prog / function_name -> not understanding sys.exit here...
+    # pylint: disable=E0606
     sys.argv = [prog, *args[1:]]
     module = import_module(module_path)
     function = getattr(module, function_name)

tests/units/anta_tests/routing/test_bgp.py

@@ -30,6 +30,7 @@
         "name": "success",
         "test": VerifyBGPPeerCount,
         "eos_data": [
+            # Need to order the output as the commands would be sorted after template rendering.
             {
                 "vrfs": {
                     "default": {
@@ -120,9 +121,10 @@
         ],
         "inputs": {
             "address_families": [
+                # evpn first to make sure that the correct mapping output to input is kept.
+                {"afi": "evpn", "num_peers": 2},
                 {"afi": "ipv4", "safi": "unicast", "vrf": "default", "num_peers": 2},
                 {"afi": "ipv4", "safi": "sr-te", "vrf": "MGMT", "num_peers": 1},
-                {"afi": "evpn", "num_peers": 2},
                 {"afi": "link-state", "num_peers": 2},
                 {"afi": "path-selection", "num_peers": 2},
             ]
@@ -652,9 +654,10 @@
         ],
         "inputs": {
             "address_families": [
+                # Path selection first to make sure input to output mapping is correct.
+                {"afi": "path-selection"},
                 {"afi": "ipv4", "safi": "unicast", "vrf": "default"},
                 {"afi": "ipv4", "safi": "sr-te", "vrf": "MGMT"},
-                {"afi": "path-selection"},
                 {"afi": "link-state"},
             ]
         },
@@ -1081,6 +1084,8 @@
         ],
         "inputs": {
             "address_families": [
+                # Path selection first to make sure input to output mapping is correct.
+                {"afi": "path-selection", "peers": ["10.1.255.20", "10.1.255.22"]},
                 {
                     "afi": "ipv4",
                     "safi": "unicast",
@@ -1093,7 +1098,6 @@
                     "vrf": "MGMT",
                     "peers": ["10.1.255.10", "10.1.255.12"],
                 },
-                {"afi": "path-selection", "peers": ["10.1.255.20", "10.1.255.22"]},
                 {"afi": "link-state", "peers": ["10.1.255.30", "10.1.255.32"]},
             ]
         },

tests/units/anta_tests/test_field_notices.py

@@ -128,6 +128,26 @@
             "messages": ["device is not impacted by FN044"],
         },
     },
+    {
+        "name": "failure-no-aboot-component",
+        "test": VerifyFieldNotice44Resolution,
+        "eos_data": [
+            {
+                "imageFormatVersion": "1.0",
+                "uptime": 1109144.35,
+                "modelName": "DCS-7280QRA-C36S",
+                "details": {
+                    "deviations": [],
+                    "components": [{"name": "NotAboot", "version": "Aboot-veos-4.0.1-3255441"}],
+                },
+            },
+        ],
+        "inputs": None,
+        "expected": {
+            "result": "failure",
+            "messages": ["Aboot component not found"],
+        },
+    },
     {
         "name": "success-JPE",
         "test": VerifyFieldNotice72Resolution,

tests/units/anta_tests/test_interfaces.py

@@ -1742,7 +1742,7 @@
                 },
             },
         ],
-        "inputs": {"mtu": 9214},
+        "inputs": {"mtu": 9214, "ignored_interfaces": ["Loopback", "Port-Channel", "Management", "Vxlan"], "specific_mtu": [{"Ethernet10": 9214}]},
         "expected": {"result": "success"},
     },
     {