信雅纳网络测试的二次开发集成：XOA（Xena Open-Source Automation）开源自动化测试

目录

XOA是什么

XOA CLI

XOA Python API

​XOA Python Test Suite/测试套件

XOA Converter

Source Code

XOA是什么

XOA（Xena Open-Source Automation）是一个开源的测试自动化框架，追求“高效、易用、灵活”的跨操作系统的开发框架。能与Xena现有解决方案无缝配合，借助XOA可调用Xena（Z系列打流仪、E系列损伤仪）完成自动化测试任务。同时它提供了操作接口可将其他仪表/设备做并栈集成测试验证，统一整理输出测试报告。

XOA包含：
XOA CLI、XOA Python API、XOA Python TestSuite、XOA Converter

信雅纳：网络测试自动化

XOA CLI

XOA CLI 提供了一套简洁直观的基于文本语言的独立命令，用以控制和集成 Xena测试仪硬件，实现各种测试任务的自动化。
任何客户端平台/编程语言（如 Python、Tcl、Bash）都可与 XOA CLI 配合使用。
CLI 可在远程登录终端上使用，直接向 Xena 测试仪发送命令。
ValkyrieManager通过测试端口配置文件（.xpc ），可在 XOA CLI 环境之间无缝转换。

XOA Python API

XOA Python API与XOA CLI和XenaManager无缝集成

  • 面向对象的高级抽象： XOA Python API采用面向对象的方法，提供了更高层次的抽象，加快了自动化脚本的开发。
  • 集成开发环境自动完成，内置手册： XOA Python API 包含 IDE 自动完成以及类、函数和 API 内置手册等功能，可显著提高开发效率。
  • 命令分组和响应自动匹配：该功能允许命令分组和响应自动匹配，从而优化了测试执行效率。
  • 服务器到客户端推送通知订阅： XOA Python API 支持服务器到客户端的推送通知订阅，降低了用户代码的复杂性。
  • 支持 Python 3.8 及更高版本： XOA Python API 兼容 Python 3.8 及更高版本，确保与现代 Python 环境兼容。 
    import asyncio
from contextlib import suppress
from xoa_driver import testers
from xoa_driver import modules
from xoa_driver import ports
from xoa_driver import utils
from xoa_driver import enums
from xoa_driver import exceptions
from ipaddress import IPv4Address, IPv6Address
from binascii import hexlify
from xoa_driver.misc import Hex#---------------------------
# Global parameters
#---------------------------CHASSIS_IP = "10.165.16.70"      # Chassis IP address or hostname
USERNAME = "XOA"                # Username
MODULE_INDEX = 4                # Module index
TX_PORT_INDEX = 0               # TX Port indexFRAME_SIZE_BYTES = 4178         # Frame size on wire including the FCS.
FRAME_COUNT = 20              # The number of frames including the first, the middle, and the last.
REPETITION = 1                  # The number of repetitions of the frame sequence, set to 0 if you want the port to repeat over and over
TRAFFIC_RATE_FPS = 100          # Traffic rate in frames per second
TRAFFIC_RATE_PERCENT = int(4/10 * 1000000)SHOULD_BURST = False            # Whether the middle frames should be bursty
BURST_SIZE_FRAMES = 9           # Burst size in frames for the middle frames
INTER_BURST_GAP_BYTES = 3000    # The inter-burst gap in bytes
INTRA_BURST_GAP_BYTES = 1000    # The inter-frame gap within a burst, aka. intra-burst gap, in bytes#---------------------------
# Header content for streams
#---------------------------
class Ethernet:def __init__(self):self.dst_mac = "0000.0000.0000"self.src_mac = "0000.0000.0000"self.ethertype = "86DD"def __str__(self):_dst_mac = self.dst_mac.replace(".", "")_src_mac = self.src_mac.replace(".", "")_ethertype = self.ethertypereturn f"{_dst_mac}{_src_mac}{_ethertype}".upper()class IPV4:def __init__(self):self.version = 4self.header_length = 5self.dscp = 0self.ecn = 0self.total_length = 42self.identification = "0000"self.flags = 0self.offset = 0self.ttl = 255self.proto = 255self.checksum = "0000"self.src = "0.0.0.0"self.dst = "0.0.0.0"def __str__(self):_ver = '{:01X}'.format(self.version)_header_length = '{:01X}'.format(self.header_length)_dscp_ecn = '{:02X}'.format((self.dscp<<2)+self.ecn)_total_len = '{:04X}'.format(self.total_length)_ident = self.identification_flag_offset = '{:04X}'.format((self.flags<<13)+self.offset)_ttl = '{:02X}'.format(self.ttl)_proto = '{:02X}'.format(self.proto)_check = self.checksum_src = hexlify(IPv4Address(self.src).packed).decode()_dst = hexlify(IPv4Address(self.dst).packed).decode()return f"{_ver}{_header_length}{_dscp_ecn}{_total_len}{_ident}{_flag_offset}{_ttl}{_proto}{_check}{_src}{_dst}".upper()class IPV6:def __init__(self):self.version = 6self.traff_class = 8self.flow_label = 0self.payload_length = 0self.next_header = "11"self.hop_limit = 1self.src = "2000::2"self.dst = "2000::100"def __str__(self):_ver = '{:01X}'.format(self.version)_traff_class = '{:01X}'.format(self.traff_class)_flow_label = '{:06X}'.format(self.flow_label)_payload_len = '{:04X}'.format(self.payload_length)_next_header = self.next_header_hop_limit = '{:02X}'.format(self.hop_limit)_src = hexlify(IPv6Address(self.src).packed).decode()_dst = hexlify(IPv6Address(self.dst).packed).decode()return f"{_ver}{_traff_class}{_flow_label}{_payload_len}{_next_header}{_hop_limit}{_src}{_dst}".upper()class UDP:def __init__(self):self.src_port = 0self.dst_port = 0self.length = 0self.checksum = 0def __str__(self):_src_port = '{:04X}'.format(self.src_port)_dst_port = '{:04X}'.format(self.dst_port)_length = '{:04X}'.format(self.length)_checksum = '{:04X}'.format(self.checksum)return f"{_src_port}{_dst_port}{_length}{_checksum}".upper()class ROCEV2:def __init__(self):self.opcode = 0self.solicited_event = 0self.mig_req = 0self.pad_count = 1self.header_version = 0self.partition_key = 65535self.reserved = 7self.dest_queue_pair = 2self.ack_request = 0self.reserved_7bits = 0self.packet_seq_number =0def __str__(self):_opcode = '{:02X}'.format(self.opcode)_combo_1 = '{:02X}'.format((self.solicited_event<<7)+(self.mig_req<<6)+(self.pad_count<<4)+self.header_version)_pk = '{:04X}'.format(self.partition_key)_reserved = '{:02X}'.format(self.reserved)_qp = '{:06X}'.format(self.dest_queue_pair)_combo_2 = '{:02X}'.format((self.ack_request<<7)+self.reserved_7bits)_ps = '{:06X}'.format(self.packet_seq_number)return f"{_opcode}{_combo_1}{_pk}{_reserved}{_qp}{_combo_2}{_ps}".upper()#------------------------------
# def my_awesome_func()
#------------------------------
async def my_awesome_func(stop_event: asyncio.Event, should_burst: bool) -> None:"""This Python function uses XOA Python API to configure the TX port:param stop_event::type stop_event: asyncio.Event:param should_burst: Whether the middle frames should be bursty.:type should_burst: bool"""# create tester instance and establish connectiontester = await testers.L23Tester(CHASSIS_IP, USERNAME, enable_logging=False) # access the module on the testermodule = tester.modules.obtain(MODULE_INDEX)# check if the module is of type Loki-100G-5S-2Pif not isinstance(module, modules.ModuleChimera):# access the txport on the moduletxport = module.ports.obtain(TX_PORT_INDEX)#---------------------------# Port reservation#---------------------------print(f"#---------------------------")print(f"# Port reservation")print(f"#---------------------------")if txport.is_released():print(f"The txport is released (not owned by anyone). Will reserve the txport to continue txport configuration.")await txport.reservation.set_reserve() # set reservation , means txport will be controlled by our sessionelif not txport.is_reserved_by_me():print(f"The txport is reserved by others. Will relinquish and reserve the txport to continue txport configuration.")await txport.reservation.set_relinquish() # send relinquish the txportawait txport.reservation.set_reserve() # set reservation , means txport will be controlled by our session#---------------------------# Start port configuration#---------------------------print(f"#---------------------------")print(f"# Start port configuration")print(f"#---------------------------")print(f"Reset the txport")await txport.reset.set()print(f"Configure the txport")await utils.apply(# txport.speed.mode.selection.set(mode=enums.PortSpeedMode.F100G),txport.comment.set(comment="RoCE2 on Loki"),txport.tx_config.enable.set_on(),txport.latency_config.offset.set(offset=0),txport.latency_config.mode.set(mode=enums.LatencyMode.LAST2LAST),txport.tx_config.burst_period.set(burst_period=0),txport.tx_config.packet_limit.set(packet_count_limit=FRAME_COUNT*REPETITION),txport.max_header_length.set(max_header_length=128),txport.autotrain.set(interval=0),txport.loop_back.set_none(),                                # If you want loopback the port TX to its own RX, change it to set_txoff2rx()txport.checksum.set(offset=0),txport.tx_config.delay.set(delay_val=0),txport.tpld_mode.set_normal(),txport.payload_mode.set_normal(),#txport.rate.pps.set(port_rate_pps=TRAFFIC_RATE_FPS),       # If you want to control traffic rate with FPS, uncomment this.txport.rate.fraction.set(TRAFFIC_RATE_PERCENT),                          # If you want to control traffic rate with fraction, uncomment this. 1,000,000 = 100%)if should_burst:await txport.tx_config.mode.set_burst()else:await txport.tx_config.mode.set_sequential()#--------------------------------------# Configure stream_0 on the txport#--------------------------------------print(f"   Configure first-packet stream on the txport")stream_0 = await txport.streams.create()eth = Ethernet()eth.src_mac = "aaaa.aaaa.0005"eth.dst_mac = "bbbb.bbbb.0005"ipv4 = IPV4()ipv4.src = "1.1.1.5"ipv4.dst = "2.2.2.5"ipv6 = IPV6()ipv6.src = "2001::5"ipv6.dst = "2002::5"udp = UDP()udp.src_port = 4791udp.dst_port = 4791rocev2 = ROCEV2()rocev2.opcode = 0rocev2.dest_queue_pair = 2rocev2.packet_seq_number = 0await utils.apply(stream_0.enable.set_on(),stream_0.packet.limit.set(packet_count=1),stream_0.comment.set(f"First packet"),stream_0.rate.fraction.set(stream_rate_ppm=10000),stream_0.packet.header.protocol.set(segments=[enums.ProtocolOption.ETHERNET,enums.ProtocolOption.IPV6,enums.ProtocolOption.UDP,enums.ProtocolOption.RAW_12,]),stream_0.packet.header.data.set(hex_data=Hex(str(eth)+str(ipv6)+str(udp)+str(rocev2))),stream_0.packet.length.set(length_type=enums.LengthType.FIXED, min_val=FRAME_SIZE_BYTES, max_val=FRAME_SIZE_BYTES),stream_0.payload.content.set(payload_type=enums.PayloadType.PATTERN, hex_data=Hex("AABBCCDD")),stream_0.tpld_id.set(test_payload_identifier = 0),stream_0.insert_packets_checksum.set_on())if should_burst:await stream_0.burst.burstiness.set(size=1, density=100)await stream_0.burst.gap.set(inter_packet_gap=0, inter_burst_gap=0)#--------------------------------------# Configure stream_1 on the txport#--------------------------------------print(f"   Configure middle-packets stream on the txport")stream_1 = await txport.streams.create()rocev2.opcode = 1rocev2.dest_queue_pair = 2rocev2.packet_seq_number = 1await utils.apply(stream_1.enable.set_on(),stream_1.packet.limit.set(packet_count=FRAME_COUNT-2),stream_1.comment.set(f"Middle packets"),stream_1.rate.fraction.set(stream_rate_ppm=10000),stream_1.packet.header.protocol.set(segments=[enums.ProtocolOption.ETHERNET,enums.ProtocolOption.IPV6,enums.ProtocolOption.UDP,enums.ProtocolOption.RAW_12,]),stream_1.packet.header.data.set(hex_data=Hex(str(eth)+str(ipv6)+str(udp)+str(rocev2))),stream_1.packet.length.set(length_type=enums.LengthType.FIXED, min_val=FRAME_SIZE_BYTES, max_val=FRAME_SIZE_BYTES),stream_1.payload.content.set(payload_type=enums.PayloadType.PATTERN, hex_data=Hex("AABBCCDD")),stream_1.tpld_id.set(test_payload_identifier = 1),stream_1.insert_packets_checksum.set_on())if should_burst:await stream_1.burst.burstiness.set(size=BURST_SIZE_FRAMES, density=100)await stream_1.burst.gap.set(inter_packet_gap=INTRA_BURST_GAP_BYTES, inter_burst_gap=INTER_BURST_GAP_BYTES)# Configure a modifier on the stream_1await stream_1.packet.header.modifiers.configure(1)# Modifier on the SQNmodifier = stream_1.packet.header.modifiers.obtain(0)await modifier.specification.set(position=72, mask="FFFF0000", action=enums.ModifierAction.INC, repetition=1)await modifier.range.set(min_val=1, step=1, max_val=FRAME_COUNT-2)#--------------------------------------# Configure stream_2 on the txport#--------------------------------------print(f"   Configure last-packet stream on the txport")stream_2 = await txport.streams.create()rocev2.opcode = 2rocev2.dest_queue_pair = 2rocev2.packet_seq_number = FRAME_COUNT-1await utils.apply(stream_2.enable.set_on(),stream_2.packet.limit.set(packet_count=1),stream_2.comment.set(f"Last packet"),stream_2.rate.fraction.set(stream_rate_ppm=10000),stream_2.packet.header.protocol.set(segments=[enums.ProtocolOption.ETHERNET,enums.ProtocolOption.IPV6,enums.ProtocolOption.UDP,enums.ProtocolOption.RAW_12,]),stream_2.packet.header.data.set(hex_data=Hex(str(eth)+str(ipv6)+str(udp)+str(rocev2))),stream_2.packet.length.set(length_type=enums.LengthType.FIXED, min_val=FRAME_SIZE_BYTES, max_val=FRAME_SIZE_BYTES),stream_2.payload.content.set(payload_type=enums.PayloadType.PATTERN, hex_data=Hex("AABBCCDD")),stream_2.tpld_id.set(test_payload_identifier = 2),stream_2.insert_packets_checksum.set_on())if should_burst:await stream_2.burst.burstiness.set(size=1, density=100)await stream_2.burst.gap.set(inter_packet_gap=0, inter_burst_gap=0)async def main():stop_event =asyncio.Event()try:await my_awesome_func(stop_event, should_burst=SHOULD_BURST)except KeyboardInterrupt:stop_event.set()if __name__=="__main__":asyncio.run(main())

信雅纳：Python XOA测试自动化
 

XOA Python Test Suite/测试套件

XOA Python 测试套件是一个测试框架，为开发人员和测试专家执行和集成 Xena 测试套件提供了定义明确的 API。
该框架以自动化方式处理各种任务，如测试资源管理、测试执行和发布测试结果。
每个 RFC 测试套件都被设计成独立的 "插件"，可根据需要有选择性地集成到项目中。
目前，XOA Python 测试套件包括
- RFC2544
- RFC2889
- RFC3918

XOA Converter

如果您希望将当前的 Xena 测试套件配置快速迁移到 XOA，现在使用 XOA 转换器工具比以往任何时候都更容易。

以前，Xena的测试套件应用程序仅与Windows兼容。但今后，所有现有和未来的测试套件都将并入 XOA Python 测试套件，从而消除 Windows 限制。

为了简化过渡，我们推出了 XOA 转换器。该工具允许用户将现有的Xena测试套件配置（Xena2544、Xena2889和Xena3918）从Xena窗口桌面应用程序无缝迁移到XOA Python测试套件中。有了 XOA 转换器，迁移过程变得轻松简单。

信雅纳：网络测试仪自动化测试

Source Code

GitHub 是我们托管 XOA 源代码的首选平台，因为它具有出色的版本控制和协作能力。它为管理代码变更提供了一个极佳的环境，确保项目的历史记录完备且易于访问。我们崇尚开放，鼓励每个人使用、分享、贡献和反馈我们的源代码。GitHub 允许进行无缝协作，并促进以社区为导向的方法，让每个人都能积极参与 XOA 的开发和改进。我们重视来自社区的意见和贡献，因为这能提高源代码的整体质量和创新性。

  • XOA Python API Source Code
  • XOA Python Test Suite – Core Source Code
  • XOA Python Test Suite – Plugin Source Code
  • XOA ANLT Utility Source Code
  • XOA Converter Source Code

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/541325.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

细说C++反向迭代器：原理与用法

细说C++反向迭代器：原理与用法

文章目录 一、引言二、反向迭代器的原理与实现细节三、模拟实现C反向迭代器反向迭代器模板类的设计反向迭代器的使用示例与测试 一、引言 迭代器与反向迭代器的概念引入 迭代器&#xff08;Iterator&#xff09;是C标准模板库&#xff08;STL&#xff09;中的一个核心概念&am…
阅读更多...
3.15作业

3.15作业

1、思维导图 2、模拟面试题 1&#xff09;什么是IP地址&#xff1f; 答&#xff1a;ip地址是主机在网络中的唯一标识&#xff0c;分为IPv4和IPv6&#xff0c;IP网络号主机号 2&#xff09;IP地址和MAC地址的区别 答&#xff1a;IP地址是由数字和点分十进制表示的主机在网络中…
阅读更多...
VS2022 配置QT5.9.9

VS2022 配置QT5.9.9

QT安装 下载地址：https://download.qt.io/archive/qt/ 下载安装后进行配置 无法运行 rc.exe 下载VS2022 官网下载 配置 1.扩展-管理扩展-下载Qt Visual Studio Tools 安装 2.安装完成后，打开vs2022,点击扩展，会发现多出了QT VS Tools,点击它，选择Qt Versions并进行配置…
阅读更多...
Java开发从入门到精通（七）：Java的面向对象编程OOP：语法、原理、this、构造器

Java开发从入门到精通（七）：Java的面向对象编程OOP：语法、原理、this、构造器

Java大数据开发和安全开发 &#xff08;一&#xff09;Java的面向对象编程1.1 什么是面向对象1.2 面向对象和面向过程的区别1.3 面向对象开发设计特征1.4 面向对象语法1.4.1 先创建对象模板1.4.2 实例化对象1.4.3 对象又该怎么理解?1.4.4 对象在计算机中的执行原理 1.5 类和对…
阅读更多...
python：消息推送 - 飞书机器人推送 - 富文本格式

python：消息推送 - 飞书机器人推送 - 富文本格式

简介&#xff1a;机器人 ( bot ) 是一种自动化的程序&#xff0c;可以用拟人化的身份自动推送消息&#xff0c;或在聊天里与你进行简单的交互。在自动化完成测试任务后&#xff0c;推送测试报告等是一种很常用的收尾工具。 历史攻略&#xff1a; python&#xff1a;消息推送 …
阅读更多...
搭建知识管理系统并不复杂，这篇教程来帮你

搭建知识管理系统并不复杂，这篇教程来帮你

许多人都有这样的体验&#xff1a;我们抓住的想法和知识总在不经意间溜走&#xff0c;我们想要的信息总是一时无法找到。因此&#xff0c;搭建一个能够系统化、分类和索引存储这些知识的“知识管理系统”是必要的。听上去很专业&#xff0c;其实并不复杂&#xff0c;让我们一步…
阅读更多...
单目测距+姿态识别+yolov8界面+车辆行人跟踪计数

单目测距+姿态识别+yolov8界面+车辆行人跟踪计数

yolov5单目测距速度测量目标跟踪&#xff08;算法介绍和代码&#xff09; 1.单目测距实现方法 在目标检测的基础上&#xff0c;我们可以通过计算物体在图像中的像素大小来估计其距离。具体方法是&#xff0c;首先确定某个物体的实际尺寸&#xff0c;然后根据该物体在图像中的像…
阅读更多...
可视化展示与交互编辑：探索3D Web轻量化平台HOOPS WEB Platform在BIM中的新可能性

可视化展示与交互编辑：探索3D Web轻量化平台HOOPS WEB Platform在BIM中的新可能性

随着数字技术的飞速发展&#xff0c;建筑行业也在不断迈向数字化转型的道路。在这个过程中&#xff0c;BIM&#xff08;Building Information Modeling&#xff0c;建筑信息模型&#xff09;技术已经成为建筑设计、施工和管理领域中的一项重要工具。 而在BIM的应用中&#xff…
阅读更多...
TikTok禁令引发数据安全和控制问题

TikTok禁令引发数据安全和控制问题

darkreading网站消息&#xff0c;继众议院能源与商务委员会上周通过禁用流行社交媒体平台TikTok的法案后&#xff0c;美国国会于周三投票赞成该法案。该法案规定&#xff0c;任何受“外国”控股的企业需在180天内撤资。 长期以来&#xff0c;TikTok作为字节跳动的子公司&#x…
阅读更多...
C#判断素数的方法：试除法 vs 优化的试除法 vs 米勒-拉宾素数检测算法

C#判断素数的方法：试除法 vs 优化的试除法 vs 米勒-拉宾素数检测算法

目录 1.素数也就质数 2. 试除法 3.优化的试除法_1 4.优化的试除法_2 5.优化的试除法_3 6.米勒-拉宾素数检测算法 1.素数也叫质数 一个质数是一个大于1的自然数&#xff0c;只有两个正因数&#xff1a;1和它自身。这意味着如果一个数只有两个正因数&#xff0c;那么它就是…
阅读更多...
Google Play上架：谷歌支付政策变更

Google Play上架：谷歌支付政策变更

目录 政策发布时间概括内容付款政策变动内容归纳google付款用户政策政策发布时间 2024 年 3 月 6 日 概括内容 为遵守《数字市场法案》(DMA) 的规定，从 2024 年 3 月 6 日起，我们将更新付款政策，允许开发者将欧洲经济区 (EEA) 的用户引导至其应用之外的平台（包括推广应用…
阅读更多...
Swift：.ignoresSafeArea()：自由布局的全方位掌握

Swift：.ignoresSafeArea()：自由布局的全方位掌握

ignoresSafeArea(_ regions : edges:)修饰符的说明 SwiftUI布局系统会调整视图的尺寸和位置&#xff0c;以避免特定的安全区域。这就确保了系统内容&#xff08;比如软件键盘&#xff09;或设备边缘不会遮挡您的视图。要将您的内容扩展到这些区域&#xff0c;您可以通过应用该修…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023