实现Media config的切换，使得Loki-100G-5S-2P测试板卡可以链接在50GbE模式下进行流量测试

实现Media config的切换，使得测试板卡那个链接在50GbE模式下进行流量测试

main.py

# coding=UTF-8
import time, json, random, queue, types, sys, socket, math, os
from binascii import hexlify
from TestUtilsL23 import XenaScriptTools#########################################
##############Descripting################
## Aurthor: Zohn Yang                  ##
## Date : 2019-12-9                    ##
## Email : szy@xenanetworks.com        ##
## Software : Pyhton 3.7.4             ##
#########################################def runtest(xm, ports, rate, size, duration):##Print messagemodule = str(ports[0])[0:1]print("Start the scripting...")##RELEASE THE ports and modules, relinquish other users.xm.Send(module + ' M_RESERVATION RELEASE')xm.Send(module + ' M_RESERVATION relinquish')xm.Send(ports[0] + ' P_RESERVATION RELEASE')xm.Send(ports[0] + ' P_RESERVATION relinquish')xm.Send(ports[1] + ' P_RESERVATION RELEASE')xm.Send(ports[1] + ' P_RESERVATION relinquish')##Reserve the modulexm.Send(module + ' M_RESERVATION RESERVE')##set port speed mode into 4*50Gprint("Set the mode into 50G...")#xm.SendExpectOK(module + ' M_CFPCONFIGEXT 2 100000 100000')#xm.SendExpectOK(module + ' M_CFPCONFIGEXT 8 25000 25000 25000 25000 25000 25000 25000 25000')xm.SendExpectOK(module + ' M_CFPCONFIGEXT 4 50000 50000 50000 50000')time.sleep(5)xm.Send(module + ' M_RESERVATION RELEASE')print("Successful...")##Release the port and then Reserve the portsprint ("Resever the port...")xm.Send(ports[0] + ' P_RESERVATION RESERVE')  xm.Send(ports[1] + ' P_RESERVATION RESERVE')##RESET PORTSxm.Send(ports[0] + ' P_RESET')xm.Send(ports[1] + ' P_RESET')time.sleep(1)MAC1= '000000000002'MAC2= '000000000001'IP1 = '192.168.100.100'IP2 = '192.168.100.101'IP1 = hexlify(socket.inet_aton(IP1)).decode()IP2 = hexlify(socket.inet_aton(IP2)).decode()hearder1 = '0x' + str(MAC2) + str(MAC1) + '08004500002E000000007FFFF0B6' + str(IP1) + str(IP2) hearder2 = '0x' + str(MAC1) + str(MAC2) + '08004500002E000000007FFFF0B6' + str(IP2) + str(IP1) print("Start to configure the streams...")##Create the streams in port 0##Create the SID index of streamxm.SendExpectOK(ports[0] + " PS_CREATE [0]")##Create the TPLD index of streamxm.SendExpectOK(ports[0] + " PS_TPLDID [0] 0")##Configure the packet sizexm.SendExpectOK(ports[0] + " PS_PACKETLENGTH [0]" + size)##Configure the packet typexm.SendExpectOK(ports[0] + " PS_HEADERPROTOCOL [0] ETHERNET IP")##Configure the packet headerxm.SendExpectOK(ports[0] + " PS_PACKETHEADER [0] "+ str(hearder1))##Enable streamsxm.SendExpectOK(ports[0] + " PS_ENABLE [0] on")##Configure the stream ratexm.SendExpectOK(ports[0] + " PS_RATEFRACTION [0] " + str(int(rate)*10000))##Create the streams in port 1xm.SendExpectOK(ports[1] + " PS_CREATE [1]")xm.SendExpectOK(ports[1] + " PS_TPLDID [1] 1")xm.SendExpectOK(ports[1] + " PS_PACKETLENGTH [1]" + size)xm.SendExpectOK(ports[1] + " PS_HEADERPROTOCOL [1] ETHERNET IP")xm.SendExpectOK(ports[1] + " PS_PACKETHEADER [1] "+ str(hearder2))xm.SendExpectOK(ports[1] + " PS_ENABLE [1] on")xm.SendExpectOK(ports[1] + " PS_RATEFRACTION [1] " + str(int(rate)*10000))print("Start the traffic...")#####START TRAFFICxm.SendExpectOK(ports[0] + ' P_TRAFFIC ON')xm.SendExpectOK(ports[1] + ' P_TRAFFIC ON')time.sleep(int(duration))xm.SendExpectOK(ports[0] + ' P_TRAFFIC OFF')xm.SendExpectOK(ports[1] + ' P_TRAFFIC OFF')print("Stop the traffic and collect the result...")xm.Send(ports[0] + ' P_RESERVATION RELEASE')xm.Send(ports[1] + ' P_RESERVATION RELEASE')time.sleep(2)####Get traffic result##Get the TX and RX resultTX1 = (filter(xm, ports[0] + ' PT_STREAM [0] ?'))[3]TX2 = (filter(xm, ports[1] + ' PT_STREAM [1] ?'))[3]RX1 = (filter(xm, ports[1] + ' PR_TPLDTRAFFIC [0] ?'))[3]RX2 = (filter(xm, ports[0] + ' PR_TPLDTRAFFIC [1] ?'))[3]##Get the latency latency1 = (filter(xm, ports[1] + ' PR_TPLDLATENCY [0] ?'))[1]latency2 = (filter(xm, ports[0] + ' PR_TPLDLATENCY [1] ?'))[1]##Get the errorerror1 = filter(xm, ports[1] + ' PR_TPLDERRORS[1] ?')error2 = filter(xm, ports[0] + ' PR_TPLDERRORS[0] ?')##Get the FCSFCS1 = (filter(xm, ports[0] + ' PR_EXTRA ?'))[0]FCS2 = (filter(xm, ports[1] + ' PR_EXTRA ?'))[0]##Caculate the lostLost1 = int(TX1) - int(RX1)Lost2 = int(TX2) - int(RX2)##print the resultprint()print ('----------------------------------------------------------------------------------------------------')print ('Stream1 |  TX: ' + str(TX1) + '  |  RX: ' + str(RX1) + '  |  Lost :' + str(Lost1) + ' |  FCS: ' + str(FCS2) + '  |  Misoder Error: ' + error1[2] + '  |  Payload Errors: ' + error1[3])print ('----------------------------------------------------------------------------------------------------')print ('Stream2 |  TX: ' + str(TX2) + '  |  RX: ' + str(RX2) + '  |  Lost :' + str(Lost2) + ' |  FCS: ' + str(FCS1) + ' |  Misoder Error: ' + error2[2] + '  |  Payload Errors: ' + error2[3])print ('----------------------------------------------------------------------------------------------------')print ('Ending.......')def filter(xm, cmd):getvalue = xm.Send(cmd)getvalue = getvalue.split('  ')[-1]getvalue = getvalue.split(' ')return getvaluedef main(argv):with open('config.txt','r+') as f:configs = f.readlines()config_dict = {}for conf in configs:parsed = conf.strip('\n').split(':')if len(parsed) > 1:config_dict[parsed[0]] = parsed[1]ip_address = config_dict.get('ip_address')print(ip_address)ports = config_dict.get('ports').split(' ')print(ports)rate = config_dict.get('rate')size = config_dict.get('size')duration = config_dict.get('duration')xm = XenaScriptTools(ip_address)print('Start to connect to the chassis...')xm.LogonSetOwner("xena", "python_test_1")print('Logon successful...')runtest(xm, ports, rate, size, duration)if __name__ == '__main__':sys.exit(main(sys.argv))

SockeDrives.py

import socket
import sysclass SimpleSocket(object):def __init__(self, hostname, port = 22611, timeout = 20):hostname = hostname.encode()self.hostname = hostnametry:self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)except socket.error as msg:sys.stderr.write("[Socket connection error] Cannot connect to %s, error: %s\n" % (hostname, msg[0]))sys.exit(1)self.sock.settimeout(timeout)try:self.sock.connect((hostname, port))except socket.error as msg:sys.stderr.write("[Socket connection error] Cannot connect to %s, error: %s\n" % (hostname, msg[0]))sys.exit(2)def __del__(self):self.sock.close()def SendCommand(self, cmd):cmd = cmd.encode()self.sock.send(cmd + b'\n')def Ask(self, cmd):cmd = cmd.encode()self.sock.send(cmd + b'\n')return self.sock.recv(2048)def set_keepalives(self):self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 2)

TestUtisL23.py

import os, sys, time, threading, inspect, typesfrom SocketDrivers import SimpleSocketLOGFILE = "XENALOG"RESET           = " p_reset"
RESERVATION     = " p_reservation ?"
RESERVE         = " p_reservation reserve"
RELINQUISH      = " p_reservation relinquish"
RELEASE         = " p_reservation release"
TRAFFIC_ON      = " p_traffic on"
TRAFFIC_OFF     = " p_traffic off"
TCLEAR          = " pt_clear"
RCLEAR          = " pr_clear"
DELETE          = " ps_delete"
LIMIT           = " P_TXTIMELIMIT"COMMENT_START   = ';'def errexit(msg):#print (str(msg) + ", exiting...]")sys.exit()## Keepalive thread to ensure tcp connection is kept open
# do not edit this
class KeepAliveThread(threading.Thread):message = ''def __init__(self, connection, interval = 10):threading.Thread.__init__(self)self.connection = connectionself.interval = intervalself.finished = threading.Event()self.setDaemon(True)print ('[KeepAliveThread] Thread initiated, interval %d seconds' % (self.interval))def stop (self):self.finished.set()self.join()def run (self):while not self.finished.isSet():self.finished.wait(self.interval)self.connection.Ask(self.message)## Low level driver for TCPIP based queried
# do not edit this
class XenaSocketDriver(SimpleSocket):def __init__(self, hostname, port = 22611):SimpleSocket.__init__(self, hostname = hostname, port = port)SimpleSocket.set_keepalives(self)self.access_semaphor = threading.Semaphore(1)def SendCommand(self, cmd):self.access_semaphor.acquire()SimpleSocket.SendCommand(self, cmd)self.access_semaphor.release()def Ask(self, cmd):self.access_semaphor.acquire()reply = SimpleSocket.Ask(self, cmd).strip(b'\n')self.access_semaphor.release()return reply## Xena supplied class example for Scripting via Python
# feel free to add functions below
#
class XenaScriptTools:def __init__(self, ip):self.ip    = ipself.debug = 0self.halt  = 1 self.log   = 0self.cmds  = []self.logf  = os.environ.get(LOGFILE)if self.logf != None:self.log = 1self.driver= XenaSocketDriver(ip)def __del__(self):if self.log:lf = open(log.txt, 'w')for cmd in self.cmds:lf.write(cmd + "\n")lf.close()return## Enable debug - prints commands and errorsdef debugOn(self):self.debug = 1return ## Disable debug (default) - no printed outputdef debugOff(self):self.debug = 0return def debugMsg(self, msg):if self.debug == 1:print ("message")print (msg)def logCmd(self, cmd):if self.log == 1:self.cmds.append(cmd)## Enable halt on error - calls sys.exit(1) upon errordef haltOn(self):self.halt = 1return ## Disable halt on error (default)def haltOff(self):self.halt = 0return ## Print diagnostics msg and haltdef errexit(self, msg):if self.halt == 1:#print#print (msg + ", exiting...]")printsys.exit(1)###############################################
## Send and Expect primitives
################################################# Send command and return responsedef Send(self, cmd):res = self.driver.Ask(cmd)self.debugMsg("Send()         : " + cmd)res = res.decode()self.debugMsg("Send() received: " + res)self.logCmd(cmd)return res## Send command and expect response (typically <OK>)def SendExpect(self, cmd, resp):self.debugMsg("SendExpect("+resp+"): " + cmd)self.logCmd(cmd)res = self.driver.Ask(cmd).decode()if res.rstrip('\n') == resp:return True;else:self.debugMsg("SendExpect() failed")self.debugMsg("   Expected: " + resp)if type(res) != type(str()):res = res.decode()self.debugMsg("   Received: " + res)self.errexit("Test result :Failed;[Send:" + str(cmd) + "; Received: " + str(res))return Falsedef SendExpectSync(self, cmd, resp):self.debugMsg("SendExpect("+resp+"): " + cmd)self.logCmd(cmd)res = self.driver.Ask(cmd).decode()sync = res.rstrip('\n').split('  ')[-1]if sync == resp:return True;else:self.debugMsg("SendExpect() failed")self.debugMsg("   Expected: " + resp)self.debugMsg("   Received: " + res)self.errexit("Test result :Failed;[Port " + str(res))return False## Send commands and expect <OK>def SendExpectOK(self, cmd):return self.SendExpect(cmd, "<OK>")def SendExpectSyncOK(self, cmd):return self.SendExpectSync(cmd, "IN_SYNC")## Send command and match response with specified stringdef SendAndMatch(self, cmd, str):self.debugMsg("SendAndMatch() : " + cmd)self.logCmd(cmd)res = self.driver.Ask(cmd)if res.find(str) != -1:return Trueelse:self.debugMsg("SendAndMatch() failed")self.debugMsg("   Expected: " + str)self.debugMsg("   Got     : " + res)print ("Test result :Failed;[测试异常，请重新测试]"),self.errexit("Halting in line %d" % (inspect.currentframe().f_back.f_lineno))return False###############################################
## Xena Scripting API specific commands
############################################################################## Chassis and Logon Commands ################################ Logondef Logon(self, pwd):self.SendExpectOK("c_logon \"" + pwd  + "\"")## Logon and set ownerdef LogonSetOwner(self, pwd, owner):self.Logon(pwd)self.SendExpectOK("c_owner \"" + owner + "\"")## Logon to chassis, set user name and password, then reserve portsdef LogonAndReserve(self, ports, pwd, owner):if type(ports) == type(str()):ports = [ports]self.LogonSetOwner(pwd, owner)self.PortReserve(ports)# Reserve chassis, release or relinquish first if necessarydef ChassisReserve(self):self.ChassisRelease()self.SendExpectOK("C_RESERVATION reserve")# Reserve chassis, release or relinquish first if necessarydef ChassisRelease(self):res = self.Send("C_RESERVATION ?").split()[1]if res.find("RESERVED_BY_YOU") != -1:self.debugMsg("Chassis is reserved by me - release")self.SendExpectOK("C_RESERVATION release")elif res.find("RESERVED_BY_OTHER") != -1:self.debugMsg("Chassis is reserved by other - relinquish")self.SendExpectOK("C_RESERVATION relinquish")elif res.find("RELEASED") != -1:self.debugMsg("Chassis is released - do nothing")else:self.errexit("Halting in line %d" % (inspect.currentframe().f_back.f_lineno))############################### Misc Commands##############################def Comment(self, text):self.Send("; ######################################")text = text.decode()self.Send("; " + text)self.Send("; ######################################")############################### Port Commands################################ Reserve a port - if port is reserved, release or relinquish, then reservedef PortReserve(self, ports):if type(ports) == type(str()):ports = [ports]for port in ports:res = self.Send(port + RESERVATION)if res.find("RESERVED_BY_OTHER") != -1:self.debugMsg("Port " + port + " is reserved by other - relinquish")self.SendExpectOK(port + RELINQUISH)self.SendExpectOK(port + RESERVE)elif res.find("RESERVED_BY_YOU") != -1:self.debugMsg("Port " + port + " is reserved by me - do nothing")else:self.SendExpectOK(port + RESERVE)def PortRelease(self, ports):if type(ports) == type(str()):ports = [ports]for port in ports:res = self.Send(port + RESERVATION)if res.find("RESERVED_BY_YOU") != -1:self.SendExpectOK(port + RELEASE)def PortRelinquish(self, ports):if type(ports) == type(str()):ports = [ports]for port in ports:res = self.Send(port + RESERVATION)if res.find("RESERVED_BY_OTHER") != -1:self.Send(port + RELINQUISH)def PortSyncCheck(self, ports):if type(ports) == type(str()):ports = [ports]for port in ports:res = self.SendExpectSyncOK(port + " P_RECEIVESYNC ?")## Start traffic on ports def PortTrafficStart(self, ports):if type(ports) == type(str()):ports = [ports]for port in ports:res = self.SendExpectOK(port + TRAFFIC_ON)##Start learning mac addressdef Statisticsclear(self, ports):if type(ports) == type(str()):ports = [ports]for port in ports:res = self.SendExpectOK(port + TCLEAR)res = self.SendExpectOK(port + RCLEAR)## Stop traffic on ports def PortTrafficStop(self, ports):if type(ports) == type(str()):ports = [ports]for port in ports:res = self.SendExpectOK(port + TRAFFIC_OFF)## Reset ports def Portreset(self, ports):if type(ports) == type(str()):ports = [ports]for port in ports:res = self.SendExpectOK(port + RESET)def get_module_port_prefix(self, moduleIndex, portIndex):return "%d/%d" % (moduleIndex, portIndex)def load_script(self, filename, moduleIndex, portIndex, israwlines=False):module_port_prefix = self.get_module_port_prefix(moduleIndex, portIndex)self.PortReserve(module_port_prefix)if not israwlines:self.driver.SendCommand(module_port_prefix)line_number = 0;send_count = 0;for line in open(filename, 'r'):command = line.strip('\n')line_number += 1if command.startswith(COMMENT_START):continuesuccess = self.SendExpectOK(command.strip('\n'))if not success:print ('[XenaManager] Error in script at line: %d, [%s]' % (line_number, command))print ('[XenaManager] Load interrupted!')returnsend_count += 1if send_count % 100 == 0:print ("\r[XenaManager] (Sent %d commands ...)" % send_count),print ("\r[XenaManager] Script '%s' (%d commands) loaded succesfully." % (filename, send_count))##Streams configurationdef StreamsDelete(self, ports):if type(ports) == type(str()):ports = [ports]for port in ports:i = 0for i in range(24):res = self.Send(port + DELETE + '[' + str(i) + ']')i += 1def Porttimelimit(self, ports, packettime):if type(ports) == type(str()):ports = [ports]for port in ports:res = self.Send(port + " P_TXTIMELIMIT " + packettime + "000000")

config.json

ip_address:176.22.65.118
ports:7/0 7/2
rate:100
size:FIXED 64 1518
duration:20

运行结果

Microsoft Windows [版本 10.0.17763.914]
(c) 2018 Microsoft Corporation。保留所有权利。D:\Xena_Python\50G Media type>python main.py config.txt
176.22.65.118
['7/0', '7/2']
Start to connect to the chassis...
Logon successful...
Start the scripting...
Set the mode into 50G...
Successful...
Resever the port...
Start to configure the streams...
Start the traffic...
Stop the traffic and collect the result...----------------------------------------------------------------------------------------------------
Stream1 |  TX: 1531076544  |  RX: 1531076544  |  Lost :0 |  FCS: 0  |  Misoder Error: 0  |  Payload Errors: 0
----------------------------------------------------------------------------------------------------
Stream2 |  TX: 1528691904  |  RX: 1528691904  |  Lost :0 |  FCS: 0 |  Misoder Error: 0  |  Payload Errors: 0
----------------------------------------------------------------------------------------------------
Ending.......

实现Media config的切换，使得Loki-100G-5S-2P测试板卡可以链接在50GbE模式下进行流量测试相关推荐

命令行界面命令模式及相互切换、交换机命令行操作模式及模式间的切换过程、命令行界面基本功能、命令的快捷键功能、交换机基本配置命令、交换机特权模式下基本命令、交换机全局配置模式下基本命令、
1.命令行界面命令模式及相互切换锐捷交换机命令行管理界面分成若干不同的模式,用户当前所处的命令模式决定了可以使用的命令,不可跨模式执行命令.以下给出三种基本模式. 用户模式(User EXE ...
MySQL MGR 单主模式下单点故障时的节点角色切换规则
MGR单主模式下,有一个节点可读可写,其余节点都是只读,其中表现为super_read_only被自动设为了ON. 那么,如果可读可写的节点异常宕机了,会进行怎样的切换呢? 在选择新的可写角色时,主要 ...
Zynq-7000 AMP运行模式下的软件切换及多版本固化运行
Zynq平台AMP运行模式下的软件切换及多版本固化运行 1. Zynq双裸核AMP运行环境构建 1 1.1 Zynq架构概要 1 1.2 Zynq启动与配置 2 1.2.1 启动流程概述 3 1.2. ...
android otg主从切换,一种自动切换OTG设备主从模式下的Vbus电压的方法与电路与流程...
本发明涉及otg设备的检测领域,更具体地说,涉及一种自动切换otg设备主从模式下的vbus电压的方法与电路. 背景技术: 现有技术中,具备otg功能的设备通过检测usb_id引脚的状态来判断工作为ho ...
非抢占模式下故障后优先级降低却不切换问题--使用Keepalived实现非抢占模式主备切换
使用Keepalived实现非抢占模式主备切换–解释非抢占模式下故障后优先级降低却不切换问题背景最近在用keepalived时发现,配置为非抢占模式后,主节点的服务检测脚本返回非0后,不发生切换, ...
自动undo表空间模式下切换新的undo表空间
在undo自动表空间模式下切换新的undo表空间后都需要将老的undo表空间给drop,可是由于切换时由于活动事物并未完全结束, 于是导正drop旧的undo表空间时报"ORA- ...
T400/500支持集成显卡/独立显卡切换，但在集成显卡模式下不支持2504扩展坞的DVI输出，必须切换到独显模式。
T400/500支持集成显卡/独立显卡切换,但在集成显卡模式下不支持2504扩展坞的DVI输出,必须切换到独显模式. 另如果在扩展坞上接了PS2鼠标,小红点将无法使用.接PS2键盘不冲突.
android 默认静音模式设置,Android：如何在振动设置之间进行切换（始终，永不，仅在静音模式下，仅在非静音模式下）？再访...
寻找方法来切换之间:从不模式模式始终仅在静音仅在非静音这些选择是通过路径找到的 - - 菜单>>设置>>声音>>振动---在手机上. 它是简单的导航手机上的 ...
win10系统cmd模式下，如何完美地切换到其它盘
写在这里的初衷,一是备忘,二是希望得到高人指点,三是希望能遇到志同道合的朋友. 目录 1.快捷打开cmd:win+r 2.通过指令,切换到其它盘 3.输入cd 目录名,可以直接到目录下 4.cd .. ...

实现Media config的切换，使得Loki-100G-5S-2P测试板卡可以链接在50GbE模式下进行流量测试

实现Media config的切换，使得Loki-100G-5S-2P测试板卡可以链接在50GbE模式下进行流量测试相关推荐

最新文章

热门文章