初めに

pyATS/GenieはCisco DevNetで公開されているテスト自動化ツールです。pyATSがテストフレームワークを提供し、GenieがParsers、Models、Apis、Harnessなどのネットワーク機器に対するテスト機能を提供します。ネットワークテストの自動化に特化しており、かなり使えます。
　今回は、Keysight(Ixia) IxNetworkでのプロトコル試験、トラフィック試験を自動化するRestPy、マルチベンダー環境のネットワーク設定を抽象化しAPIを公開するNSOと連携させたテストスクリプトを作ってみました。

テストシナリオ

pyATS AEtest でテストスクリプトを作成します。 Keysight(Ixia) IxNetwork VE でトラフィックを流しつつ、 v-dist-rtr01 の GigabitEthernet0/0/0/0 を shutdown、し、トラフィックのPacket Loss Duration (ms)が100ms以下、Store-Forward Max Latency (ns)が20000000ns以下ならPassというシナリオです。
トラフィックテストが完了すればGigabitEthernet0/0/0/0をno shutdownし、Genie ops で取得したテスト前後のスナップショットを比較し差異がないことを確認します。

Keysight(Ixia) IxNetwork VE はトラフィックジェネレータの仮装版で、プロトコルやトラフィックを作成するAPI serverである IxNetwork VE と、トラフィックを送受信するポートを持つ仮装シャーシである Virtual Test Appliance のセットで使用します。

NSO を使用した環境では各デバイスのコンフィグはNSOのシンクロしています。設定を変更する場合シンクロが外れないように NSO 経由で変更するようにしています。

ネットワーク構成

CML2のIOS XRv 9000で構築したSR-MPLSネットワークを利用しています。
各デバイスのマネージメントインターフェースをexternal connector0に接続し、外部のpyATSをインストールしたDockerコンテナからアクセスできるようにしています。
IxNetworkで作成したトラフィックを送信受信するために、Virtual Test Applianceのポートをexternal connector1、external connector2に割り当てたVLANにそれぞれ接続します。

testbed.yamlファイル

connectionsにはtelnet、ssh、netconfを定義していますが、今回pyATSからはtelnetで接続します。
NSO環境下では各デバイスのコンフィグはNSOと同期されているので、argumentsにinit_exec_commands: []、init_config_commands: []を指定してデバイス接続時に設定を追加しないようにしています。

testbed.yaml

testbed.yaml

testbed:
  name: 'Nail Network'
  credentials:
    default:
      username: admin
      password: admin
devices:
  v-core-rtr01:
    os: iosxr
    platform: iosxrv9k
    alias: uut
    type: router
    connections:
      defaults:
        via: cli
      cli:
        ip: 172.16.1.73
        protocol: telnet
        port: 23
        arguments:
          init_exec_commands: []
          init_config_commands: []
      ssh:
        ip: 172.16.1.73
        protocol: ssh
        arguments:
          init_exec_commands: []
          init_config_commands: []
      netconf:
        ip: 172.16.1.73
        port: 830
        class: yang.connector.Netconf
        credentials:
          netconf:
            username: admin
            password: admin
    credentials:
      default:
        username: admin
        password: admin
      enable:
        password: admin
  v-core-rtr02:
    os: iosxr
    platform: iosxrv9k
    alias: xr02
    type: router
    connections:
      defaults:
        via: cli
      cli:
        ip: 172.16.1.74
        protocol: telnet
        port: 23
        arguments:
          init_exec_commands: []
          init_config_commands: []
      ssh:
        ip: 172.16.1.74
        protocol: ssh
        arguments:
          init_exec_commands: []
          init_config_commands: []
      netconf:
        ip: 172.16.1.74
        port: 830
        class: yang.connector.Netconf
        credentials:
          netconf:
            username: admin
            password: admin
    credentials:
      default:
        username: admin
        password: admin
      enable:
        password: admin
  v-dist-rtr01:
    os: iosxr
    platform: iosxrv9k
    alias: xr03
    type: router
    connections:
      defaults:
        via: cli
      cli:
        ip: 172.16.1.75
        protocol: telnet
        port: 23
        arguments:
          init_exec_commands: []
          init_config_commands: []
      ssh:
        ip: 172.16.1.75
        protocol: ssh
        arguments:
          init_exec_commands: []
          init_config_commands: []
      netconf:
        ip: 172.16.1.75
        port: 830
        class: yang.connector.Netconf
        credentials:
          netconf:
            username: admin
            password: admin
    credentials:
      default:
        username: admin
        password: admin
      enable:
        password: admin
  v-dist-rtr02:
    os: iosxr
    platform: iosxrv9k
    alias: xr04
    type: router
    connections:
      defaults:
        via: cli
      cli:
        ip: 172.16.1.76
        protocol: telnet
        port: 23
        arguments:
          init_exec_commands: []
          init_config_commands: []
      ssh:
        ip: 172.16.1.76
        protocol: ssh
        arguments:
          init_exec_commands: []
          init_config_commands: []
      netconf:
        ip: 172.16.1.76
        port: 830
        class: yang.connector.Netconf
        credentials:
          netconf:
            username: admin
            password: admin
    credentials:
      default:
        username: admin
        password: admin
      enable:
        password: admin
  v-edge-rtr01:
    os: iosxr
    platform: iosxrv9k
    alias: xr05
    type: router
    connections:
      defaults:
        via: cli
      cli:
        ip: 172.16.1.77
        protocol: telnet
        port: 23
        arguments:
          init_exec_commands: []
          init_config_commands: []
      ssh:
        ip: 172.16.1.77
        protocol: ssh
        arguments:
          init_exec_commands: []
          init_config_commands: []
      netconf:
        ip: 172.16.1.77
        port: 830
        class: yang.connector.Netconf
        credentials:
          netconf:
            username: admin
            password: admin
    credentials:
      default:
        username: admin
        password: admin
      enable:
        password: admin
  v-edge-rtr02:
    os: iosxe
    platform: csr1000v
    alias: xe01
    type: router
    connections:
      defaults:
        via: cli
      cli:
        ip: 172.16.1.78
        protocol: telnet
        arguments:
          init_exec_commands: []
          init_config_commands: []
        port: 23
      ssh:
        ip: 172.16.1.78
        protocol: ssh
        arguments:
          init_exec_commands: []
          init_config_commands: []
      netconf:
        ip: 172.16.1.78
        port: 830
        class: yang.connector.Netconf
        credentials:
          netconf:
            username: admin
            password: admin
    credentials:
      default:
        username: admin
        password: admin
      enable:
        password: admin
  v-edge-sw01:
    os: nxos
    platform: n9k
    alias: nx01
    type: nxos
    connections:
      defaults:
        via: cli
      cli:
        ip: 172.16.1.79
        protocol: telnet
        port: 23
        arguments:
          init_exec_commands: []
          init_config_commands: []
      ssh:
        ip: 172.16.1.79
        protocol: ssh
        arguments:
          init_exec_commands: []
          init_config_commands: []
      netconf:
        ip: 172.16.1.79
        port: 830
        class: yang.connector.Netconf
        credentials:
          netconf:
            username: admin
            password: admin
    credentials:
      default:
        username: admin
        password: admin
      enable:
        password: admin
  v-edge-sw02:
    os: nxos
    platform: n9k
    alias: nx02
    type: nxos
    connections:
      defaults:
        via: cli
      cli:
        ip: 172.16.1.80
        protocol: telnet
        port: 23
        arguments:
          init_exec_commands: []
          init_config_commands: []
      ssh:
        ip: 172.16.1.80
        protocol: ssh
        arguments:
          init_exec_commands: []
          init_config_commands: []
      netconf:
        ip: 172.16.1.80
        port: 830
        class: yang.connector.Netconf
        credentials:
          netconf:
            username: admin
            password: admin
  v-user-sw01:
    os: nxos
    platform: n9k
    alias: nx03
    type: nxos
    connections:
      defaults:
        via: cli
      cli:
        ip: 172.16.1.81
        protocol: telnet
        port: 23
        arguments:
          init_exec_commands: []
          init_config_commands: []
      ssh:
        ip: 172.16.1.81
        protocol: ssh
        arguments:
          init_exec_commands: []
          init_config_commands: []
      netconf:
        ip: 172.16.1.81
        port: 830
        class: yang.connector.Netconf
        credentials:
          netconf:
            username: admin
            password: admin
    credentials:
      default:
        username: admin
        password: admin
      enable:
        password: admin
  vswitch:
    os: nxos
    platform: n9k
    alias: nx04
    type: nxos
    connections:
      defaults:
        via: cli
      cli:
        ip: 172.16.1.82
        protocol: telnet
        port: 23
        arguments:
          init_exec_commands: []
          init_config_commands: []
      ssh:
        ip: 172.16.1.82
        protocol: ssh
        arguments:
          init_exec_commands: []
          init_config_commands: []
      netconf:
        ip: 172.16.1.82
        port: 830
        class: yang.connector.Netconf
        credentials:
          netconf:
            username: admin
            password: admin
    credentials:
      default:
        username: admin
        password: admin
      enable:
        password: admin
topology:
  v-core-rtr01:
    interfaces:
      Loopback0:
        type: loopback
      MgmtEth0/0/CPU0/0:
        link: mgmt_uut
        type: ethernet
      GigabitEthernet0/0/0/0:
        link: uut_xr02
        type: ethernet
      GigabitEthernet0/0/0/1:
        link: uut_xr03
        type: ethernet
      GigabitEthernet0/0/0/2:
        link: uut_xr04
        type: ethernet
      GigabitEthernet0/0/0/3:
        link: uut_xr05
        type: ethernet
      GigabitEthernet0/0/0/4:
        link: uut_xe01
        type: ethernet
  v-core-rtr02:
    interfaces:
      Loopback0:
        type: loopback
      MgmtEth0/0/CPU0/0:
        link: mgmt_xr02
        type: ethernet
      GigabitEthernet0/0/0/0:
        link: uut_xr02
        type: ethernet
      GigabitEthernet0/0/0/1:
        link: xr02_xr03
        type: ethernet
      GigabitEthernet0/0/0/2:
        link: xr02_xr04
        type: ethernet
      GigabitEthernet0/0/0/3:
        link: xr02_xr05
        type: ethernet
      GigabitEthernet0/0/0/4:
        link: xr02_xe01
        type: ethernet
  v-dist-rtr01:
    interfaces:
      Loopback0:
        type: loopback
      MgmtEth0/0/CPU0/0:
        link: mgmt_xr03
        type: ethernet
      GigabitEthernet0/0/0/0:
        link: uut_xr03
        type: ethernet
      GigabitEthernet0/0/0/1:
        link: xr02_xr03
        type: ethernet
      GigabitEthernet0/0/0/2:
        link: xr03_xr04
        type: ethernet
      GigabitEthernet0/0/0/3:
        link: xr03_nx01
        type: ethernet
  v-dist-rtr02:
    interfaces:
      Loopback0:
        type: loopback
      MgmtEth0/0/CPU0/0:
        link: mgmt_xr04
        type: ethernet
      GigabitEthernet0/0/0/0:
        link: uut_xr04
        type: ethernet
      GigabitEthernet0/0/0/1:
        link: xr02_xr04
        type: ethernet
      GigabitEthernet0/0/0/2:
        link: xr03_xr04
        type: ethernet
      GigabitEthernet0/0/0/3:
        link: xr04_nx02
        type: ethernet
  v-edge-rtr01:
    interfaces:
      Loopback0:
        type: loopback
      MgmtEth0/0/CPU0/0:
        link: mgmt_xr05
        type: ethernet
      GigabitEthernet0/0/0/0:
        link: uut_xr05
        type: ethernet
      GigabitEthernet0/0/0/1:
        link: xr02_xr05
        type: ethernet
      GigabitEthernet0/0/0/2:
        link: xr05_nx03
        type: ethernet
  v-edge-rtr02:
    interfaces:
      Loopback0:
        type: loopback
      GigabitEthernet4:
        link: mgmt_xe01
        type: ethernet
      GigabitEthernet1:
        link: uut_xe01
        type: ethernet
      GigabitEthernet2:
        link: xr02_xe01
        type: ethernet
  v-edge-sw01:
    interfaces:
      mgmt0:
        link: mgmt_nx01
        type: ethernet
      Ethernet1/1:
        link: xr03_nx01
        type: ethernet
      Ethernet1/2:
        link: nx01a_nx02a
        type: ethernet
      Ethernet1/3:
        link: nx01b_nx02b
        type: ethernet
      Ethernet1/4:
        link: nx01_nx04
        type: ethernet
  v-edge-sw02:
    interfaces:
      mgmt0:
        link: mgmt_nx02
        type: ethernet
      Ethernet1/1:
        link: xr04_nx02
        type: ethernet
      Ethernet1/2:
        link: nx01a_nx02a
        type: ethernet
      Ethernet1/3:
        link: nx01b_nx02b
        type: ethernet
      Ethernet1/4:
        link: nx02_nx04
        type: ethernet
  v-user-sw01:
    interfaces:
      mgmt0:
        link: mgmt_nx03
        type: ethernet
      Ethernet1/1:
        link: xr05_nx03
        type: ethernet
      Ethernet1/2:
        link: access_to_user
        type: ethernet
  vswitch:
    interfaces:
      mgmt0:
        link: mgmt_nx03
        type: ethernet
      Ethernet1/1:
        link: nx01_nx04
        type: ethernet
      Ethernet1/2:
        link: nx02_nx04
        type: ethernet
      Ethernet1/3:
        link: access_to_app
        type: ethernet

RestPyとNSOのコード

pyATSのテストスクリプトを作成する前に、IxNetwork VEでトラフィックを作成するためのRestPyのコード、ルータのインターフェースをshut 、no shutするNSOのコードを準備します。

Ixia RestPyコード

RestPyはKEYSIGHT IxNetworkのオープソース自動化ライブラリです。チュートリアルやサンプルスクリプトが公開されています。

-KEYSIGHT openixia

テストスクリプトで使用するRestPyを利用したスクリプトを作成しました。

(参考)ixia_functions.py

ixia_functions.py

import json, sys, os, traceback
from ixnetwork_restpy import SessionAssistant, Files


class IxiaFunctions():
    """
    Class for IxNetwork actions.
    How to call:
        1. Create a object with below.
           - IxNetwork api server's address 
           - Ixia Chassis list
           - Ixia portlist of Chassis
           - Username of IxNetwork api server
           - Password of IxNetwork api server
           - Debug mode
           - Json config file of IxNetwork
        2. Execute any function.
    """

    def __init__(self):
        """
        Parameters
        ----------
        ixnetwork_addr : str
            IxNetwork api server's address.
        chassis_list : list
            Ixia Chassis list.
        portlist : list
            Ixia portlist of Chassis.
        username : str
            Username of IxNetwork api server.
        password : str
            Password of IxNetwork api server.
        debug_mode : bool
            For linux and windowsConnectionMgr only. 
            Set to True to leave the session alive for debugging.
        force_take_port_ownership : bool
            Forcefully take port ownership if the portList are owned by other users.
        json_config_file : str
            Config file of IxNetwork.
        """
        self.ixnetwork_addr = '{ "your ixnetwork address" }'
        self.chassis_list = ['{ "your ixis chassis address" }']
        self.portlist = [[self.chassis_list[0], 1, 1], [self.chassis_list[0], 1, 2]]
        self.username = '{ "your ixnetwork username" }'
        self.password = '{ "your ixnetwork password" }'
        self.debug_mode = False
        self.force_take_port_ownership = True
        self.json_config_file = '{ "your ixnetwork config file" }'
        try:
            self.session = SessionAssistant(
                IpAddress=self.ixnetwork_addr,
                RestPort=None,
                UserName=self.username,
                Password=self.password,
                SessionName=None,
                SessionId=None,
                ApiKey=None,
                ClearConfig=True,
                LogLevel='all',
                LogFilename='restpy.log'
            )
            self.ixNetwork = self.session.Ixnetwork
        except Exception as errMsg:
            print('\n%s' % traceback.format_exc())
            if self.debug_mode == False and 'session' in dir(self):
                self.session.Session.remove()

    def start_traffic(self):
        try:

            # Loads a saved .json config file
            self.ixNetwork.info(
                f'\nLoading JSON config file: {self.json_config_file}'
            )
            self.ixNetwork.ResourceManager.ImportConfigFile(
                Files(self.json_config_file, local_file=True), Arg3=True
            )

            # Assign ports.
            portmap = self.session.PortMapAssistant()
            vport = dict()
            for index, port in enumerate(self.portlist):
                portname = self.ixNetwork.Vport.find()[index].Name
                portmap.Map(
                    IpAddress=port[0],
                    CardId=port[1],
                    PortId=port[2],
                    Name=portname
                )
            portmap.Connect(self.force_take_port_ownership)

            # Start all protocols
            self.ixNetwork.StartAllProtocols(Arg1='sync')
            self.ixNetwork.info('Verify protocol sessions\n')

            # Verify all protocols
            protocolSummary = self.session.StatViewAssistant(
                'Protocols Summary'
            )
            protocolSummary.CheckCondition(
                'Sessions Not Started',
                protocolSummary.EQUAL,
                0
            )
            protocolSummary.CheckCondition(
                'Sessions Down',
                protocolSummary.EQUAL,
                0
            )
            self.ixNetwork.info(protocolSummary)

            # Get the Traffic Item name for getting Traffic Item statistics.
            trafficItem = self.ixNetwork.Traffic.TrafficItem.find()[0]

            # Start traffic
            trafficItem.Generate()
            self.ixNetwork.Traffic.Apply()
            self.ixNetwork.Traffic.StartStatelessTrafficBlocking()

        except Exception as errMsg:
            print('\n%s' % traceback.format_exc())
            if self.debug_mode == False and 'session' in dir(self):
                self.session.Session.remove()

    def view_statistics(self):
        try:
            trafficItemStatistics = self.session.StatViewAssistant('Traffic Item Statistics')
            self.ixNetwork.info(f'{trafficItemStatistics}\n')
            txFrames = trafficItemStatistics.Rows['Tx Frames']
            rxFrames = trafficItemStatistics.Rows['Rx Frames']
            duration = trafficItemStatistics.Rows['Packet Loss Duration (ms)']
            latency = trafficItemStatistics.Rows['Store-Forward Avg Latency (ns)']
            self.ixNetwork.info(f'\nTraffic Item Stats:\n\tTxFrames: {txFrames}  RxFrames: {rxFrames}  Duration: {duration}  Latency(ns): {latency} \n')
            output = {
                "txFrames": txFrames,
                "rxFrames": rxFrames,
                "duration": duration,
                "latency": latency
            }
            return output

        except Exception as errMsg:
            print('\n%s' % traceback.format_exc())
            if self.debug_mode == False and 'session' in dir(self):
                self.session.Session.remove()
            return False

    def stop_traffic(self):
        self.ixNetwork.Traffic.StopStatelessTrafficBlocking()
        if self.debug_mode == False:
            # For Linux and connection_manager only
                self.session.Session.remove()

if __name__ == "__main__":
    import time
    ixia_object = IxiaFunctions()
    ixia_object.start_traffic()
    time.sleep(60)
    ixia_object.view_statistics()
    ixia_object.stop_traffic()

__init__()
変数を定義し、IxNetworkのセッションを作成。
start_traffic()
トラフィックを定義済みのIxNetworkのコンフィグファイルをロード、スタート。
view_statistics()
Traffic Statisticsを抽出、構造化された結果をリターン。
stop_traffic()
トラフィックをストップ、セッションを削除。

NSOコード

NSOはサービスは作るのではなく、Restconfからcli nedでルータのインターフェースをshut、no shutするようにしています。
インターフェースのshut、no shutならGenieのApisを使用することもできますが、NSO環境下では各デバイスのコンフィグはNSOと同期されているので、NSO経由で実施しています。

(参考)nso_actions.py

nso_actions.py

import requests
import json
import sys
import logging


class NsoActions():
    """
    Class for nso actions.
    How to call:
        1. Create a object with the NSO address username and password.
        2. Execute any function.
    """

    def __init__(self):
        """
        Parameters
        ----------
        host : str
            NSO address or FQDN
        auth : Tuple
            NSO username and password
        """
        self.host = " {your_nso_address} "
        self.auth = (" {your_nso_username} ", " {your_nso_password} ")
        self.headers = {
            'Accept': 'application/yang-data+json',
            'Content-Type': 'application/yang-data+json',
        }

    def get_xr_intf_status(self, dev, intf_name, intf_num):
        """
        Parameters
        ----------
        dev : str
            device name
        intf_name : str
            interface name you want get status
        intf_num
            interface number you want get status

        Returns
        -------
        nothing
        """
        intf_num_url = intf_num.replace("/", "%2F")
        url = f"http://{self.host}/restconf/data/tailf-ncs:devices/device={dev}/config/tailf-ned-cisco-ios-xr:interface/{intf_name}={intf_num_url}/shutdown?content=config"
        payload = ""
        response = requests.request(
            "GET", url, headers=self.headers, auth=self.auth, data=payload)
        if response.status_code == 200:
            return False
        else:
            return True

    def shut_xr_intf(self, dev, intf_name, intf_num):
        """
        Parameters
        ----------
        dev : str
            device name
        intf_name : str
            interface name you want get status
        intf_num
            interface number you want get status

        Returns
        -------
        nothing
        """
        intf_num_url = intf_num.replace("/", "%2F")
        url = f"http://{self.host}/restconf/data/tailf-ncs:devices/device={dev}/config/tailf-ned-cisco-ios-xr:interface/{intf_name}={intf_num_url}/shutdown?content=config"
        payload = '{"tailf-ned-cisco-ios-xr:shutdown": [null]}'
        response = requests.request(
            "put", url, headers=self.headers, auth=self.auth, data=payload)
        if response.status_code == 201:
            message = f"Success shut interface {intf_name} {intf_num} of {dev}"
            logging.info(message)
        else:
            message = f"Failed shut interface {intf_name} {intf_num} of {dev}"
            logging.info(message)

    def noshut_xr_intf(self, dev, intf_name, intf_num):
        """
        Parameters
        ----------
        dev : str
            device name
        intf_name : str
            interface name you want get status
        intf_num
            interface number you want get status

        Returns
        -------
        nothing
        """
        intf_num_url = intf_num.replace("/", "%2F")
        url = f"http://{self.host}/restconf/data/tailf-ncs:devices/device={dev}/config/tailf-ned-cisco-ios-xr:interface/{intf_name}={intf_num_url}/shutdown?content=config"
        payload = ""
        response = requests.request(
            "delete", url, headers=self.headers, auth=self.auth, data=payload)
        if response.status_code == 204:
            message = f"Success no shut interface {intf_name} {intf_num} of {dev}"
            logging.info(message)
        else:
            message = f"Failed no shut interface {intf_name} {intf_num} of {dev}"
            logging.info(message)

AEtest

pyATSのメインコンポーネントであるAEtestでテストスクリプトを作成していきます。AEtestのテストスクリプトは CommonSetup 、Testcase 、CommonCleanup の3つのコンテナで構成されます。
　また、コンテナの中もサブセクションに分割することができ、CommonSetup 、 CommonCleanup は subsection 、Testcase はsetup、test、cleanupに分割することができ、具体的なアクションを定義していくことができます。
　さらに、stepsを使えばtestセクションをさらに分割して複数テストの評価をすることができます。

テストスクリプトの実行

AEtestのテストスクリプトを実行する方法は大きく2つあります。 Standalone での実行と、 EasypyJobfiles での実行です。
今回はどちらでも実行できるようにテストスクリプトを作成していきます。

Standalone
pyatsインスタンス内で、Linuxコマンドラインから直接テストスクリプトを実行します。これにより、テストスクリプトを独立して実行でき、すべてのログ出力はデフォルトでスクリーンプリントのみになります。
ログアーカイブなどを作成することなく、迅速で軽量なスクリプト開発と反復サイクルに最適です。

EasypyJobfiles
EasypyJobfilesを通してテストスクリプトをタスクとして実行します。この方法では、Easypy-Runtime Environmentを使用する必要があり、ログとアーカイブを生成します。
標準環境、レポート、ログアーカイブが必要な場合、および事後デバッグが必要な場合の実行に最適です。

AEtestテストスクリプト

まずは、必要なモジュールをインポートし、loggerを作成します。

from pyats import aetest
from unicon.core.errors import TimeoutError, StateMachineError, ConnectionError
import os
import time
import logging
import nso_actions
import ixia_functions

logger = logging.getLogger(__name__)

CommonSetup

class CommonSetup(aetest.CommonSetup):

CommonSetup にはサブセクションを5つ作成します。

①subsection testbed.yamlの確認

テストスクリプトにtestbed.yamlが提供されているかを確認します。

@aetest.subsection
def assert_testbed_common(self, testbed):
    """
    Assering testbed
    """
    logger.info(f"{os.path.basename(__file__)} test start")
    assert testbed, "Hey where is Testbed ?"

②subsection parent.parametersのアップデート

testbedオブジェクトからdevicesオブジェクトを抜き出し、リスト化してparent.parametersに追加します。そうすることによって、TestCaseTemplate　、CommonCleanup　からでもdevicesオブジェクトを参照することができます。

@aetest.subsection
def setup_device_objects_common(self, testbed):
    """
    Creating testbed device object list
    """
    self.testbed_dev_list = [
        dev for dev in testbed.devices.keys()
    ]
    self.devobj_list = [
        testbed.devices[devobj] for devobj in self.testbed_dev_list
    ]
    self.parent.parameters.update(
        devobj_list=self.devobj_list
    )

③subsection テスト対象デバイスのインターフェース状態のチェック

テスト対象のデバイス名、インターフェース名、インターフェース番号を定義し、それらをparent.parametersに渡すと共に、インターフェースが no shut であるかをチェックします。

@aetest.subsection
def check_interface_status(self):
    """
    Checking target interface status
    """
    self.dev = "v-dist-rtr01"
    self.intf_name = "GigabitEthernet"
    self.intf_num = "0/0/0/0"
    self.parent.parameters.update(
        dev=self.dev
    )
    self.parent.parameters.update(
        intf_name=self.intf_name
    )
    self.parent.parameters.update(
        intf_num=self.intf_num
    )
    assert nso_actions.NsoActions().get_xr_intf_status(
        self.dev, self.intf_name, self.intf_num
    ), f"Hey interface {self.intf_name} {self.intf_name} of {self.dev} is shutdowned !!"

④subsection 全てのデバイスへの接続チェック

testbed.yamlの全てのデバイスへ接続できるかどうかをチェックします。

@aetest.subsection
def connect_common(self):
    """
    Connecting testbed all devices
    """

    for devobj in self.devobj_list:
        try:
            devobj.connect()
        except (TimeoutError, StateMachineError, ConnectionError):
            logger.info("Unable to connect to devices")

⑤subsection テスト対象のデバイスの事前スナップショットの取得

テスト前後で各フィーチャーの状態に差異がないことを確認するために、Genie ops で各フィーチャーのスナップショットをテスト前後で取得します。
前後で取得したスナップショットを比較するとき、カウンターのように時間と共に変動があるものなどについては比較の対象外にしたいので、diff_ignore.appendで対象外にします。
コードにもあるように、対象外にしたい値を書き換えることで対象外にすることもできます。

@aetest.subsection
def snapshot_before(self):
    """
    Gathering features snapshot before testing
    """
    self.snapshot_b = {}
    for devobj in self.devobj_list:
        if devobj.name == self.dev:
            # Feature bgp
            bgpb = devobj.learn("bgp")
            bgpb.diff_ignore.append(
                '[info][instance][default][vrf][default][neighbor][(.*)][up_time]'
            )
            bgpb.diff_ignore.append(
                '[info][instance][default][vrf][default][neighbor][(.*)][bgp_neighbor_counters][messages][sent][totals]'
            )
            bgpb.diff_ignore.append(
                '[info][instance][default][vrf][default][neighbor][(.*)][bgp_neighbor_counters][messages][sent][keepalives]'
            )
            bgpb.diff_ignore.append(
                '[info][instance][default][vrf][default][neighbor][(.*)][bgp_neighbor_counters][messages][received][totals]'
            )
            bgpb.diff_ignore.append(
                '[info][instance][default][vrf][default][neighbor][(.*)][bgp_neighbor_counters][messages][received][keepalives]'
            )
            for k in bgpb.routes_per_peer["instance"]["default"]["vrf"]["default"]["neighbor"].keys():
                try:
                    bgpb.routes_per_peer["instance"]["default"]["vrf"]["default"]["neighbor"][
                        k]["address_family"]["vpnv4 unicast"]["msg_rcvd"] = "diff_ignore"
                except:
                    pass
            for k in bgpb.routes_per_peer["instance"]["default"]["vrf"]["default"]["neighbor"].keys():
                try:
                    bgpb.routes_per_peer["instance"]["default"]["vrf"]["default"]["neighbor"][
                        k]["address_family"]["vpnv4 unicast"]["msg_sent"] = "diff_ignore"
                except:
                    pass
            for k in bgpb.routes_per_peer["instance"]["default"]["vrf"]["default"]["neighbor"].keys():
                try:
                    bgpb.routes_per_peer["instance"]["default"]["vrf"]["default"]["neighbor"][
                        k]["address_family"]["vpnv4 unicast"]["up_down"] = "diff_ignore"
                except:
                    pass
            self.snapshot_b[devobj.name, "bgp"] = bgpb
            # Feature isis
            isisb = devobj.learn("isis")
            isisb.diff_ignore.append(
                '[info][instance][100][vrf][default][interfaces][(.*)][adjacencies][(.*)][neighbor_snpa][(.*)][level][level-2][hold_timer]'
            )
            isisb.diff_ignore.append(
                '[info][instance][100][vrf][default][interfaces][(.*)][adjacencies][(.*)][neighbor_snpa][(.*)][level][level-2][lastuptime]'
            )
            isisb.diff_ignore.append(
                '[info][instance][100][vrf][default][lsp_log]'
            )
            isisb.diff_ignore.append(
                '[info][instance][100][vrf][default][spf_log]'
            )
            isisb.diff_ignore.append(
                '[info][instance][100][vrf][default][interfaces][(.*)][packet_counters]'
            )
            for k in isisb.lsdb["instance"]["100"]["vrf"]["default"]["level_db"][2].keys():
                try:
                    isisb.lsdb["instance"]["100"]["vrf"]["default"]["level_db"][2][k]["checksum"] = "diff_ignore"
                except:
                    pass
            for k in isisb.lsdb["instance"]["100"]["vrf"]["default"]["level_db"][2].keys():
                try:
                    isisb.lsdb["instance"]["100"]["vrf"]["default"]["level_db"][2][k]["sequence"] = "diff_ignore"
                except:
                    pass
            for k in isisb.lsdb["instance"]["100"]["vrf"]["default"]["level_db"][2].keys():
                try:
                    isisb.lsdb["instance"]["100"]["vrf"]["default"]["level_db"][2][k]["remaining_lifetime"] = "diff_ignore"
                except:
                    pass
            self.snapshot_b[devobj.name, "isis"] = isisb
            # Feature routing
            routingb = devobj.learn("routing")
            for kvrf in routingb.info["vrf"].keys():
                try:
                    for kroutes in routingb.info["vrf"][kvrf]["address_family"]["ipv4"]["routes"].keys():
                        try:
                            for index in routingb.info["vrf"][kvrf]["address_family"]["ipv4"]["routes"][kroutes]["next_hop"]["next_hop_list"].keys():
                                try:
                                    routingb.info["vrf"][kvrf]["address_family"]["ipv4"]["routes"][
                                        kroutes]["next_hop"]["next_hop_list"][index]["updated"] = "diff_ignore"
                                except:
                                    pass
                        except:
                            pass
                except:
                    pass

            for kvrf in routingb.info["vrf"].keys():
                try:
                    for kroutes in routingb.info["vrf"][kvrf]["address_family"]["ipv6"]["routes"].keys():
                        try:
                            for index in routingb.info["vrf"][kvrf]["address_family"]["ipv6"]["routes"][kroutes]["next_hop"]["next_hop_list"].keys():
                                try:
                                    routingb.info["vrf"][kvrf]["address_family"]["ipv6"]["routes"][
                                        kroutes]["next_hop"]["next_hop_list"][index]["updated"] = "diff_ignore"
                                except:
                                    pass
                        except:
                            pass
                except:
                    pass
            self.snapshot_b[devobj.name, "routing"] = routingb
            # Feature interface
            intb = devobj.learn("interface")
            intb.diff_ignore.append(
                '[info][(.*)][accounting]'
            )
            intb.diff_ignore.append(
                '[info][(.*)][counters]'
            )
            self.snapshot_b[devobj.name, "interface"] = intb
            # Feature arp
            arpb = devobj.learn("arp")
            arpb.diff_ignore.append(
                '[info][statistics]'
            )
            self.snapshot_b[devobj.name, "arp"] = arpb
            # Feature nd
            ndb = devobj.learn("nd")
            ndb.diff_ignore.append(
                '[info][interfaces][(.*)][neighbors][(.*)][age]'
            )
            self.snapshot_b[devobj.name, "nd"] = ndb
        else:
            pass
    self.parent.parameters.update(
        snapshot_b=self.snapshot_b
    )

Testcase

class TestCaseTrafficStatisticsCheck(aetest.Testcase):

Testcase にはsetupを1つtestを3つ、cleanupを1つ作成します。
なお、setup、cleanupは1つずつしか作成することができません。

setup IxNetworkトラフィックスタート

setupセクションで定義済みのIxNetworkトラフィックをスタートします。

@aetest.setup
def setup(self):
    """
    Start ixia traffic
    """
    self.ixia = ixia_functions.IxiaFunctions()
    self.ixia.start_traffic()

test① ルータのインターフェース`shutdown`

testセクションで v-dist-rtr01 の GigabitEthernet 0/0/0/0 をNS0からshutdownし、インターフェースの状態をチェックします。

@aetest.test
def shutdown_interface(self, dev, intf_name, intf_num):
    """
    Shutdown interface
    """
    time.sleep(120)
    nso_actions.NsoActions().shut_xr_intf(
        dev, intf_name, intf_num
    )
    time.sleep(60)
    if not nso_actions.NsoActions().get_xr_intf_status(
        dev, intf_name, intf_num
    ):
        self.passed(
            f'Interface {intf_name} {intf_num} is shutdowned at {dev} !!'
        )
    else:
        self.failed(
            f'Interface {intf_name} {intf_num} is not shutdowned at {dev} !!'
        )

test② IxNetworkトラフィック測定結果の判定

testセクションでインターフェースをshutした後にIxNetworkから Traffic Item Statistics を取得し、Packet Loss Duration (ms)とStore-Forward Avg Latency (ns)が閾値以下であるかを判定します。
testセクションを steps で分割することで、一つのtestセクションで複数の判定を行うことができます。

@aetest.test
def traffic_statistics(self, steps):
    """
    Viewing ixia traffic statistics
    """
    statistics = self.ixia.view_statistics()
    if not statistics:
        self.errored('IXIA statistics is nothing !!')
    else:
        with steps.start(
            "Asserting duration",
            description='Asserting duration',
            continue_=True
        ) as duration_step:
            if float(statistics["duration"]) < 100:
                duration_step.passed(
                    f'Packet Loss Duration (ms) is good. {statistics["duration"]}'
                )
            else:
                duration_step.failed(
                    f'Packet Loss Duration (ms) is too long ! {statistics["duration"]}'
                )
        with steps.start(
            "Asserting latency",
            description='Asserting latency',
            continue_=True
        ) as latency_step:
            if float(statistics["latency"]) < 20000000:
                latency_step.passed(
                    f'Latency (ns) is good. {statistics["latency"]}'
                )
            else:
                latency_step.failed(
                    f'Latency (ns) is too long ! {statistics["latency"]}'
                )

test③ ルータのインターフェース`no shutdown`

testセクションで v-dist-rtr01 の GigabitEthernet 0/0/0/0 をNSOからno shutdownし、インターフェースの状態をチェックします。

@aetest.test
def noshutdown_interface(self, dev, intf_name, intf_num):
    """
    No shutsown interface
    """
    nso_actions.NsoActions().noshut_xr_intf(
        dev, intf_name, intf_num
    )
    time.sleep(60)
    if nso_actions.NsoActions().get_xr_intf_status(
        dev, intf_name, intf_num
    ):
        self.passed(
            f'Interface {intf_name} {intf_num} is noshutdowned at {dev} !!'
        )
    else:
        self.failed(
            f'Interface {intf_name} {intf_num} is not noshutdowned at {dev} !!'
        )

cleanup IxNetworkトラフィックストップ

cleanupセクションでIxNetworkトラフィックをストップします。

@aetest.cleanup
def cleanup(self):
    """
    Stop ixia traffic
    """
    self.ixia.stop_traffic()

CommonCleanup

class CommonCleanup(aetest.CommonCleanup):

CommonCleanup にはサブセクションを2つ作成します。

subsection① 事後スナップショットの取得と比較

事前と同じように取得した Genie Ops オブジェクトスナップショットに比較対象外の設定します。 Genie Ops オブジェクトは==で比較することができます。差異がある場合はdiffを使うことでその差異を取得することができます。

@aetest.subsection
def check_snapshot_before_after(self, devobj_list, dev, snapshot_b):
    """
    Gathering feature snapshot after testing and check before and after
    """
    self.snapshot_a = {}
    for devobj in devobj_list:
        if devobj.name == dev:
            # Feature bgp
            bgpa = devobj.learn("bgp")
            bgpa.diff_ignore.append(
                '[info][instance][default][vrf][default][neighbor][(.*)][up_time]'
            )
            bgpa.diff_ignore.append(
                '[info][instance][default][vrf][default][neighbor][(.*)][bgp_neighbor_counters][messages][sent][totals]'
            )
            bgpa.diff_ignore.append(
                '[info][instance][default][vrf][default][neighbor][(.*)][bgp_neighbor_counters][messages][sent][keepalives]'
            )
            bgpa.diff_ignore.append(
                '[info][instance][default][vrf][default][neighbor][(.*)][bgp_neighbor_counters][messages][received][totals]'
            )
            bgpa.diff_ignore.append(
                '[info][instance][default][vrf][default][neighbor][(.*)][bgp_neighbor_counters][messages][received][keepalives]'
            )
            for k in bgpa.routes_per_peer["instance"]["default"]["vrf"]["default"]["neighbor"].keys():
                try:
                    bgpa.routes_per_peer["instance"]["default"]["vrf"]["default"]["neighbor"][
                        k]["address_family"]["vpnv4 unicast"]["msg_rcvd"] = "diff_ignore"
                except:
                    pass
            for k in bgpa.routes_per_peer["instance"]["default"]["vrf"]["default"]["neighbor"].keys():
                try:
                    bgpa.routes_per_peer["instance"]["default"]["vrf"]["default"]["neighbor"][
                        k]["address_family"]["vpnv4 unicast"]["msg_sent"] = "diff_ignore"
                except:
                    pass
            for k in bgpa.routes_per_peer["instance"]["default"]["vrf"]["default"]["neighbor"].keys():
                try:
                    bgpa.routes_per_peer["instance"]["default"]["vrf"]["default"]["neighbor"][
                        k]["address_family"]["vpnv4 unicast"]["up_down"] = "diff_ignore"
                except:
                    pass
            self.snapshot_a[devobj.name, "bgp"] = bgpa
            # Feature isis
            isisa = devobj.learn("isis")
            isisa.diff_ignore.append(
                '[info][instance][100][vrf][default][interfaces][(.*)][adjacencies][(.*)][neighbor_snpa][(.*)][level][level-2][ hold_timer]'
            )
            isisa.diff_ignore.append(
                '[info][instance][100][vrf][default][interfaces][(.*)][adjacencies][(.*)][neighbor_snpa][(.*)][level][level-2][lastuptime]'
            )
            isisa.diff_ignore.append(
                '[info][instance][100][vrf][default][lsp_log]'
            )
            isisa.diff_ignore.append(
                '[info][instance][100][vrf][default][spf_log]'
            )
            isisa.diff_ignore.append(
                '[info][instance][100][vrf][default][interfaces][(.*)][packet_counters]'
            )
            for k in isisa.lsdb["instance"]["100"]["vrf"]["default"]["level_db"][2].keys():
                try:
                    isisa.lsdb["instance"]["100"]["vrf"]["default"]["level_db"][2][k]["checksum"] = "diff_ignore"
                except:
                    pass
            for k in isisa.lsdb["instance"]["100"]["vrf"]["default"]["level_db"][2].keys():
                try:
                    isisa.lsdb["instance"]["100"]["vrf"]["default"]["level_db"][2][k]["sequence"] = "diff_ignore"
                except:
                    pass
            for k in isisa.lsdb["instance"]["100"]["vrf"]["default"]["level_db"][2].keys():
                try:
                    isisa.lsdb["instance"]["100"]["vrf"]["default"]["level_db"][2][k]["remaining_lifetime"] = "diff_ignore"
                except:
                    pass
            self.snapshot_a[devobj.name, "isis"] = isisa
            # Feature routing
            for kvrf in routinga.info["vrf"].keys():
                try:
                    for kroutes in routinga.info["vrf"][kvrf]["address_family"]["ipv4"]["routes"].keys():
                        try:
                            for index in routinga.info["vrf"][kvrf]["address_family"]["ipv4"]["routes"][kroutes]["next_hop"]["next_hop_list"].keys():
                                try:
                                    routinga.info["vrf"][kvrf]["address_family"]["ipv4"]["routes"][
                                        kroutes]["next_hop"]["next_hop_list"][index]["updated"] = "diff_ignore"
                                except:
                                    pass
                        except:
                            pass
                except:
                    pass

            for kvrf in routinga.info["vrf"].keys():
                try:
                    for kroutes in routinga.info["vrf"][kvrf]["address_family"]["ipv6"]["routes"].keys():
                        try:
                            for index in routinga.info["vrf"][kvrf]["address_family"]["ipv6"]["routes"][kroutes]["next_hop"]["next_hop_list"].keys():
                                try:
                                    routinga.info["vrf"][kvrf]["address_family"]["ipv6"]["routes"][
                                        kroutes]["next_hop"]["next_hop_list"][index]["updated"] = "diff_ignore"
                                except:
                                    pass
                        except:
                            pass
                except:
                    pass
            self.snapshot_a[devobj.name, "routing"] = routinga
            # Feature interface
            inta = devobj.learn("interface")
            inta.diff_ignore.append(
                '[info][(.*)][accounting]'
            )
            inta.diff_ignore.append(
                '[info][(.*)][counters]'
            )
            self.snapshot_a[devobj.name, "interface"] = inta
            # Feature arp
            arpa = devobj.learn("arp")
            arpa.diff_ignore.append(
                '[info][statistics]'
            )
            self.snapshot_a[devobj.name, "arp"] = arpa
            # Feature nd
            nda = devobj.learn("nd")
            nda.diff_ignore.append(
                '[info][interfaces][(.*)][neighbors][(.*)][age]'
            )
            self.snapshot_a[devobj.name, "nd"] = nda
        else:
            pass
    # Compare before and after snapshot
    for kb in snapshot_b.keys():
        items = snapshot_b[kb].diff(self.snapshot_a[kb]).diffs
        assert snapshot_b[kb] == self.snapshot_a[kb], f"Hey {kb[0]} {kb[1]} wrong {items} !!"

subsection② 全てのデバイスから`disconnect`

connect()したデバイスからdisconnect()します。

@aetest.subsection
def disconnect_common(self, devobj_list):
    """
    Disconnecting device.
    """
    for devobj in devobj_list:
        devobj.disconnect()

スタンドアローン実行

Standalone 、 EasypyJobfiles どちらでも実行できるようにします。
Standalone での実行でもtestbedオブジェクトがスクリプトに渡されるようにするために、argparseでスクリプトに渡すtypeを genie.testbed のloadに指定します。

if __name__ == "__main__":
    # Standalne running without easypy
    import argparse
    from genie.testbed import load

    parser = argparse.ArgumentParser(
        description="For supplying testbed object to test script parameter"
    )
    parser.add_argument(
        "--testbed",
        dest="testbed",
        help="testbed YAML",
        type=load,
        default=None
    )

    args = parser.parse_known_args()[0]
    res = aetest.main(testbed=args.testbed)
    aetest.exit_cli_code(res)

Standalone はaetest.main()でテストスクリプトを実行することができます。作成したオブジェクトをaetest.exit_cli_code()に渡せば、テスト結果にFailedや、Erroredなどがあればexit code 1で終了するので、GitlabなどのCI/CDパイプラインでの判定にも使用することができます。

(参考)ae_linkdown.py

ae_linkdown.py

from pyats import aetest
from unicon.core.errors import TimeoutError, StateMachineError, ConnectionError
import os
import time
import nso_actions
import ixia_functions
import logging

logger = logging.getLogger(__name__)


class CommonSetup(aetest.CommonSetup):

    @aetest.subsection
    def assert_testbed_common(self, testbed):
        """
        Assering testbed
        """
        logger.info(f"{os.path.basename(__file__)} test start")
        assert testbed, "Hey where is Testbed ?"

    @aetest.subsection
    def setup_device_objects_common(self, testbed):
        """
        Creating testbed device object list
        """
        self.testbed_dev_list = [
            dev for dev in testbed.devices.keys()
        ]
        self.devobj_list = [
            testbed.devices[devobj] for devobj in self.testbed_dev_list
        ]
        self.parent.parameters.update(
            devobj_list=self.devobj_list
        )

    @aetest.subsection
    def check_interface_status(self):
        """
        Checking target interface status
        """
        self.dev = "v-dist-rtr01"
        self.intf_name = "GigabitEthernet"
        self.intf_num = "0/0/0/0"
        self.parent.parameters.update(
            dev=self.dev
        )
        self.parent.parameters.update(
            intf_name=self.intf_name
        )
        self.parent.parameters.update(
            intf_num=self.intf_num
        )
        assert nso_actions.NsoActions().get_xr_intf_status(
            self.dev, self.intf_name, self.intf_num
        ), f"Hey interface {self.intf_name} {self.intf_name} of {self.dev} is shutdowned !!"

    @aetest.subsection
    def connect_common(self):
        """
        Connecting testbed all devices
        """

        for devobj in self.devobj_list:
            try:
                devobj.connect()
            except (TimeoutError, StateMachineError, ConnectionError):
                logger.info("Unable to connect to devices")

    @aetest.subsection
    def snapshot_before(self):
        """
        Gathering features snapshot before testing
        """
        self.snapshot_b = {}
        for devobj in self.devobj_list:
            if devobj.name == self.dev:
                # Feature bgp
                bgpb = devobj.learn("bgp")
                bgpb.diff_ignore.append(
                    '[info][instance][default][vrf][default][neighbor][(.*)][up_time]'
                )
                bgpb.diff_ignore.append(
                    '[info][instance][default][vrf][default][neighbor][(.*)][bgp_neighbor_counters][messages][sent][totals]'
                )
                bgpb.diff_ignore.append(
                    '[info][instance][default][vrf][default][neighbor][(.*)][bgp_neighbor_counters][messages][sent][keepalives]'
                )
                bgpb.diff_ignore.append(
                    '[info][instance][default][vrf][default][neighbor][(.*)][bgp_neighbor_counters][messages][received][totals]'
                )
                bgpb.diff_ignore.append(
                    '[info][instance][default][vrf][default][neighbor][(.*)][bgp_neighbor_counters][messages][received][keepalives]'
                )
                for k in bgpb.routes_per_peer["instance"]["default"]["vrf"]["default"]["neighbor"].keys():
                    try:
                        bgpb.routes_per_peer["instance"]["default"]["vrf"]["default"]["neighbor"][
                            k]["address_family"]["vpnv4 unicast"]["msg_rcvd"] = "diff_ignore"
                    except:
                        pass
                for k in bgpb.routes_per_peer["instance"]["default"]["vrf"]["default"]["neighbor"].keys():
                    try:
                        bgpb.routes_per_peer["instance"]["default"]["vrf"]["default"]["neighbor"][
                            k]["address_family"]["vpnv4 unicast"]["msg_sent"] = "diff_ignore"
                    except:
                        pass
                for k in bgpb.routes_per_peer["instance"]["default"]["vrf"]["default"]["neighbor"].keys():
                    try:
                        bgpb.routes_per_peer["instance"]["default"]["vrf"]["default"]["neighbor"][
                            k]["address_family"]["vpnv4 unicast"]["up_down"] = "diff_ignore"
                    except:
                        pass
                self.snapshot_b[devobj.name, "bgp"] = bgpb
                # Feature isis
                isisb = devobj.learn("isis")
                isisb.diff_ignore.append(
                    '[info][instance][100][vrf][default][interfaces][(.*)][adjacencies][(.*)][neighbor_snpa][(.*)][level][level-2][hold_timer]'
                )
                isisb.diff_ignore.append(
                    '[info][instance][100][vrf][default][interfaces][(.*)][adjacencies][(.*)][neighbor_snpa][(.*)][level][level-2][lastuptime]'
                )
                isisb.diff_ignore.append(
                    '[info][instance][100][vrf][default][lsp_log]'
                )
                isisb.diff_ignore.append(
                    '[info][instance][100][vrf][default][spf_log]'
                )
                isisb.diff_ignore.append(
                    '[info][instance][100][vrf][default][interfaces][(.*)][packet_counters]'
                )
                for k in isisb.lsdb["instance"]["100"]["vrf"]["default"]["level_db"][2].keys():
                    try:
                        isisb.lsdb["instance"]["100"]["vrf"]["default"]["level_db"][2][k]["checksum"] = "diff_ignore"
                    except:
                        pass
                for k in isisb.lsdb["instance"]["100"]["vrf"]["default"]["level_db"][2].keys():
                    try:
                        isisb.lsdb["instance"]["100"]["vrf"]["default"]["level_db"][2][k]["sequence"] = "diff_ignore"
                    except:
                        pass
                for k in isisb.lsdb["instance"]["100"]["vrf"]["default"]["level_db"][2].keys():
                    try:
                        isisb.lsdb["instance"]["100"]["vrf"]["default"]["level_db"][2][k]["remaining_lifetime"] = "diff_ignore"
                    except:
                        pass
                self.snapshot_b[devobj.name, "isis"] = isisb
                # Feature routing
                routingb = devobj.learn("routing")
                for kvrf in routingb.info["vrf"].keys():
                    try:
                        for kroutes in routingb.info["vrf"][kvrf]["address_family"]["ipv4"]["routes"].keys():
                            try:
                                for index in routingb.info["vrf"][kvrf]["address_family"]["ipv4"]["routes"][kroutes]["next_hop"]["next_hop_list"].keys():
                                    try:
                                        routingb.info["vrf"][kvrf]["address_family"]["ipv4"]["routes"][
                                            kroutes]["next_hop"]["next_hop_list"][index]["updated"] = "diff_ignore"
                                    except:
                                        pass
                            except:
                                pass
                    except:
                        pass

                for kvrf in routingb.info["vrf"].keys():
                    try:
                        for kroutes in routingb.info["vrf"][kvrf]["address_family"]["ipv6"]["routes"].keys():
                            try:
                                for index in routingb.info["vrf"][kvrf]["address_family"]["ipv6"]["routes"][kroutes]["next_hop"]["next_hop_list"].keys():
                                    try:
                                        routingb.info["vrf"][kvrf]["address_family"]["ipv6"]["routes"][
                                            kroutes]["next_hop"]["next_hop_list"][index]["updated"] = "diff_ignore"
                                    except:
                                        pass
                            except:
                                pass
                    except:
                        pass
                self.snapshot_b[devobj.name, "routing"] = routingb
                # Feature interface
                intb = devobj.learn("interface")
                intb.diff_ignore.append(
                    '[info][(.*)][accounting]'
                )
                intb.diff_ignore.append(
                    '[info][(.*)][counters]'
                )
                self.snapshot_b[devobj.name, "interface"] = intb
                # Feature arp
                arpb = devobj.learn("arp")
                arpb.diff_ignore.append(
                    '[info][statistics]'
                )
                self.snapshot_b[devobj.name, "arp"] = arpb
                # Feature nd
                ndb = devobj.learn("nd")
                ndb.diff_ignore.append(
                    '[info][interfaces][(.*)][neighbors][(.*)][age]'
                )
                self.snapshot_b[devobj.name, "nd"] = ndb
            else:
                pass
        self.parent.parameters.update(
            snapshot_b=self.snapshot_b
        )


class TestCaseTrafficStatisticsCheck(aetest.Testcase):

    @aetest.setup
    def setup(self):
        """
        Start ixia traffic
        """
        self.ixia = ixia_functions.IxiaFunctions()
        self.ixia.start_traffic()

    @aetest.test
    def shutdown_interface(self, dev, intf_name, intf_num):
        """
        Shutdown interface
        """
        time.sleep(120)
        nso_actions.NsoActions().shut_xr_intf(
            dev, intf_name, intf_num
        )
        time.sleep(60)
        if not nso_actions.NsoActions().get_xr_intf_status(
            dev, intf_name, intf_num
        ):
            self.passed(
                f'Interface {intf_name} {intf_num} is shutdowned at {dev} !!'
            )
        else:
            self.failed(
                f'Interface {intf_name} {intf_num} is not shutdowned at {dev} !!'
            )

    @aetest.test
    def traffic_statistics(self, steps):
        """
        Viewing ixia traffic statistics
        """
        statistics = self.ixia.view_statistics()
        if not statistics:
            self.errored('IXIA statistics is nothing !!')
        else:
            with steps.start(
                "Asserting duration",
                description='Asserting duration',
                continue_=True
            ) as duration_step:
                if float(statistics["duration"]) < 100:
                    duration_step.passed(
                        f'Packet Loss Duration (ms) is good. {statistics["duration"]}'
                    )
                else:
                    duration_step.failed(
                        f'Packet Loss Duration (ms) is too long ! {statistics["duration"]}'
                    )
            with steps.start(
                "Asserting latency",
                description='Asserting latency',
                continue_=True
            ) as latency_step:
                if float(statistics["latency"]) < 20000000:
                    latency_step.passed(
                        f'Latency (ns) is good. {statistics["latency"]}'
                    )
                else:
                    latency_step.failed(
                        f'Latency (ns) is too long ! {statistics["latency"]}'
                    )

    @aetest.test
    def noshutdown_interface(self, dev, intf_name, intf_num):
        """
        No shutsown interface
        """
        nso_actions.NsoActions().noshut_xr_intf(
            dev, intf_name, intf_num
        )
        time.sleep(60)
        if nso_actions.NsoActions().get_xr_intf_status(
            dev, intf_name, intf_num
        ):
            self.passed(
                f'Interface {intf_name} {intf_num} is noshutdowned at {dev} !!'
            )
        else:
            self.failed(
                f'Interface {intf_name} {intf_num} is not noshutdowned at {dev} !!'
            )

    @aetest.cleanup
    def cleanup(self):
        """
        Stop ixia traffic
        """
        self.ixia.stop_traffic()


class CommonCleanup(aetest.CommonCleanup):

    @aetest.subsection
    def check_snapshot_before_after(self, devobj_list, dev, snapshot_b):
        """
        Gathering feature snapshot after testing and check before and after
        """
        self.snapshot_a = {}
        for devobj in devobj_list:
            if devobj.name == dev:
                # Feature bgp
                bgpa = devobj.learn("bgp")
                bgpa.diff_ignore.append(
                    '[info][instance][default][vrf][default][neighbor][(.*)][up_time]'
                )
                bgpa.diff_ignore.append(
                    '[info][instance][default][vrf][default][neighbor][(.*)][bgp_neighbor_counters][messages][sent][totals]'
                )
                bgpa.diff_ignore.append(
                    '[info][instance][default][vrf][default][neighbor][(.*)][bgp_neighbor_counters][messages][sent][keepalives]'
                )
                bgpa.diff_ignore.append(
                    '[info][instance][default][vrf][default][neighbor][(.*)][bgp_neighbor_counters][messages][received][totals]'
                )
                bgpa.diff_ignore.append(
                    '[info][instance][default][vrf][default][neighbor][(.*)][bgp_neighbor_counters][messages][received][keepalives]'
                )
                for k in bgpa.routes_per_peer["instance"]["default"]["vrf"]["default"]["neighbor"].keys():
                    try:
                        bgpa.routes_per_peer["instance"]["default"]["vrf"]["default"]["neighbor"][
                            k]["address_family"]["vpnv4 unicast"]["msg_rcvd"] = "diff_ignore"
                    except:
                        pass
                for k in bgpa.routes_per_peer["instance"]["default"]["vrf"]["default"]["neighbor"].keys():
                    try:
                        bgpa.routes_per_peer["instance"]["default"]["vrf"]["default"]["neighbor"][
                            k]["address_family"]["vpnv4 unicast"]["msg_sent"] = "diff_ignore"
                    except:
                        pass
                for k in bgpa.routes_per_peer["instance"]["default"]["vrf"]["default"]["neighbor"].keys():
                    try:
                        bgpa.routes_per_peer["instance"]["default"]["vrf"]["default"]["neighbor"][
                            k]["address_family"]["vpnv4 unicast"]["up_down"] = "diff_ignore"
                    except:
                        pass
                self.snapshot_a[devobj.name, "bgp"] = bgpa
                # Feature isis
                isisa = devobj.learn("isis")
                isisa.diff_ignore.append(
                    '[info][instance][100][vrf][default][interfaces][(.*)][adjacencies][(.*)][neighbor_snpa][(.*)][level][level-2][ hold_timer]'
                )
                isisa.diff_ignore.append(
                    '[info][instance][100][vrf][default][interfaces][(.*)][adjacencies][(.*)][neighbor_snpa][(.*)][level][level-2][lastuptime]'
                )
                isisa.diff_ignore.append(
                    '[info][instance][100][vrf][default][lsp_log]'
                )
                isisa.diff_ignore.append(
                    '[info][instance][100][vrf][default][spf_log]'
                )
                isisa.diff_ignore.append(
                    '[info][instance][100][vrf][default][interfaces][(.*)][packet_counters]'
                )
                for k in isisa.lsdb["instance"]["100"]["vrf"]["default"]["level_db"][2].keys():
                    try:
                        isisa.lsdb["instance"]["100"]["vrf"]["default"]["level_db"][2][k]["checksum"] = "diff_ignore"
                    except:
                        pass
                for k in isisa.lsdb["instance"]["100"]["vrf"]["default"]["level_db"][2].keys():
                    try:
                        isisa.lsdb["instance"]["100"]["vrf"]["default"]["level_db"][2][k]["sequence"] = "diff_ignore"
                    except:
                        pass
                for k in isisa.lsdb["instance"]["100"]["vrf"]["default"]["level_db"][2].keys():
                    try:
                        isisa.lsdb["instance"]["100"]["vrf"]["default"]["level_db"][2][k]["remaining_lifetime"] = "diff_ignore"
                    except:
                        pass
                self.snapshot_a[devobj.name, "isis"] = isisa
                # Feature routing
                for kvrf in routinga.info["vrf"].keys():
                    try:
                        for kroutes in routinga.info["vrf"][kvrf]["address_family"]["ipv4"]["routes"].keys():
                            try:
                                for index in routinga.info["vrf"][kvrf]["address_family"]["ipv4"]["routes"][kroutes]["next_hop"]["next_hop_list"].keys():
                                    try:
                                        routinga.info["vrf"][kvrf]["address_family"]["ipv4"]["routes"][
                                            kroutes]["next_hop"]["next_hop_list"][index]["updated"] = "diff_ignore"
                                    except:
                                        pass
                            except:
                                pass
                    except:
                        pass

                for kvrf in routinga.info["vrf"].keys():
                    try:
                        for kroutes in routinga.info["vrf"][kvrf]["address_family"]["ipv6"]["routes"].keys():
                            try:
                                for index in routinga.info["vrf"][kvrf]["address_family"]["ipv6"]["routes"][kroutes]["next_hop"]["next_hop_list"].keys():
                                    try:
                                        routinga.info["vrf"][kvrf]["address_family"]["ipv6"]["routes"][
                                            kroutes]["next_hop"]["next_hop_list"][index]["updated"] = "diff_ignore"
                                    except:
                                        pass
                            except:
                                pass
                    except:
                        pass
                self.snapshot_a[devobj.name, "routing"] = routinga
                # Feature interface
                inta = devobj.learn("interface")
                inta.diff_ignore.append(
                    '[info][(.*)][accounting]'
                )
                inta.diff_ignore.append(
                    '[info][(.*)][counters]'
                )
                self.snapshot_a[devobj.name, "interface"] = inta
                # Feature arp
                arpa = devobj.learn("arp")
                arpa.diff_ignore.append(
                    '[info][statistics]'
                )
                self.snapshot_a[devobj.name, "arp"] = arpa
                # Feature nd
                nda = devobj.learn("nd")
                nda.diff_ignore.append(
                    '[info][interfaces][(.*)][neighbors][(.*)][age]'
                )
                self.snapshot_a[devobj.name, "nd"] = nda
            else:
                pass
        # Compare before and after snapshot
        for kb in snapshot_b.keys():
            items = snapshot_b[kb].diff(self.snapshot_a[kb]).diffs
            assert snapshot_b[kb] == self.snapshot_a[kb], f"Hey {kb[0]} {kb[1]} wrong {items} !!"

    @aetest.subsection
    def disconnect_common(self, devobj_list):
        """
        Disconnecting device.
        """
        for devobj in devobj_list:
            devobj.disconnect()


if __name__ == "__main__":
    # Standalne running without easypy
    import argparse
    from genie.testbed import load

    parser = argparse.ArgumentParser(
        description="For supplying testbed object to test script parameter"
    )
    parser.add_argument(
        "--testbed",
        dest="testbed",
        help="testbed YAML",
        type=load,
        default=None
    )

    args = parser.parse_known_args()[0]
    res = aetest.main(testbed=args.testbed)
    aetest.exit_cli_code(res)

実行結果

途中のログ出力は省略し、テスト結果だけを掲載しています。
優先パスであるインターフェースをshutdownしても、Packet Loss Duration (ms)は、81.920で Passed 、Store-Forward Min Latency (ns)も16000640で Passed となっています。
また、前後で取得したスナップショットも差異がなく、 Passed になっています。

root@422ae12e5275:~/nail# python3 ./test/ae_linkdown.py --testbed ./testbed/testbed_v.yaml 
+------------------------------------------------------------------------------+
|                            Starting common setup                             |
+------------------------------------------------------------------------------+
+------------------------------------------------------------------------------+
|                  Starting subsection assert_testbed_common                   |
+------------------------------------------------------------------------------+
The result of subsection assert_testbed_common is => PASSED
+------------------------------------------------------------------------------+
|               Starting subsection setup_device_objects_common                |
+------------------------------------------------------------------------------+
The result of subsection setup_device_objects_common is => PASSED
+------------------------------------------------------------------------------+
|                  Starting subsection check_interface_status                  |
+------------------------------------------------------------------------------+
The result of subsection check_interface_status is => PASSED
+------------------------------------------------------------------------------+
|                      Starting subsection connect_common                      |
+------------------------------------------------------------------------------+

（省略）

The result of subsection connect_common is => PASSED
+------------------------------------------------------------------------------+
|                     Starting subsection snapshot_before                      |
+------------------------------------------------------------------------------+

（省略）


The result of subsection snapshot_before is => PASSED
The result of common setup is => PASSED
+------------------------------------------------------------------------------+
|               Starting testcase TestCaseTrafficStatisticsCheck               |
+------------------------------------------------------------------------------+
+------------------------------------------------------------------------------+
|                            Starting section setup                            |
+------------------------------------------------------------------------------+

（省略）


The result of section setup is => PASSED
+------------------------------------------------------------------------------+
|                     Starting section shutdown_interface                      |
+------------------------------------------------------------------------------+
Passed reason: Interface GigabitEthernet 0/0/0/0 is shutdowned at v-dist-rtr01 !!
The result of section shutdown_interface is => PASSED
+------------------------------------------------------------------------------+
|                     Starting section traffic_statistics                      |
+------------------------------------------------------------------------------+

（省略）

2022-03-26 06:34:51 [ixnetwork_restpy.connection] [INFO] Row:0  View:Traffic Item Statistics  Sampled:2022-03-26 06:34:51.497100 UTC
	Traffic Item: Traffic Item
	Tx Frames: 2322
	Rx Frames: 2321
	Frames Delta: 1
	Loss %: 0.043
	Packet Loss Duration (ms): 81.920
	Tx Frame Rate: 11.993
	Rx Frame Rate: 11.993
	Tx L1 Rate (bps): 100168.321
	Rx L1 Rate (bps): 100168.321
	Rx Bytes: 2376704
	Tx Rate (Bps): 12281.174
	Rx Rate (Bps): 12281.174
	Tx Rate (bps): 98249.388
	Rx Rate (bps): 98249.388
	Tx Rate (Kbps): 98.249
	Rx Rate (Kbps): 98.249
	Tx Rate (Mbps): 0.098
	Rx Rate (Mbps): 0.098
	Store-Forward Avg Latency (ns): 3780497
	Store-Forward Min Latency (ns): 0
	Store-Forward Max Latency (ns): 16000640
	First TimeStamp: 00:00:01.249
	Last TimeStamp: 00:03:11.007

（省略）

2022-03-26 06:34:53 [ixnetwork_restpy.connection] [INFO] 
Traffic Item Stats:
	TxFrames: 2322  RxFrames: 2321  Duration: 81.920  Latency(ns): 16000640 

2022-03-26T06:34:53: %IXNETWORK_RESTPY-INFO: 
2022-03-26T06:34:53: %IXNETWORK_RESTPY-INFO: Traffic Item Stats:
2022-03-26T06:34:53: %IXNETWORK_RESTPY-INFO: 	TxFrames: 2322  RxFrames: 2321  Duration: 81.920  Latency(ns): 16000640 
+..............................................................................+
:                     Starting STEP 1: Asserting duration                      :
+..............................................................................+
Passed reason: Packet Loss Duration (ms) is good. 81.920
The result of STEP 1: Asserting duration is => PASSED
+..............................................................................+
:                      Starting STEP 2: Asserting latency                      :
+..............................................................................+
Passed reason: Latency (ns) is good. 16000640
The result of STEP 2: Asserting latency is => PASSED
+..........................................................+
:                       STEPS Report                       :
+..........................................................+
STEP 1 - Asserting duration                       Passed    
STEP 2 - Asserting latency                        Passed    
............................................................
The result of section traffic_statistics is => PASSED
+------------------------------------------------------------------------------+
|                    Starting section noshutdown_interface                     |
+------------------------------------------------------------------------------+
Passed reason: Interface GigabitEthernet 0/0/0/0 is noshutdowned at v-dist-rtr01 !!
The result of section noshutdown_interface is => PASSED
+------------------------------------------------------------------------------+
|                           Starting section cleanup                           |
+------------------------------------------------------------------------------+

（省略）


The result of section cleanup is => PASSED
The result of testcase TestCaseTrafficStatisticsCheck is => PASSED
+------------------------------------------------------------------------------+
|                           Starting common cleanup                            |
+------------------------------------------------------------------------------+
+------------------------------------------------------------------------------+
|               Starting subsection check_snapshot_before_after                |
+------------------------------------------------------------------------------+

（省略）

The result of subsection check_snapshot_before_after is => PASSED
+------------------------------------------------------------------------------+
|                    Starting subsection disconnect_common                     |
+------------------------------------------------------------------------------+
The result of subsection disconnect_common is => PASSED
The result of common cleanup is => PASSED
+------------------------------------------------------------------------------+
|                               Detailed Results                               |
+------------------------------------------------------------------------------+
 SECTIONS/TESTCASES                                                      RESULT   
--------------------------------------------------------------------------------
.
|-- common_setup                                                          PASSED
|   |-- assert_testbed_common                                             PASSED
|   |-- setup_device_objects_common                                       PASSED
|   |-- check_interface_status                                            PASSED
|   |-- connect_common                                                    PASSED
|   `-- snapshot_before                                                   PASSED
|-- TestCaseTrafficStatisticsCheck                                        PASSED
|   |-- setup                                                             PASSED
|   |-- shutdown_interface                                                PASSED
|   |-- traffic_statistics                                                PASSED
|   |   |-- Step 1: Asserting duration                                    PASSED
|   |   `-- Step 2: Asserting latency                                     PASSED
|   |-- noshutdown_interface                                              PASSED
|   `-- cleanup                                                           PASSED
`-- common_cleanup                                                        PASSED
    |-- check_snapshot_before_after                                       PASSED
    `-- disconnect_common                                                 PASSED
+------------------------------------------------------------------------------+
|                                   Summary                                    |
+------------------------------------------------------------------------------+
 Number of ABORTED                                                            0 
 Number of BLOCKED                                                            0 
 Number of ERRORED                                                            0 
 Number of FAILED                                                             0 
 Number of PASSED                                                             3 
 Number of PASSX                                                              0 
 Number of SKIPPED                                                            0 
 Total Number                                                                 3 
 Success Rate                                                            100.0% 
--------------------------------------------------------------------------------

まとめ

今回のようにpyATSのAEtestで作成したテストスクリプトはスタンドアローンでも実行できるので、ちょっとしたテストにも使うことができますし、汎用性の高いスクリプトをたくさん作成し、EasypyJobfilesで組み合わせて実行するなども可能だと思います。
pyATS/Genieだけでも使えるライブラリが数多くある上、他のライブラリと合わせて使用することできるので、可能性は無縁大だと思います。

参考リンク