diff options
Diffstat (limited to 'testsuites/vstf/vstf_scripts/vstf/agent/spirent/spirent.py')
| -rw-r--r-- | testsuites/vstf/vstf_scripts/vstf/agent/spirent/spirent.py | 267 |
1 files changed, 179 insertions, 88 deletions
diff --git a/testsuites/vstf/vstf_scripts/vstf/agent/spirent/spirent.py b/testsuites/vstf/vstf_scripts/vstf/agent/spirent/spirent.py index 904de736..ff2af1f8 100644 --- a/testsuites/vstf/vstf_scripts/vstf/agent/spirent/spirent.py +++ b/testsuites/vstf/vstf_scripts/vstf/agent/spirent/spirent.py @@ -14,23 +14,26 @@ import Tkinter def build_cmd(*args): cmd = '' for arg in args: - cmd = cmd+str(arg)+' ' + cmd = cmd + str(arg) + ' ' #import pdb - #pdb.set_trace() + # pdb.set_trace() return cmd class stcPython(): + def __init__(self): self.tclsh = Tkinter.Tcl() self.stcpkg = '/home/Spirent_TestCenter_4.46/Spirent_TestCenter_Application_Linux' - self.tclsh.eval("set auto_path [ linsert $auto_path 0 %s ]" %(self.stcpkg)) + self.tclsh.eval( + "set auto_path [ linsert $auto_path 0 %s ]" % + (self.stcpkg)) self.tclsh.eval("package require SpirentTestCenter") def build_cmd(self, *args): cmd = '' for arg in args: - cmd = cmd+str(arg)+' ' + cmd = cmd + str(arg) + ' ' return cmd # [ stc base interface ] @@ -38,206 +41,294 @@ class stcPython(): cmd = build_cmd('stc::init', *args) return self.tclsh.eval(cmd) # stc connect - def stc_connect(self,*args): + + def stc_connect(self, *args): cmd = build_cmd('stc::connect', *args) return self.tclsh.eval(cmd) # stc disconnect - def stc_disconnect(self,*args): + + def stc_disconnect(self, *args): cmd = build_cmd('stc::disconnect', *args) return self.tclsh.eval(cmd) # stc create - def stc_create(self,*args): + + def stc_create(self, *args): cmd = build_cmd('stc::create', *args) return self.tclsh.eval(cmd) # stc delete - def stc_delete(self,*args): + + def stc_delete(self, *args): cmd = build_cmd('stc::delete', *args) return self.tclsh.eval(cmd) # stc config - def stc_config(self,*args): + + def stc_config(self, *args): cmd = build_cmd('stc::config', *args) return self.tclsh.eval(cmd) # stc get - def stc_get(self,*args): + + def stc_get(self, *args): cmd = build_cmd('stc::get', *args) return self.tclsh.eval(cmd) # stc apply - def stc_apply(self,*args): + + def stc_apply(self, *args): cmd = build_cmd('stc::apply', *args) return self.tclsh.eval(cmd) # stc perform - def stc_perform(self,*args): + + def stc_perform(self, *args): cmd = build_cmd('stc::perform', *args) return self.tclsh.eval(cmd) # stc reserve - def stc_reserve(self,*args): + + def stc_reserve(self, *args): cmd = build_cmd('stc::reserve', *args) return self.tclsh.eval(cmd) # stc release - def stc_release(self,*args): + + def stc_release(self, *args): cmd = build_cmd('stc::release', *args) return self.tclsh.eval(cmd) # stc subscribe - def stc_subscribe(self,*args): - cmd = build_cmd('stc::subscribe',*args) + + def stc_subscribe(self, *args): + cmd = build_cmd('stc::subscribe', *args) return self.tclsh.eval(cmd) # stc unsubscribe - def stc_unsubscribe(self,*args): + + def stc_unsubscribe(self, *args): cmd = build_cmd('stc::unsubscribe', *args) return self.tclsh.eval(cmd) # stc wait until sequencer complete - def stc_waituntilcomplete(self,*args): + + def stc_waituntilcomplete(self, *args): cmd = build_cmd('stc::waituntilcomplete', *args) return self.tclsh.eval(cmd) # stc help + def stc_help(self, *args): - cmd = build_cmd('stc::help',*args) + cmd = build_cmd('stc::help', *args) return self.tclsh.eval(cmd) # [ stc expand interface ] # get one dict-key's value # return value - def stc_get_value(self,stc_dict,stc_key): - cmd = stc_dict+' -'+stc_key + def stc_get_value(self, stc_dict, stc_key): + cmd = stc_dict + ' -' + stc_key return self.stc_get(cmd) # create project # return: project_name + def stc_create_project(self): return self.stc_create('project') # create port under project # return: port name - def stc_create_port(self,project_name): - cmd = 'port -under '+project_name + + def stc_create_port(self, project_name): + cmd = 'port -under ' + project_name return self.stc_create(cmd) # config port location # return: None - def stc_config_port_location(self,port_name,chassisAddress,slot,port): + + def stc_config_port_location(self, port_name, chassisAddress, slot, port): #import pdb - #pdb.set_trace() - cmd = port_name+' -location //'+chassisAddress+'/'+slot+'/'+port+' -UseDefaultHost False' + # pdb.set_trace() + cmd = port_name + ' -location //' + chassisAddress + \ + '/' + slot + '/' + port + ' -UseDefaultHost False' return self.stc_config(cmd) # create streamblock under port # return: streamblock name - def stc_create_streamblock(self,port_name,vlan_tag,ExpectedRxPort,srcMac,dstMac,sourceAddr,destAddr): + + def stc_create_streamblock( + self, + port_name, + vlan_tag, + ExpectedRxPort, + srcMac, + dstMac, + sourceAddr, + destAddr): #import pdb - #pdb.set_trace() - if vlan_tag == None or vlan_tag == 'None': + # pdb.set_trace() + if vlan_tag is None or vlan_tag == 'None': frameStruc = '"EthernetII IPv4 Udp"' - if ExpectedRxPort == '' : - return self.stc_create( 'streamBlock -under ',port_name, - '-frameConfig ',frameStruc, - '-frame "EthernetII.srcMac',srcMac,'EthernetII.dstMac',dstMac, - 'IPv4.1.sourceAddr',sourceAddr,'IPv4.1.destAddr',destAddr,'"') - else : - return self.stc_create( 'streamBlock -under ',port_name, - '-ExpectedRxPort',ExpectedRxPort, - '-frameConfig ',frameStruc, - '-frame "EthernetII.srcMac',srcMac,'EthernetII.dstMac',dstMac, - 'IPv4.1.sourceAddr',sourceAddr,'IPv4.1.destAddr',destAddr,'"') - else : + if ExpectedRxPort == '': + return self.stc_create( + 'streamBlock -under ', + port_name, + '-frameConfig ', + frameStruc, + '-frame "EthernetII.srcMac', + srcMac, + 'EthernetII.dstMac', + dstMac, + 'IPv4.1.sourceAddr', + sourceAddr, + 'IPv4.1.destAddr', + destAddr, + '"') + else: + return self.stc_create( + 'streamBlock -under ', + port_name, + '-ExpectedRxPort', + ExpectedRxPort, + '-frameConfig ', + frameStruc, + '-frame "EthernetII.srcMac', + srcMac, + 'EthernetII.dstMac', + dstMac, + 'IPv4.1.sourceAddr', + sourceAddr, + 'IPv4.1.destAddr', + destAddr, + '"') + else: frameStruc = '"EthernetII Vlan IPv4 Udp"' - if ExpectedRxPort == '' : - return self.stc_create( 'streamBlock -under ',port_name, - '-frameConfig '+frameStruc, - '-frame "EthernetII.srcMac',srcMac,'EthernetII.dstMac',dstMac, - 'Vlan.1.id',vlan_tag, - 'IPv4.1.sourceAddr',sourceAddr,'IPv4.1.destAddr',destAddr,'"') - else : - return self.stc_create( 'streamBlock -under ',port_name, - '-ExpectedRxPort',ExpectedRxPort, - '-frameConfig '+frameStruc, - '-frame "EthernetII.srcMac',srcMac,'EthernetII.dstMac',dstMac, - 'Vlan.1.id',vlan_tag, - 'IPv4.1.sourceAddr',sourceAddr,'IPv4.1.destAddr',destAddr,'"') + if ExpectedRxPort == '': + return self.stc_create( + 'streamBlock -under ', + port_name, + '-frameConfig ' + + frameStruc, + '-frame "EthernetII.srcMac', + srcMac, + 'EthernetII.dstMac', + dstMac, + 'Vlan.1.id', + vlan_tag, + 'IPv4.1.sourceAddr', + sourceAddr, + 'IPv4.1.destAddr', + destAddr, + '"') + else: + return self.stc_create( + 'streamBlock -under ', + port_name, + '-ExpectedRxPort', + ExpectedRxPort, + '-frameConfig ' + + frameStruc, + '-frame "EthernetII.srcMac', + srcMac, + 'EthernetII.dstMac', + dstMac, + 'Vlan.1.id', + vlan_tag, + 'IPv4.1.sourceAddr', + sourceAddr, + 'IPv4.1.destAddr', + destAddr, + '"') # config streamblock with part arguments # argument list use args dictionary - def stc_config_streamblock(self,streamblock_name,args_dict): + + def stc_config_streamblock(self, streamblock_name, args_dict): cmd = '' - for key in args_dict.keys() : - temp_cmd = '-'+key+' '+str(args_dict[key]) + for key in args_dict.keys(): + temp_cmd = '-' + key + ' ' + str(args_dict[key]) cmd = cmd + temp_cmd - return self.stc_config(streamblock_name,cmd) + return self.stc_config(streamblock_name, cmd) # get generator name from port name # return: generator name - def stc_get_generator(self,port_name): - cmd = port_name+' -children-generator' + + def stc_get_generator(self, port_name): + cmd = port_name + ' -children-generator' return self.stc_get(cmd) # config generator with part arguments # argument list use args dictionary # return none - def stc_config_generator(self,generator_name,args_dict): + + def stc_config_generator(self, generator_name, args_dict): cmd = '' - for key in args_dict.keys() : - temp_cmd = '-'+key+' '+str(args_dict[key]) + for key in args_dict.keys(): + temp_cmd = '-' + key + ' ' + str(args_dict[key]) cmd = cmd + temp_cmd - return self.stc_config(generator_name,cmd) + return self.stc_config(generator_name, cmd) # attach port # return: port's parent project info - def stc_attach_ports(self,portList): + + def stc_attach_ports(self, portList): cmd = 'AttachPorts -portList {' - for port in portList : - cmd = cmd+' '+port - cmd = cmd+'} -autoConnect TRUE' + for port in portList: + cmd = cmd + ' ' + port + cmd = cmd + '} -autoConnect TRUE' return self.stc_perform(cmd) # config src mac and dst mac # return: none - def stc_config_ethII(self,ethII,src_mac,dst_mac): - cmd = ethII+' -srcMac '+src_mac+' -dstMac '+dst_mac + + def stc_config_ethII(self, ethII, src_mac, dst_mac): + cmd = ethII + ' -srcMac ' + src_mac + ' -dstMac ' + dst_mac return self.stc_config(cmd) # config src ip and dst ip # return: none - def stc_config_ethIII(self,ethIII,src_ip,dst_ip): - cmd = ethIII+' -sourceAddr '+src_ip+' -destAddr '+dst_ip + + def stc_config_ethIII(self, ethIII, src_ip, dst_ip): + cmd = ethIII + ' -sourceAddr ' + src_ip + ' -destAddr ' + dst_ip return self.stc_config(cmd) # start streamblock # return: none - def stc_streamblock_start(self,streamblock_list): + + def stc_streamblock_start(self, streamblock_list): cmd = 'StreamBlockStart -StreamBlockList {' - for streamblock in streamblock_list : - cmd = cmd+' '+streamblock - cmd = cmd+' } -ExecuteSynchronous TRUE' + for streamblock in streamblock_list: + cmd = cmd + ' ' + streamblock + cmd = cmd + ' } -ExecuteSynchronous TRUE' return self.stc_perform(cmd) # stop streamblock - def stc_streamblock_stop(self,streamblock_list): + + def stc_streamblock_stop(self, streamblock_list): cmd = 'StreamBlockStop -StreamBlockList {' - for streamblock in streamblock_list : - cmd = cmd+' '+streamblock - cmd = cmd+' } -ExecuteSynchronous TRUE' + for streamblock in streamblock_list: + cmd = cmd + ' ' + streamblock + cmd = cmd + ' } -ExecuteSynchronous TRUE' return self.stc_perform(cmd) # start generator # return: none - def stc_generator_start(self,generator_List): + + def stc_generator_start(self, generator_List): cmd = 'GeneratorStart -generatorList {' - for generator in generator_List : - cmd = cmd+' '+generator - cmd = cmd+' }' + for generator in generator_List: + cmd = cmd + ' ' + generator + cmd = cmd + ' }' return self.stc_perform(cmd) # stop generator # return: none - def stc_generator_stop(self,generator_List): + + def stc_generator_stop(self, generator_List): cmd = 'GeneratorStop -generatorList {' - for generator in generator_List : - cmd = cmd+' '+generator - cmd = cmd+' }' + for generator in generator_List: + cmd = cmd + ' ' + generator + cmd = cmd + ' }' return self.stc_perform(cmd) # create rfc2544 throughput test + def stc_setup_rfc2544_throughput(self): pass # create rfc2544 frameloss test + def stc_setup_rfc2544_frameloss(self): pass # create rfc2544 latency test + def stc_setup_rfc2544_latency(self): pass # start Sequence start + def stc_sequence_start(self): return self.stc_perform('SequencerStart') # output rfc2544 throughput result + def stc_get_rfc2544_throughput_result(self): pass # output rfc2544 frameloss result + def stc_get_rfc2544_frameloss_result(self): pass # output rfc2544 latency result + def stc_get_rfc2544_latency_result(self): pass |