memo

from pyats import aetest
from pyats.topology import loader

# テストケースを作成
class MyTestcase(aetest.Testcase):
    @aetest.setup
    def setup(self, testbed):
        # テストベッドの読み込み
        tb = loader.load(testbed)

        # デバイスへの接続
        self.parameters["devices"] = {}  # devicesキーを追加

        for device_name, device in tb.devices.items():
            device.connect(learn_hostname=True)
            self.parameters["devices"][device_name] = device

    @aetest.test
    def show_run_test(self):
        # デバイスの情報を表示
        for device_name, device in self.parameters["devices"].items():
            # show runコマンドを実行
            output = device.execute("show run")

            # 結果を表示
            print(f"--- Configuration on {device_name} ---")
            print(output)
            print("-----------------------------------")

# テストケースを実行
if __name__ == "__main__":
    # テストベッドファイルのパスを指定して実行
    aetest.main(testbed="testbed.yaml")

testbed.yaml

devices:
  Router1:
    alias: R1
    os: "iosxr"
    credentials:
      default:
        username: cisco
        password: cisco
    connections:
      cli:
        protocol: telnet
        ip: "192.168.11.56"
        arguments:
          init_config_commands: []
            # - line default exec-timeout 0 0
            # - line default absolute-timeout 0
            # - line default session-timeout 0

ログ追加

from pyats import aetest
from pyats.topology import loader
import logging

# ロガーを設定
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")

# ファイルハンドラを追加
file_handler = logging.FileHandler("log.txt")
file_handler.setLevel(logging.INFO)
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)

# テストケースを作成
class MyTestcase(aetest.Testcase):
    @aetest.setup
    def setup(self, testbed):
        # テストベッドの読み込み
        tb = loader.load(testbed)

        # デバイスへの接続
        self.parameters["devices"] = {}  # devicesキーを追加

        for device_name, device in tb.devices.items():
            device.connect(learn_hostname=True)
            self.parameters["devices"][device_name] = device

    @aetest.test
    def show_run_test(self):
        # デバイスの情報をログに記録
        for device_name, device in self.parameters["devices"].items():
            # show runコマンドを実行
            output = device.execute("show run")

            # ログに記録
            logger.info(f"--- Configuration on {device_name} ---")
            logger.info(output)
            logger.info("-----------------------------------")

# テストケースを実行
if __name__ == "__main__":
    # テストベッドファイルのパスを指定して実行
    aetest.main(testbed="testbed.yaml")

マルチプロセス

from pyats import aetest
from pyats.topology import loader
import multiprocessing

# show runを実行する関数
def execute_show_run(device_name, device):
    output = device.execute("show run")

    # 結果を表示
    print(f"--- Configuration on {device_name} ---")
    print(output)
    print("-----------------------------------")

# テストケースを作成
class MyTestcase(aetest.Testcase):
    @aetest.setup
    def setup(self, testbed):
        # テストベッドの読み込み
        tb = loader.load(testbed)

        # デバイスへの接続
        self.parameters["devices"] = {}  # devicesキーを追加

        for device_name, device in tb.devices.items():
            device.connect(learn_hostname=True)
            self.parameters["devices"][device_name] = device

    @aetest.test
    def show_run_test(self):
        # プロセスリストを作成
        processes = []

        # デバイスごとにプロセスを生成して実行
        for device_name, device in self.parameters["devices"].items():
            process = multiprocessing.Process(target=execute_show_run, args=(device_name, device))
            processes.append(process)
            process.start()

        # 全てのプロセスの終了を待機
        for process in processes:
            process.join()

# テストケースを実行
if __name__ == "__main__":
    # テストベッドファイルのパスを指定して実行
    aetest.main(testbed="testbed.yaml")