Step 1 - Create Python File for pyATS AEtest Script

Create a dedicated test case file for validating BGP EVPN neighbor adjacencies on Border Gateway (BGW) switches using the pyATS AEtest framework.

touch ~/workspace/ndlab/nac/tests/aetest_bgw_rs_evpn_neighbor_test.py
code-server -r ~/workspace/ndlab/nac/tests/aetest_bgw_rs_evpn_neighbor_test.py

Step 2 - Add Imports

Import the required pyATS AEtest module, logging utilities, and JSON library needed for parsing BGP neighbor data retrieved from Nexus Dashboard.

from pyats import aetest
import logging
import json


logger = logging.getLogger(__name__)

Step 3 - Start Adding pyATS AETest Test Case with Setup

Define the BGP EVPN neighbor verification test case class and implement the setup method that will prepare the test environment before executing validation checks.

class VerifyBgwBgpL2vpnEvpnNeighbors(aetest.Testcase):
    '''Testcase with Steps

    This testcase demonstrates the usage of testcase steps. Note that steps
    applies to subsections in CommonSetup and CommonCleanup as well.
    '''
    @aetest.setup
    def setup(self):
        '''Testcase Setup

        '''

Step 4 - Extract Data for BGW Neighbors From Data Model

Extract route server IP addresses from the YAML data model for Multi-Site Domain (MSD) fabrics using route server DCI deployment, which defines the expected BGP EVPN neighbors.

        if self.parent.parameters['data']['vxlan']['fabric']['type'] == 'MSD':
            dci_deployment = self.parent.parameters['data']['vxlan']['multisite']['overlay_dci']['deployment_method']
            if 'Route_Server' in dci_deployment:
                route_servers = self.parent.parameters['data']['vxlan']['multisite']['overlay_dci']['route_server']['peers']
                self.route_servers = [route_server['ip_address'] for route_server in route_servers]
            elif 'Direct' in dci_deployment:
                pass

Step 5 - Parse BGW Role Types From ND Inventory

Identify and filter Border Gateway switches from the fabric inventory by their role types (border_gateway, border_gateway_spine), and calculate spine and leaf switch counts for each fabric site.

            bgw_roles = ["border_gateway", "border_gateway_spine"]
            inventory = self.parent.parameters['inventory'][self.parent.parameters['fabric']]
            self.bgw_switches = [
                {**switch, 'fabric': fabric_name, 'hostname': name}
                for fabric_name, fabric_switches in inventory.items()
                for name, switch in fabric_switches.items()
                if switch.get("role") in bgw_roles
            ]

            self.spine_counts = {
                fabric: sum(1 for switch in switches.values() if 'spine' in switch['role'])
                for fabric, switches in inventory.items()
            }

            self.leaf_counts = {
                fabric: sum(1 for switch in switches.values() if switch['role'] in ['leaf', 'border', 'border_gateway'])
                for fabric, switches in inventory.items()
            }

Step 6 - Add Start of pyATS AETest Test Case Test

Implement the test method that iterates through all BGW switches, executes the "show bgp l2vpn evpn neighbors" command via Nexus Dashboard REST API, and retrieves BGP neighbor data for validation.

    @aetest.test
    def test(self, testbed, steps):
        '''Steps must pass, unless otherwise indicated

        If step fails, by default, all remaining steps are not run.
        If an assertionError is caught, it is considered Failed() instead of
        Errored().
        '''
        command = "show bgp l2vpn evpn neighbors | json"
        api = "/appcenter/cisco/ndfc/api/v1/configtemplate/rest/config/templates/execute/show"

        for switch in self.bgw_switches:
            with steps.start(f"Verify BGP EVPN neighbors for BGW {switch['hostname']}", continue_=True) as step:

                # Execute show command via Nexus Dashboard API
                payload = {"cliCommandList": [command], "ipAddress": switch['ip']}
                response = testbed.devices[self.parent.parameters['fabric']].rest.post(api, json.dumps(payload))
                output = json.loads(response[0]['response'])

                # Handle cases where TABLE_neighbor or ROW_neighbor might not exist or be empty
                if 'TABLE_neighbor' not in output or 'ROW_neighbor' not in output['TABLE_neighbor']:
                    logger.error(f"No BGP EVPN neighbor data found on {switch['hostname']}")
                    step.failed(f"No BGP EVPN neighbor data found on {switch['hostname']}")
                    continue

Step 7 - Normalize Data From ND and Test State Against Data Model

Parse the BGP neighbor JSON response into a normalized list, filter for established neighbors matching the route server IPs from the data model, and validate that the actual neighbor count matches the expected count.

                # Normalize to list (could be single dict or list of dicts)
                neighbors = output['TABLE_neighbor']['ROW_neighbor']
                neighbors = neighbors if isinstance(neighbors, list) else [neighbors]

                # Get actual neighbors for switch that should match data model
                established_neighbors = [
                    neighbor
                    for neighbor in neighbors
                    if (neighbor['state'] == 'Established') and neighbor['neighbor'] in self.route_servers
                ]

                # Get established neighbors count that should match data model
                established_count = len(established_neighbors)

                # Get expected neighbors count for switch
                expected_count = len(self.route_servers)

                # Results
                if established_count == expected_count:
                    logger.info(f" ✓ Switch {switch['hostname']} has the expected {expected_count} BGP L2VPN EVPN neighbors.")
                else:
                    logger.error(f" ✗ Switch {switch['hostname']} expected {expected_count} BGP L2VPN EVPN neighbors, but had {established_count} neighbors.")
                    step.failed()

Continue to the next section to create your second test case.