# snappi extension for Cyperf
[](https://en.wikipedia.org/wiki/MIT_License)
[](https://www.repostatus.org/#active)
[](https://github.com/open-traffic-generator/snappi-cyperf/actions)
[](https://pypi.org/project/snappi_cyperf)
[](https://pypi.python.org/pypi/snappi_cyperf)
[](https://lgtm.com/projects/g/open-traffic-generator/snappi-cyperf/alerts/)
[](https://lgtm.com/projects/g/open-traffic-generator/snappi-cyperf/context:python)
[](https://pepy.tech/project/snappi-cyperf)
This extension allows executing tests written using [snappi](https://github.com/open-traffic-generator/snappi) against
Cyperf, (one of) Keysight's implementation of [Open Traffic Generator](https://github.com/open-traffic-generator/models/releases).
> The repository is under active development.
## Install on a client
```sh
python -m pip install --upgrade "snappi[cyperf]"
```
## Start scripting
```python
import sys
"""
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 Cyperf API Server
api = snappi.api(location="https://localhost:443", ext="cyperf")
config = api.config()
# port location is agent-ip
tx = config.ports.port(name="tx", location="10.39.44.147")[-1]
rx = config.ports.port(name="rx", location="10.39.44.190")[-1]
# configure layer 1 properties
(d1, d2) = config.devices.device(name="d1").device(name="d2")
(e1,) = d1.ethernets.ethernet(name="d1.e1")
e1.connection.port_name = "tx"
e1.mac = "01:02:03:04:05:06"
e1.step = "00:00:00:00:00:01"
e1.count = 1
e1.mtu = 1488
(e2,) = d2.ethernets.ethernet(name="d2.e2")
e2.connection.port_name = "rx"
e2.mac = "01:02:03:04:06:06"
e2.step = "00:00:00:00:00:01"
e2.count = 2
e2.mtu = 1488
(vlan1,) = e1.vlans.vlan(name="vlan1")
vlan1.id = 1
vlan1.priority = 1
vlan1.tpid = "x8100"
vlan1.count = 1
vlan1.step = 1
vlan1.per_count = 1
(vlan2,) = e2.vlans.vlan(name="vlan2")
vlan2.id = 1
vlan2.priority = 1
vlan2.tpid = "x8100"
vlan2.count = 1
vlan2.step = 1
vlan2.per_count = 1
(ip1,) = e1.ipv4_addresses.ipv4(name="e1.ipv4")
ip1.address = "10.0.0.10"
ip1.gateway = "10.0.0.1"
ip1.step = "0.0.0.1"
ip1.count = 1
ip1.prefix = 16
(ip2,) = e2.ipv4_addresses.ipv4(name="e2.ipv4")
ip2.address = "10.0.0.20"
ip2.gateway = "10.0.0.1"
ip2.step = "0.0.0.1"
ip2.count = 1
ip2.prefix = 16
# TCP/UDP configs
(t1,) = d1.tcps.tcp(name="Tcp1")
t1.ip_interface_name = ip1.name
t1.receive_buffer_size = 1111
t1.transmit_buffer_size = 1112
t1.retransmission_minimum_timeout = 100
t1.retransmission_maximum_timeout = 1001
t1.minimum_source_port = 100
t1.maximum_source_port = 101
(t2,) = d2.tcps.tcp(name="Tcp2")
t2.ip_interface_name = ip2.name
t2.receive_buffer_size = 2222
t2.transmit_buffer_size = 2221
t2.retransmission_minimum_timeout = 200
t2.retransmission_maximum_timeout = 2002
t2.minimum_source_port = 200
t2.maximum_source_port = 202
(http_1,) = d1.https.http(name="HTTP1")
http_1.profile = "Chrome"
http_1.version = "HTTP11"
http_1.connection_persistence = "ConnectionPersistenceStandard"
(http_client,) = http_1.clients.client()
http_client.cookie_reject_probability = False
http_client.max_persistent_requests = 1
(http_2,) = d2.https.http(name="HTTP2")
http_2.profile = "Apache"
http_2.version = "HTTP11"
http_2.connection_persistence = "ConnectionPersistenceEnabled"
(http_server,) = http_2.servers.server()
(tp1,) = config.trafficprofile.trafficprofile()
(segment1, segment2) = tp1.segment.segment().segment()
segment1.name = "Linear segment1"
segment1.duration = 40
segment2.name = "Linear segment2"
segment2.duration = 70
tp1.timeline = [segment1.name, segment2.name]
tp1.objective_type = ["Connections per second", "Simulated users"]
tp1.objective_value = [100, 200]
(obj1, obj2) = tp1.objectives.objective().objective()
print("In test before set_config")
response = api.set_config(config)
print("In test after set_config")
print(response)
api.close()
cs = api.control_state()
cs.app.state = "start" # cs.app.state.START
response1 = api.set_control_state(cs)
print(response1)
cs.app.state = "stop" # cs.app.state.START
api.set_control_state(cs)
Raw data
{
"_id": null,
"home_page": "https://github.com/open-traffic-generator/snappi-cyperf",
"name": "snappi-cyperf",
"maintainer": null,
"docs_url": null,
"requires_python": "<4,>=2.7",
"maintainer_email": null,
"keywords": "snappi cyperf testing open traffic generator automation",
"author": "dipendughosh",
"author_email": "dipendu.ghosh@keysight.com",
"download_url": "https://files.pythonhosted.org/packages/59/c6/9000dd66fb540949fe05260129d2005a2cc547e26f0f61a8ef03833ad679/snappi_cyperf-0.0.3.tar.gz",
"platform": null,
"description": "# snappi extension for Cyperf\n\n[](https://en.wikipedia.org/wiki/MIT_License)\n[](https://www.repostatus.org/#active)\n[](https://github.com/open-traffic-generator/snappi-cyperf/actions)\n[](https://pypi.org/project/snappi_cyperf)\n[](https://pypi.python.org/pypi/snappi_cyperf)\n[](https://lgtm.com/projects/g/open-traffic-generator/snappi-cyperf/alerts/)\n[](https://lgtm.com/projects/g/open-traffic-generator/snappi-cyperf/context:python)\n[](https://pepy.tech/project/snappi-cyperf)\n\nThis extension allows executing tests written using [snappi](https://github.com/open-traffic-generator/snappi) against \nCyperf, (one of) Keysight's implementation of [Open Traffic Generator](https://github.com/open-traffic-generator/models/releases).\n\n> The repository is under active development.\n\n## Install on a client\n\n```sh\npython -m pip install --upgrade \"snappi[cyperf]\"\n```\n\n## Start scripting\n\n```python\nimport sys\n\n\"\"\"\nConfigure a raw TCP flow with,\n- tx port as source to rx port as destination\n- frame count 10000, each of size 128 bytes\n- transmit rate of 1000 packets per second\nValidate,\n- frames transmitted and received for configured flow is as expected\n\"\"\"\n\nimport snappi\n\n# host is Cyperf API Server\napi = snappi.api(location=\"https://localhost:443\", ext=\"cyperf\")\nconfig = api.config()\n# port location is agent-ip\ntx = config.ports.port(name=\"tx\", location=\"10.39.44.147\")[-1]\nrx = config.ports.port(name=\"rx\", location=\"10.39.44.190\")[-1]\n# configure layer 1 properties\n(d1, d2) = config.devices.device(name=\"d1\").device(name=\"d2\")\n(e1,) = d1.ethernets.ethernet(name=\"d1.e1\")\ne1.connection.port_name = \"tx\"\ne1.mac = \"01:02:03:04:05:06\"\ne1.step = \"00:00:00:00:00:01\"\ne1.count = 1\ne1.mtu = 1488\n\n(e2,) = d2.ethernets.ethernet(name=\"d2.e2\")\ne2.connection.port_name = \"rx\"\ne2.mac = \"01:02:03:04:06:06\"\ne2.step = \"00:00:00:00:00:01\"\ne2.count = 2\ne2.mtu = 1488\n\n(vlan1,) = e1.vlans.vlan(name=\"vlan1\")\nvlan1.id = 1\nvlan1.priority = 1\nvlan1.tpid = \"x8100\"\nvlan1.count = 1\nvlan1.step = 1\nvlan1.per_count = 1\n\n(vlan2,) = e2.vlans.vlan(name=\"vlan2\")\nvlan2.id = 1\nvlan2.priority = 1\nvlan2.tpid = \"x8100\"\nvlan2.count = 1\nvlan2.step = 1\nvlan2.per_count = 1\n\n(ip1,) = e1.ipv4_addresses.ipv4(name=\"e1.ipv4\")\nip1.address = \"10.0.0.10\"\nip1.gateway = \"10.0.0.1\"\nip1.step = \"0.0.0.1\"\nip1.count = 1\nip1.prefix = 16\n\n(ip2,) = e2.ipv4_addresses.ipv4(name=\"e2.ipv4\")\nip2.address = \"10.0.0.20\"\nip2.gateway = \"10.0.0.1\"\nip2.step = \"0.0.0.1\"\nip2.count = 1\nip2.prefix = 16\n\n# TCP/UDP configs\n\n(t1,) = d1.tcps.tcp(name=\"Tcp1\")\nt1.ip_interface_name = ip1.name\nt1.receive_buffer_size = 1111\nt1.transmit_buffer_size = 1112\nt1.retransmission_minimum_timeout = 100\nt1.retransmission_maximum_timeout = 1001\nt1.minimum_source_port = 100\nt1.maximum_source_port = 101\n\n(t2,) = d2.tcps.tcp(name=\"Tcp2\")\nt2.ip_interface_name = ip2.name\nt2.receive_buffer_size = 2222\nt2.transmit_buffer_size = 2221\nt2.retransmission_minimum_timeout = 200\nt2.retransmission_maximum_timeout = 2002\nt2.minimum_source_port = 200\nt2.maximum_source_port = 202\n\n(http_1,) = d1.https.http(name=\"HTTP1\")\nhttp_1.profile = \"Chrome\"\nhttp_1.version = \"HTTP11\"\nhttp_1.connection_persistence = \"ConnectionPersistenceStandard\"\n(http_client,) = http_1.clients.client()\nhttp_client.cookie_reject_probability = False\nhttp_client.max_persistent_requests = 1\n\n(http_2,) = d2.https.http(name=\"HTTP2\")\nhttp_2.profile = \"Apache\"\nhttp_2.version = \"HTTP11\"\nhttp_2.connection_persistence = \"ConnectionPersistenceEnabled\"\n(http_server,) = http_2.servers.server()\n\n(tp1,) = config.trafficprofile.trafficprofile()\n(segment1, segment2) = tp1.segment.segment().segment()\nsegment1.name = \"Linear segment1\"\nsegment1.duration = 40\nsegment2.name = \"Linear segment2\"\nsegment2.duration = 70\ntp1.timeline = [segment1.name, segment2.name]\ntp1.objective_type = [\"Connections per second\", \"Simulated users\"]\ntp1.objective_value = [100, 200]\n(obj1, obj2) = tp1.objectives.objective().objective()\n\nprint(\"In test before set_config\")\nresponse = api.set_config(config)\nprint(\"In test after set_config\")\nprint(response)\napi.close()\n\ncs = api.control_state()\ncs.app.state = \"start\" # cs.app.state.START\nresponse1 = api.set_control_state(cs)\nprint(response1)\ncs.app.state = \"stop\" # cs.app.state.START\napi.set_control_state(cs)\n\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "The Snappi Cyperf Open Traffic Generator Python Package",
"version": "0.0.3",
"project_urls": {
"Homepage": "https://github.com/open-traffic-generator/snappi-cyperf"
},
"split_keywords": [
"snappi",
"cyperf",
"testing",
"open",
"traffic",
"generator",
"automation"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "b83cc69963c5e712263323b2e1a9d6cccededfbb133daede28704133083a89a5",
"md5": "9f02d94ac4d0f92884395f176fe111a3",
"sha256": "aef71633388d81c355d9640e4891f1cd552618717fdae359c541f3f74e5afd55"
},
"downloads": -1,
"filename": "snappi_cyperf-0.0.3-py2.py3-none-any.whl",
"has_sig": false,
"md5_digest": "9f02d94ac4d0f92884395f176fe111a3",
"packagetype": "bdist_wheel",
"python_version": "py2.py3",
"requires_python": "<4,>=2.7",
"size": 46315,
"upload_time": "2024-08-14T12:44:46",
"upload_time_iso_8601": "2024-08-14T12:44:46.375161Z",
"url": "https://files.pythonhosted.org/packages/b8/3c/c69963c5e712263323b2e1a9d6cccededfbb133daede28704133083a89a5/snappi_cyperf-0.0.3-py2.py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "59c69000dd66fb540949fe05260129d2005a2cc547e26f0f61a8ef03833ad679",
"md5": "8a0baaa82836d714920a7cd7d2238e97",
"sha256": "7e2c7e8e30ccee354b7e6eaaffc6e45c863fd1c3c25d8ea666bc23e7bcad8464"
},
"downloads": -1,
"filename": "snappi_cyperf-0.0.3.tar.gz",
"has_sig": false,
"md5_digest": "8a0baaa82836d714920a7cd7d2238e97",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4,>=2.7",
"size": 41107,
"upload_time": "2024-08-14T12:44:47",
"upload_time_iso_8601": "2024-08-14T12:44:47.508114Z",
"url": "https://files.pythonhosted.org/packages/59/c6/9000dd66fb540949fe05260129d2005a2cc547e26f0f61a8ef03833ad679/snappi_cyperf-0.0.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-08-14 12:44:47",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "open-traffic-generator",
"github_project": "snappi-cyperf",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"requirements": [
{
"name": "mock",
"specs": []
},
{
"name": "ipaddr",
"specs": [
[
"==",
"2.2.0"
]
]
},
{
"name": "netaddr",
"specs": [
[
"==",
"0.8.0"
]
]
},
{
"name": "ipaddress",
"specs": [
[
"==",
"1.0.23"
]
]
},
{
"name": "flake8",
"specs": []
},
{
"name": "dpkt",
"specs": []
},
{
"name": "requests",
"specs": []
},
{
"name": "black",
"specs": []
},
{
"name": "PyYAML",
"specs": []
},
{
"name": "flask",
"specs": []
},
{
"name": "urllib3",
"specs": []
},
{
"name": "simplejson",
"specs": []
},
{
"name": "requests",
"specs": []
},
{
"name": "tabulate",
"specs": []
},
{
"name": "pandas",
"specs": []
},
{
"name": "requests_toolbelt",
"specs": []
},
{
"name": "timer",
"specs": []
}
],
"lcname": "snappi-cyperf"
}