Getting Started

基本的な書方

testbed.yaml

testbed:
    name: IOS-XR_Testbed
    credentials:
        default:
            username: cisco
            password: cisco
        # enable:
        #     password: cisco
devices:
    ios-1: # <----- must match to your device hostname in the prompt
        os: iosxr
        type: iosxr
        connections:
            a:
                protocol: telnet
                ip: 1.1.1.1
                port: 11023
    ios-2:
        os: iosxr
        type: iosxr
        connections:
            a:
                protocol: telnet
                ip: 1.1.1.2
                port: 11024
            vty:
                protocol: ssh
                ip: 5.5.5.5
topology:
    ios-xr1:
        interfaces:
            GigabitEthernet0/0/0/0:
                ipv4: 10.10.10.1/24
                ipv6: '10:10:10::1/64'
                link: link-1
                type: ethernet
            Loopback0:
                ipv4: 192.168.0.1/32
                ipv6: '192::1/128'
                link: xr1_Loopback0
                type: loopback
    ios-xr2:
        interfaces:
            GigabitEthernet0/0/0/0:
                ipv4: 10.10.10.2/24
                ipv6: '10:10:10::2/64'
                link: link-1
                type: ethernet
            Loopback0:
                ipv4: 192.168.0.2/32
                ipv6: '192::2/128'
                link: xr2_Loopback0
                type: loopback

Connect and Issue Commands

# loader our newly minted testbed file
from pyats.topology import loader
testbed = loader.load('testbed.yaml')

# access the devices
testbed.devices
# AttrDict({'ios-1': <Device ott-tb1-n7k4 at 0xf77190cc>,
#           'ios-2': <Device ott-tb1-n7k5 at 0xf744e16c>})
ios_1 = testbed.devices['ios-1']
ios_2 = testbed.devices['ios-2']

# find links from one device to another
for link in ios_1.find_links(ios_2):
    print(repr(link))
# <Link link-1 at 0xf744ef8c>

# establish basic connectivity
ios_1.connect()

# issue commands
print(ios_1.execute('show version'))
ios_1.configure('''
    interface GigabitEthernet0/0
        ip address 10.10.10.1 255.255.255.0
''')

# establish multiple, simultaneous connections
ios_2.connect(alias = 'console', via = 'a')
ios_2.connect(alias = 'vty_1', via = 'vty')

# issue commands through each connection separately
ios_2.vty_1.execute('show running')
ios_2.console.execute('reload')

# creating connection pools
ios_2.start_pool(alias = 'pool', size = 2)

# use connection pool in multiprocessing paradigms
# each process will be allocated a connection - whenever one is available
def sleep(seconds):
    ios_2.pool.execute('sleep %s' % seconds)
import multiprocessing
p1 = multiprocessing.Process(target=sleep, args = (10, ))
p2 = multiprocessing.Process(target=sleep, args = (10, ))
p3 = multiprocessing.Process(target=sleep, args = (10, ))
p1.start(); p2.start(); p3.start()
p1.join(); p2.join(); p3.join()

connectivity_check.py

from pyats import aetest

class CommonSetup(aetest.CommonSetup):

    @aetest.subsection
    def check_topology(self,
                       testbed,
                       ios1_name = 'ios-xr1',
                       ios2_name = 'ios-xr2'):
        ios1_xr1 = testbed.devices[ios1_name]
        ios2_xr2 = testbed.devices[ios2_name]

        # add them to testscript parameters
        self.parent.parameters.update(ios1 = ios_xr1, ios2 = ios_xr2)

        # get corresponding links
        links = ios1.find_links(ios2)

        assert len(links) >= 1, 'require one link between ios1 and ios2'


    @aetest.subsection
    def establish_connections(self, steps, ios1, ios2):
        with steps.start('Connecting to %s' % ios1.name):
            ios1.connect()

        with steps.start('Connecting to %s' % ios2.name):
            ios2.connect()

@aetest.loop(device=('ios1', 'ios2'))
class PingTestcase(aetest.Testcase):

    @aetest.test.loop(destination=('10.10.10.1', '10.10.10.2'))
    def ping(self, device, destination):
        try:
            result = self.parameters[device].ping(destination)

        except Exception as e:
            self.failed('Ping {} from device {} failed with error: {}'.format(
                                destination,
                                device,
                                str(e),
                            ),
                        goto = ['exit'])
        else:
            match = re.search(r'Success rate is (?P<rate>\d+) percent', result)
            success_rate = match.group('rate')

            logger.info('Ping {} with success rate of {}%'.format(
                                        destination,
                                        success_rate,
                                    )
                               )

class CommonCleanup(aetest.CommonCleanup):

    @aetest.subsection
    def disconnect(self, steps, ios1, ios2):
        with steps.start('Disconnecting from %s' % ios1.name):
            ios1.disconnect()

        with steps.start('Disconnecting from %s' % ios2.name):
            ios2.disconnect()

if __name__ == '__main__':
    import argparse
    from pyats.topology import loader

    parser = argparse.ArgumentParser()
    parser.add_argument('--testbed', dest = 'testbed',
                        type = loader.load)

    args, unknown = parser.parse_known_args()

    aetest.main(**vars(args))

Run

python connectivity_check.py --testbed ios_testbed.yaml

job file

job.py

from pyats.easypy import run

def main():
    # run api launches a testscript as an individual task.
    run('connectivity_check.py')

pyats run job ios_job.py --testbed-file ios_testbed.yaml --html-logs

• --html-logsHTMLログファイルの生成

Info

デフォルトでは、ジョブファイルの結果はアーカイブです。 実行時環境、実行内容、結果XMLファイル、ログファイルを記述したファイルを含む zipフォルダーです。