Skip to content

open-traffic-generator/snappi-ixnetwork

Folders and files

NameName
Last commit message
Last commit date

Latest commit

2ab398b · Feb 5, 2025
Feb 5, 2025
Jun 28, 2021
Oct 3, 2021
Jan 29, 2021
Feb 5, 2025
Feb 5, 2025
Nov 24, 2020
Mar 4, 2022
Jan 31, 2022
Jan 28, 2021
Jan 31, 2025
Jun 28, 2021
Jun 28, 2021
Sep 30, 2024
Jan 31, 2025
Jan 30, 2025

Repository files navigation

snappi Extension for IxNetwork

license Project Status: Active - The project has reached a stable, usable state and is being actively developed. Build pypi python Total alerts Language grade: Python downloads

This extension allows executing test scripts written using snappi against
IxNetwork, (one of) Keysight's implementation of Open Traffic Generator.

The repository is under active development.

To start contributing, please see contributing.md.

Install on a client

python -m pip install --upgrade "snappi[ixnetwork]"

Start scripting

"""
Configure a raw TCP flow with,
- tx port as source to rx port as destination
- frame count 10000, each of size 128 bytes
- transmit rate of 1000 packets per second
Validate,
- frames transmitted and received for configured flow is as expected
"""

import snappi
# host is IxNetwork API Server
api = snappi.api(location='https://localhost:443', ext='ixnetwork')
# new config
config = api.config()
# port location is chassis-ip;card-id;port-id
tx, rx = (
    config.ports
    .port(name='tx', location='192.168.0.1;2;1')
    .port(name='rx', location='192.168.0.1;2;2')
)
# configure layer 1 properties
ly, = config.layer1.layer1(name='ly')
ly.port_names = [tx.name, rx.name]
ly.speed = ly.SPEED_10_GBPS
ly.media = ly.FIBER
# configure flow properties
flw, = config.flows.flow(name='flw')
# flow endpoints
flw.tx_rx.port.tx_name = tx.name
flw.tx_rx.port.rx_name = rx.name
# enable flow metrics
flw.metrics.enable = True
# configure rate, size, frame count
flw.size.fixed = 128
flw.rate.pps = 1000
flw.duration.fixed_packets.packets = 10000
# configure protocol headers with defaults fields
flw.packet.ethernet().vlan().ipv4().tcp()
# push configuration
api.set_config(config)
# start transmitting configured flows
control_state = api.control_state()
control_state.choice = control_state.TRAFFIC
control_state.traffic.choice = control_state.traffic.FLOW_TRANSMIT
control_state.traffic.flow_transmit.state = control_state.traffic.flow_transmit.START  # noqa
res = api.set_control_state(control_state)
if len(res.warnings) > 0:
    print("Warnings: {}".format(res.warnings))
# create a query for flow metrics
req = api.metrics_request()
req.flow.flow_names = [flw.name]
# wait for flow metrics to be as expected
while True:
    res = api.get_metrics(req)
    if all([m.frames_tx == 10000 == m.frames_rx for m in res.flow_metrics]):
        break