初めに
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のオープソース自動化ライブラリです。チュートリアルやサンプルスクリプトが公開されています。
テストスクリプトで使用する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だけでも使えるライブラリが数多くある上、他のライブラリと合わせて使用することできるので、可能性は無縁大だと思います。