PyATSとSi-R

この記事は富士通の有志によるFUJITSU Advent Calendar 2023の20日目の投稿です。
https://qiita.com/advent-calendar/2023/fujitsu

# はじめに
はじめまして、koishiと申します。
とある富士通のグループ会社でネットワークまわりの技術支援とかとかのお仕事をさせて頂いております。



# この記事はなに?
PyATSで富士通ルータであるSi-Rをテストをしようとちょっと頑張った記録です。
どなたかのお役に立てれば幸いです。

:::note info
[PyATS(**Py**thon **A**utomated **T**est **S**ystems)](https://developer.cisco.com/docs/pyats-getting-started/)とは
**Cisco社が開発しているPythonベースのテスト自動化ソリューションとなります。**
読み方は"ぱいえーてぃーえす"とのこと
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/185259/baf47daa-45f2-57e3-1b4a-05b47a4c5ef6.png)

:::

## ネットワーク屋のテストについて
ネットワーク屋のテストは複雑な経路のテストとかもありますが、それだけでなくバージョン番号をチェックするだけの単純な確認テストをネットワーク機器の数だけしないといけないということもあります(3桁4桁の台数あるときも)

**あるプロジェクトでPyATSのトライしてみたところ、とあるテストの工数が200分の1になったとかならないとか。**





# なんで書いた
**そんなPyATSで富士通製NW機器のテストできたらいいのにな・・・**

そう言われたのは、現場SEさんにPyATSのご紹介を終えたあとの一言でした。
前述の通り、PyATSは便利なソリューションではありますが、
<font color="red">**残念なことに富士通製のネットワーク機器は対応しておりません。:sob:**</font>


そのため幾度となく"富士通製は対応してないんですよね。。。"とお茶を濁して、ごめんなさいをしておりました。

しかしながら、アドベントカレンダーのネタに困っていたので、やってみました。



# PyATSで3rd party機器(Si-R)のテスト実現に必要なこと
この企画を実現するには
1.**Uniconライブラリ** と
2.**Genieパーサー** 
に手を加える必要があるとわかりました。

:::note info
**Uniconライブラリ**はNW機器に接続するためのコネクションライブラリで、接続のタイミングで必要なコマンドとかうまいことしてくれるものです。
**Genieパーサー**はNW機器から取得したshowコマンド等の出力をうまいことJSONにパーッスしてくれるものです。
※[Uniconライブラリ](https://pubhub.devnetcloud.com/media/unicon/docs/user_guide/supported_platforms.html)と[Genieパーサー](https://pubhub.devnetcloud.com/media/genie-feature-browser/docs/#/parsers)のサポート状況はリンクの通りです。(前述のとおり富士通製の機器は対応していません)
:::


様々なアプローチがあるとは思いますが、今回はそれぞれ以下の対応としました。
**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)

<details><summary>サンプル(__init__.py)</summary>

```Python:__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()

```
</details>

また初期にページャーを停止したいのでsettings.pyに記載しています。
エラーパターンはお借りしたサンプルスクリプトからSi-R用にアップデートできていません。。。

<details><summary>サンプル(settings.py)</summary>

```Python: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

```

</details>

## 2.Connection Providerの作成
Connection ProviderはNW機器の接続プロセスを定義するものです。
サンプルではsingleRP機器の接続プロセスをベースとしています。


<details><summary>サンプル(provider.py)</summary>

```Python: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

```
</details>

<details><summary>サンプル(statements.py)</summary>

```Python: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)
```
</details>


## 3.StateMachineの作成
StateMachineはNW機器の状態および各状態のプロンプト定義するものです。Cisco機器の特権モード、コンフィグレーションモードと同じようにSi-Rもモードがありますので、それを定義するものです。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/185259/8ffdcb06-a66f-e1db-1347-b7dbfa13849b.png)
[Si-R G100B/G110B/G200B コマンドユーザーズガイドより引用](https://www.fujitsu.com/jp/documents/products/network/router/manual/sir-g/cmd_users-g100b_g110b_g200b.pdf)

まずは状態を定義します。
以下のサンプルでは、showコマンドのみチャレンジしたかったため、運用管理モードの一般ユーザクラス(disable)から管理者クラス(enable)に相当する箇所を読み替えたものとなっております。

<details><summary>サンプル(statemachine.py)</summary>

```Python: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)
```

</details>

次は各状態におけるプロンプトを定義し、プロンプトから現在の状態を判別します。
showコマンドのみチャレンジのため、運用管理モードの一般ユーザークラス、管理者クラスに該当するもののみ読み替えてアップデートしており、それ以外はお借りしたサンプルスクリプトからアップデートできていない箇所があります。

<details><summary>サンプル(patterns.py)</summary>

```Python: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: *?'

```

</details>

## 4.Service定義の作成
ここでは、コマンドの実行や設定等、PyATSが提供するサービスについて定義をします。今回はshowコマンドの実行のみのため、下記サンプルでは不要なサービスも残ってしまっていますが、ご容赦ください。

<details><summary>サンプル(services.py)</summary>

```Python: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
```

</details>

## 5.setupファイルの作成
最後にこれまで作ったものをインストールするためのスクリプトを作ります。


<details><summary>サンプル(setup.py)</summary>

```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,
)
```
</details>

なお、階層は以下の通りです
```
    # ├── setup.py
    # └── src
    #     ├── sir(ここに1~4までで作ったものをいれます)
```


## 6.作ったUnicon Pluginのインストール
インストール方法は2パターンあります。今回は後者のほうでやっています。
```bash:作ったプラグインを恒久的にインストールする方法
python3 setup.py install
```

```bash:作ったプラグインは開発用としてインストールされ、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例
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/185259/6aab652c-064c-e96d-e592-8f46b2390038.png)
テストNG例
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/185259/e08b00ea-82e2-cd01-c06b-78083eedaef2.png)

できました~~~


# 参考
いつもお世話になっている方のリポジトリを参考にさせて頂きました。
サンプルの大部分がこのリポジトリに依存しております。
Unicon Pluginでの3rdparty接続は他に日本語情報が見当たらず、こちらがなかったら、できませんでした。感謝申し上げます:pray:
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コマンド実行/解析が動いた時点で満足している感があり、理解不足/誤り、お茶を濁した説明がございます。
お気づきの点がある際には、コメント頂けると幸いです。

本投稿において表明される意見は、投稿者本人の個人的意見です。