この記事は富士通の有志によるFUJITSU Advent Calendar 2023の20日目の投稿です。
https://qiita.com/advent-calendar/2023/fujitsu
はじめに
はじめまして、koishiと申します。
とある富士通のグループ会社でネットワークまわりの技術支援とかとかのお仕事をさせて頂いております。
この記事はなに?
PyATSで富士通ルータであるSi-Rをテストをしようとちょっと頑張った記録です。
どなたかのお役に立てれば幸いです。
PyATS(Python Automated Test Systems)とは
Cisco社が開発しているPythonベースのテスト自動化ソリューションとなります。
読み方は"ぱいえーてぃーえす"とのこと
ネットワーク屋のテストについて
ネットワーク屋のテストは複雑な経路のテストとかもありますが、それだけでなくバージョン番号をチェックするだけの単純な確認テストをネットワーク機器の数だけしないといけないということもあります(3桁4桁の台数あるときも)
あるプロジェクトでPyATSのトライしてみたところ、とあるテストの工数が200分の1になったとかならないとか。
なんで書いた
そんなPyATSで富士通製NW機器のテストできたらいいのにな・・・
そう言われたのは、現場SEさんにPyATSのご紹介を終えたあとの一言でした。
前述の通り、PyATSは便利なソリューションではありますが、
残念なことに富士通製のネットワーク機器は対応しておりません。
そのため幾度となく"富士通製は対応してないんですよね。。。"とお茶を濁して、ごめんなさいをしておりました。
しかしながら、アドベントカレンダーのネタに困っていたので、やってみました。
PyATSで3rd party機器(Si-R)のテスト実現に必要なこと
この企画を実現するには
1.Uniconライブラリ と
2.Genieパーサー
に手を加える必要があるとわかりました。
UniconライブラリはNW機器に接続するためのコネクションライブラリで、接続のタイミングで必要なコマンドとかうまいことしてくれるものです。
GenieパーサーはNW機器から取得したshowコマンド等の出力をうまいことJSONにパーッスしてくれるものです。
※UniconライブラリとGenieパーサーのサポート状況はリンクの通りです。(前述のとおり富士通製の機器は対応していません)
様々なアプローチがあるとは思いますが、今回はそれぞれ以下の対応としました。
1.Unicon Pluginをつくろう
2.TextFSMで解析しよう(GenieParserを頑張るのはまたの機会!)
各対応についてご紹介していきます。
Unicon Pluginをつくろう
Unicon Pluginとは3rd party機器向けの独自のUniconライブラリをつくれるものです。
そのために5つのステップがあります。
1.Connection Classの作成
2.Connection Providerの作成
3.StateMachineの作成
4.Service定義の作成
5.setupファイルの作成
6.作ったUnicon Pluginのインストール
一つずつフォーカスしていきたいと思います。
1.Connection Classの作成
Connection Classは機器接続のための最初の定義で、ここにはOSの種類やシャーシタイプ、モデル(今回は省略)のほか、以下のようなものが定義され、機器接続の基盤となります。
機器の状態遷移を示すクラス(state_machine_class)
接続方式を示すクラス(connection_provider_class)
機器へのサービスの指定(subcommand_list)
機器接続の設定(settings)
サンプル(__init__.py)
__init__.py
from unicon.bases.routers.connection import BaseSingleRpConnection
from .statemachine import sirSingleRpStateMachine
from .provider import sirConnectionProvider
from .services import sirServiceList
from .settings import sirSettings
class sirSingleRPConnection(BaseSingleRpConnection):
os = 'sir'
chassis_type = 'single_rp'
state_machine_class = sirSingleRpStateMachine
connection_provider_class = sirConnectionProvider
subcommand_list = sirServiceList
settings = sirSettings()
また初期にページャーを停止したいのでsettings.pyに記載しています。
エラーパターンはお借りしたサンプルスクリプトからSi-R用にアップデートできていません。。。
サンプル(settings.py)
settings.py
from unicon.plugins.generic.settings import GenericSettings
# see GenericSettings class to confirm default value
class sirSettings(GenericSettings):
def __init__(self):
super().__init__()
self.CONNECTION_TIMEOUT = 60 * 5
# overwrite init_exec_commands
self.HA_INIT_EXEC_COMMANDS = ['terminal pager disable']
# overwrite init_config_commands
self.HA_INIT_CONFIG_COMMANDS = []
# append ERROR_PATTERN
self.ERROR_PATTERN += [
r"<ERROR> Invalid input detected at '\^' marker\.",
r"<ERROR> '\S+' is Unrecognized command",
r"<ERROR> Unrecognized command",
]
# append CONFIGURE_ERROR_PATTERN
self.CONFIGURE_ERROR_PATTERN += [
r"<ERROR> Invalid input detected at '\^' marker\.",
r'\S+: No such file or directory',
]
# see less_prompt_handler in statements.py
self.LESS_CONTINUE = ' '
# overwrite RELOAD_TIMEOUT, see services.py
# interval 30 sec, 20 attempts, total 600 sec
self.RELOAD_RECONNECT_ATTEMPTS = 20 # default 3
self.RELOAD_WAIT = 30 # default 240
self.RELOAD_TIMEOUT = 600 # default 300
2.Connection Providerの作成
Connection ProviderはNW機器の接続プロセスを定義するものです。
サンプルではsingleRP機器の接続プロセスをベースとしています。
サンプル(provider.py)
provider.py
from unicon.plugins.generic.connection_provider import GenericSingleRpConnectionProvider
from unicon.eal.dialogs import Dialog
from .statemachine import sirStatements
# see __init__.py
class sirConnectionProvider(GenericSingleRpConnectionProvider):
def get_connection_dialog(self):
statements = sirStatements()
# TODO: telnet login
dialog = super().get_connection_dialog()
dialog += Dialog([statements.bad_password_stmt])
return dialog
サンプル(statements.py)
statements.py
#
# ダイアログを自動処理するための文(statements)を定義する
#
from unicon.eal.dialogs import Statement
from unicon.plugins.generic.statements import enable_password_handler
from unicon.plugins.generic.statements import bad_password_handler
from .patterns import sirPatterns
def less_prompt_handler(spawn):
# remove ':' here works properly, but junk '\r' left
# spawn.match.match_output = spawn.buffer.replace('\r\n:', '\r\n')
# to remove ':' see services.py
spawn.match.match_output = spawn.buffer
# send ' '
spawn.send(spawn.settings.LESS_CONTINUE)
class sirStatements():
def __init__(self) -> None:
patterns = sirPatterns()
# see statemachine.py
self.enable_password_stmt = Statement(pattern=patterns.enable_password,
action=enable_password_handler,
args=None,
loop_continue=True,
continue_timer=False)
# see statemachine.py
# see provider.py
self.bad_password_stmt = Statement(pattern=patterns.bad_password,
action=bad_password_handler,
args=None,
loop_continue=False,
continue_timer=False)
# see services.py
self.less_stmt = Statement(pattern=patterns.less_prompt,
action=less_prompt_handler,
args=None,
loop_continue=True,
continue_timer=False)
# see services.py
self.save_stmt = Statement(pattern=patterns.confirm_save,
action='sendline(y)',
args=None,
loop_continue=True,
continue_timer=False)
# see services.py
self.load_stmt = Statement(pattern=patterns.confirm_load,
action='sendline(y)',
args=None,
loop_continue=True,
continue_timer=False)
# see services.py
self.restore_stmt = Statement(pattern=patterns.confirm_restore,
action='sendline(y)',
args=None,
loop_continue=True,
continue_timer=False)
# see services.py
self.refresh_stmt = Statement(pattern=patterns.confirm_refresh,
action='sendline(y)',
args=None,
loop_continue=True,
continue_timer=False)
# see services.py
self.reset_stmt = Statement(pattern=patterns.confirm_reset,
action='sendline(y)',
args=None,
loop_continue=True,
continue_timer=False)
3.StateMachineの作成
StateMachineはNW機器の状態および各状態のプロンプト定義するものです。Cisco機器の特権モード、コンフィグレーションモードと同じようにSi-Rもモードがありますので、それを定義するものです。
Si-R G100B/G110B/G200B コマンドユーザーズガイドより引用
まずは状態を定義します。
以下のサンプルでは、showコマンドのみチャレンジしたかったため、運用管理モードの一般ユーザクラス(disable)から管理者クラス(enable)に相当する箇所を読み替えたものとなっております。
サンプル(statemachine.py)
statemachine.py
from unicon.eal.dialogs import Dialog
from unicon.statemachine import State, Path
from unicon.statemachine import StateMachine
from .patterns import sirPatterns
from .statements import sirStatements
# see __init__.py
class sirSingleRpStateMachine(StateMachine):
def create(self):
patterns = sirPatterns()
statements = sirStatements()
# 'disable' State definition
disable = State('disable', patterns.disable_prompt)
# 'enable' State definition
enable = State('enable', patterns.enable_prompt)
# 'config' State definition
config = State('config', patterns.config_prompt)
disable_to_enable = Path(disable, enable, 'enable',
Dialog([statements.enable_password_stmt,
statements.bad_password_stmt]))
enable_to_disable = Path(enable, disable, 'disable', None)
enable_to_config = Path(enable, config, 'config terminal', None)
config_to_enable = Path(config, enable, 'end', None)
self.add_state(disable)
self.add_state(enable)
self.add_state(config)
self.add_path(disable_to_enable)
self.add_path(enable_to_disable)
self.add_path(enable_to_config)
self.add_path(config_to_enable)
次は各状態におけるプロンプトを定義し、プロンプトから現在の状態を判別します。
showコマンドのみチャレンジのため、運用管理モードの一般ユーザークラス、管理者クラスに該当するもののみ読み替えてアップデートしており、それ以外はお借りしたサンプルスクリプトからアップデートできていない箇所があります。
サンプル(patterns.py)
patterns.py
class sirPatterns:
def __init__(self) -> None:
# priv mode, #
# priv mode, hostname#
self.enable_prompt = r'^(.*?)%N{0,1}#\s?$'
# user mode, >
# user mode, hostname>
self.disable_prompt = r'^(.*?)%N{0,1}>\s?$'
self.config_prompt = r'^(.*?)%N{0,1}\(config.*\)#\s*$'
self.enable_password = r'^Password:\s?$'
# password:
# <ERROR> Authentication failed
# password:
self.bad_password = r'^<ERROR> Authentication failed$'
# pager less is admind by default
self.less_prompt = r'\r\n:'
# save
# save ok?[y/N]:
self.confirm_save = r'^(.*?)save ok\?\[y/N\]:\s?$'
# load
# load ok?[y/N]:
self.confirm_load = r'^(.*?)load ok\?\[y/N\]:\s?$'
# restore
# restore ok?[y/N]:
self.confirm_restore = r'^(.*?)restore ok\?\[y/N\]:\s?$'
# refresh
# refresh ok?[y/N]:
self.confirm_refresh = r'^(.*?)refresh ok\?\[y/N\]:\s?$'
# reset
# reset ok?[y/N]:
self.confirm_reset = r'^(.*?)reset ok\?\[y/N\]:\s?$'
# telnet login
self.login_prompt = r'Login: *?'
self.password_prompt = r'Password: *?'
4.Service定義の作成
ここでは、コマンドの実行や設定等、PyATSが提供するサービスについて定義をします。今回はshowコマンドの実行のみのため、下記サンプルでは不要なサービスも残ってしまっていますが、ご容赦ください。
サンプル(services.py)
services.py
import logging
from unicon.eal.dialogs import Dialog
## create service from scratch
# from unicon.core.errors import SubCommandFailure
# from unicon.bases.routers.services import BaseService
## extend generic plugin services
from unicon.plugins.generic.service_implementation import Execute as GenericExecute
from unicon.plugins.generic.service_implementation import Configure as GenericConfigure
from unicon.plugins.generic.service_implementation import Reload as GenericReload
from unicon.plugins.generic import service_implementation as svc
from unicon.plugins.generic.utils import GenericUtils
from .statements import sirStatements
logger = logging.getLogger(__name__)
class sirExecute(GenericExecute):
def __init__(self, connection, context, **kwargs):
super().__init__(connection, context, **kwargs)
# if 'no more' is not sent, less pager is admind
self.is_pager_admind = connection.init_exec_commands is not None and 'terminal pager disable' not in connection.init_exec_commands
# in case of pager admind, add extra Dialog
if self.is_pager_admind:
statements = sirStatements()
self.dialog += Dialog([statements.less_stmt])
def call_service(self, *args, **kwargs):
super().call_service(*args, **kwargs)
def extra_output_process(self, output):
# in case of pager admind, strip ':'
if self.is_pager_admind:
utils = GenericUtils()
output = utils.remove_backspace_ansi_escape(output)
output = output.replace('\r\n:\r', '\r\n')
return output
class sirConfigure(GenericConfigure):
def __init__(self, connection, context, **kwargs):
super().__init__(connection, context, **kwargs)
self.start_state = 'configure'
self.end_state = 'admin'
self.valid_transition_commands = ['end']
def call_service(self, command=[], reply=Dialog([]), timeout=None, *args, **kwargs):
super().call_service(command, reply=reply, timeout=timeout, *args, **kwargs)
class sirSave(GenericExecute):
def __init__(self, connection, context, **kwargs):
super().__init__(connection, context, **kwargs)
self.start_state = 'admin'
self.end_state = 'admin'
def call_service(self, *args, **kwargs):
if 'moff' not in args:
statements = sirStatements()
self.dialog += Dialog([statements.save_stmt])
command = ' '.join(['save'] + list(args))
super().call_service(command, **kwargs)
class sirLoad(GenericExecute):
def __init__(self, connection, context, **kwargs):
super().__init__(connection, context, **kwargs)
self.start_state = 'admin'
self.end_state = 'admin'
def call_service(self, *args, **kwargs):
if 'moff' not in args:
statements = sirStatements()
self.dialog += Dialog([statements.load_stmt])
command = ' '.join(['load'] + list(args))
super().call_service(command, **kwargs)
class sirRestore(GenericExecute):
def __init__(self, connection, context, **kwargs):
super().__init__(connection, context, **kwargs)
self.start_state = 'admin'
self.end_state = 'admin'
def call_service(self, *args, **kwargs):
if 'moff' not in args:
statements = sirStatements()
self.dialog += Dialog([statements.restore_stmt])
command = ' '.join(['restore'] + list(args))
super().call_service(command, **kwargs)
class sirRefresh(GenericExecute):
def __init__(self, connection, context, **kwargs):
super().__init__(connection, context, **kwargs)
self.start_state = 'admin'
self.end_state = 'admin'
def call_service(self, *args, **kwargs):
if 'moff' not in args:
statements = sirStatements()
self.dialog += Dialog([statements.refresh_stmt])
command = ' '.join(['refresh'] + list(args))
super().call_service(command, **kwargs)
class sirReset(GenericReload):
def __init__(self, connection, context, **kwargs):
super().__init__(connection, context, **kwargs)
self.start_state = 'admin'
self.end_state = 'admin'
self.timeout = connection.settings.RELOAD_TIMEOUT
def call_service(self, *args, **kwargs):
if 'moff' in args:
dialog = Dialog([])
else:
statements = sirStatements()
dialog = Dialog([statements.reset_stmt])
command = ' '.join(['reset'] + list(args))
# TODO: ignore unicon.core.errors.TimeoutError
super().call_service(reload_command=command, dialog=dialog, timeout=self.timeout, kwargs=kwargs)
#
# ServiceList, see __init__.py
#
class sirServiceList:
def __init__(self):
self.execute = sirExecute
self.configure = sirConfigure
self.save = sirSave
self.load = sirLoad
self.restore = sirRestore
self.refresh = sirRefresh
self.reset = sirReset
# mixin common services
self.send = svc.Send
self.sendline = svc.Sendline
self.transmit = svc.Send
self.receive = svc.ReceiveService
self.receive_buffer = svc.ReceiveBufferService
self.expect = svc.Expect
self.expect_log = svc.ExpectLogging
self.log_user = svc.LogUser
self.log_file = svc.LogFile
self.ping = svc.Ping
5.setupファイルの作成
最後にこれまで作ったものをインストールするためのスクリプトを作ります。
サンプル(setup.py)
Python(setup.py)
#! /usr/bin/env python
# see https://developer.cisco.com/docs/unicon/
import os
from setuptools import setup, find_packages
def get_readme():
with open(os.path.join(os.path.dirname(__file__), 'README.md')) as f:
return f.read()
setup(
# name of the package
# this name is displayed in 'pip list' output
name='unicon-plugins-sir',
version='1.0',
description='Unicon Plugin for sir',
long_description=get_readme(),
url='',
author='koishi,
author_email='',
license='Apache 2.0',
# see https://pypi.python.org/pypi?%3Aaction=list_classifiers
classifiers=[
'Development Status :: 5 - Production/Stable',
'Environment :: Console',
'Intended Audience :: Developers',
'Intended Audience :: Telecommunications Industry',
'Intended Audience :: Information Technology',
'License :: OSI Approved :: Apache Software License',
'Operating System :: POSIX :: Linux',
'Programming Language :: Python :: 3.4',
'Programming Language :: Python :: 3 :: Only',
'Programming Language :: Python :: Implementation :: CPython',
'Topic :: Software Development :: Testing',
'Topic :: Software Development :: Build Tools',
'Topic :: Software Development :: Libraries',
'Topic :: Software Development :: Libraries :: Python Modules',
],
keywords='pyats unicon plugin connection',
packages=find_packages(where='src'),
package_dir={
'': 'src',
},
package_data={},
# entry_points = {'unicon.plugins': ['<platform_name> = <module_name>']},
#
# module_name is a name of the directory under src
#
# ├── setup.py
# └── src
# ├── sir
#
# in this case module_name is 'sir'
entry_points={'unicon.plugins': ['sir = sir']},
install_requires=['setuptools', 'unicon'],
extras_require={},
data_files=[],
cmdclass={},
zip_safe=False,
)
なお、階層は以下の通りです
# ├── setup.py
# └── src
# ├── sir(ここに1~4までで作ったものをいれます)
6.作ったUnicon Pluginのインストール
インストール方法は2パターンあります。今回は後者のほうでやっています。
作ったプラグインを恒久的にインストールする方法
python3 setup.py install
作ったプラグインは開発用としてインストールされ、src/フォルダ内で修正されたものは即時反映される
python3 setup.py develop
インストールが完了すると、以下のように"pip list"で今回作ったものが見えてきます。
unicon 23.11
unicon.plugins 23.11
unicon-plugins-sir 1.0 /home/koishi/scripts/3rdParty_pyats-modular-test-tool/unicon.plugins/src
これにて、Unicon Pluginの作成と実装は終了です
TextFSMをつくろう
TextFSMはテンプレートでCLI出力等を解析するPythonライブラリです。
社内に共有するため、初心者でも親しみやすいTextFSMにしました! GenieParserを仕立てるには時間もない、技術もない
今回解析するコマンドはこちら"show system information"!!!
ここのFirm Ver. の"04.03"だけ抜き出します!
HOSTNAME# sh system information
Current-time : Wed Dec 20 01:21:42 2023
Startup-time : Mon Dec 4 09:27:01 2023
System : Si-R G110B
Serial No. : SERIAL!!!!!
ROM Ver. : 1.1
Firm Ver. : V04.03 NY0020 Tue Nov 21 10:20:30 JST 2017
Startup-config : Mon Dec 18 22:02:38 2023 config1
Running-config : Mon Dec 4 09:27:01 2023
MAC : MAC!!!!!
Memory : 256MB
USB : ------
そのためのTemplateはこちら
Value FIRM_VERSION (\d+\.\d+)
Start
^.*Firm\sVer.\s:\sV${FIRM_VERSION}
これをテストスクリプトに入れ込みます。(テストするためのスクリプト本体は割愛)
結果
ではチャレンジしてみた結果はこちら!!!
テストOK例
テストNG例
できました~~~
参考
いつもお世話になっている方のリポジトリを参考にさせて頂きました。
サンプルの大部分がこのリポジトリに依存しております。
Unicon Pluginでの3rdparty接続は他に日本語情報が見当たらず、こちらがなかったら、できませんでした。感謝申し上げます
https://github.com/takamitsu-iida/pyats-fitelnet
おわりに
これで他の3rd party機器もPyATSでテストできる可能性を感じました。
マルチベンダーのニーズにも答えられそうです。
今後は以下の4点に取り組めればなーと思っています。
1.スクリプトをきれいにする
2.現場のSEさんが使えるように整える
3.他の3rdPartyをつくる
4.Si-R用のGenieParserを仕立てる
今回の主な動作環境
Si-R G110
WSL1(Ubuntu 22.04.3 LTS)
Python 3.10.12
PyATS/Unicon 23.11
Textfsm 1.1.3
最後までご覧いただきありがとうございました。
筆者はPython/PyATS初心者であり、Si-Rへの接続、showコマンド実行/解析が動いた時点で満足している感があり、理解不足/誤り、お茶を濁した説明がございます。
お気づきの点がある際には、コメント頂けると幸いです。
本投稿において表明される意見は、投稿者本人の個人的意見です。