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 | 334 |
1 files changed, 0 insertions, 334 deletions
diff --git a/testsuites/vstf/vstf_scripts/vstf/agent/spirent/spirent.py b/testsuites/vstf/vstf_scripts/vstf/agent/spirent/spirent.py deleted file mode 100644 index ff2af1f8..00000000 --- a/testsuites/vstf/vstf_scripts/vstf/agent/spirent/spirent.py +++ /dev/null @@ -1,334 +0,0 @@ -############################################################################## -# Copyright (c) 2015 Huawei Technologies Co.,Ltd and others. -# -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# http://www.apache.org/licenses/LICENSE-2.0 -############################################################################## - - -import Tkinter - - -def build_cmd(*args): - cmd = '' - for arg in args: - cmd = cmd + str(arg) + ' ' - #import pdb - # 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("package require SpirentTestCenter") - - def build_cmd(self, *args): - cmd = '' - for arg in args: - cmd = cmd + str(arg) + ' ' - return cmd - - # [ stc base interface ] - def stc_init(self, *args): - cmd = build_cmd('stc::init', *args) - return self.tclsh.eval(cmd) - # stc connect - - def stc_connect(self, *args): - cmd = build_cmd('stc::connect', *args) - return self.tclsh.eval(cmd) - # stc disconnect - - def stc_disconnect(self, *args): - cmd = build_cmd('stc::disconnect', *args) - return self.tclsh.eval(cmd) - # stc create - - def stc_create(self, *args): - cmd = build_cmd('stc::create', *args) - return self.tclsh.eval(cmd) - # stc delete - - def stc_delete(self, *args): - cmd = build_cmd('stc::delete', *args) - return self.tclsh.eval(cmd) - # stc config - - def stc_config(self, *args): - cmd = build_cmd('stc::config', *args) - return self.tclsh.eval(cmd) - # stc get - - def stc_get(self, *args): - cmd = build_cmd('stc::get', *args) - return self.tclsh.eval(cmd) - # stc apply - - def stc_apply(self, *args): - cmd = build_cmd('stc::apply', *args) - return self.tclsh.eval(cmd) - # stc perform - - def stc_perform(self, *args): - cmd = build_cmd('stc::perform', *args) - return self.tclsh.eval(cmd) - # stc reserve - - def stc_reserve(self, *args): - cmd = build_cmd('stc::reserve', *args) - return self.tclsh.eval(cmd) - # stc release - - 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) - return self.tclsh.eval(cmd) - # stc unsubscribe - - 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): - cmd = build_cmd('stc::waituntilcomplete', *args) - return self.tclsh.eval(cmd) - # stc help - - def stc_help(self, *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 - 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 - return self.stc_create(cmd) - # config port location - # return: None - - 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' - 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): - #import pdb - # 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: - 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, - '"') - # config streamblock with part arguments - # argument list use args dictionary - - def stc_config_streamblock(self, streamblock_name, args_dict): - cmd = '' - for key in args_dict.keys(): - temp_cmd = '-' + key + ' ' + str(args_dict[key]) - cmd = cmd + temp_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' - 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): - cmd = '' - for key in args_dict.keys(): - temp_cmd = '-' + key + ' ' + str(args_dict[key]) - cmd = cmd + temp_cmd - return self.stc_config(generator_name, cmd) - # attach port - # return: port's parent project info - - def stc_attach_ports(self, portList): - cmd = 'AttachPorts -portList {' - 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 - 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 - return self.stc_config(cmd) - # start streamblock - # return: none - - def stc_streamblock_start(self, streamblock_list): - cmd = 'StreamBlockStart -StreamBlockList {' - 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): - cmd = 'StreamBlockStop -StreamBlockList {' - 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): - cmd = 'GeneratorStart -generatorList {' - 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): - cmd = 'GeneratorStop -generatorList {' - 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 |