diff options
Diffstat (limited to 'testsuites/vstf/vstf_scripts/vstf/agent/spirent/tools.py')
| -rw-r--r-- | testsuites/vstf/vstf_scripts/vstf/agent/spirent/tools.py | 424 |
1 files changed, 238 insertions, 186 deletions
diff --git a/testsuites/vstf/vstf_scripts/vstf/agent/spirent/tools.py b/testsuites/vstf/vstf_scripts/vstf/agent/spirent/tools.py index 088a7b13..0936d39d 100644 --- a/testsuites/vstf/vstf_scripts/vstf/agent/spirent/tools.py +++ b/testsuites/vstf/vstf_scripts/vstf/agent/spirent/tools.py @@ -11,324 +11,376 @@ import time from spirent import stcPython + class Spirent_Tools(object): baseAPI = stcPython() + def __init__(self): """This class provide API of Spirent - + """ super(Spirent_Tools, self).__init__() - - def send_packet(self,flow): + + def send_packet(self, flow): try: #import pdb - #pdb.set_trace() + # pdb.set_trace() flow = eval(flow) - #stc init action + # stc init action self.baseAPI.stc_perform(' ResetConfig -config system1') self.baseAPI.stc_init() - #create project + # create project project = self.baseAPI.stc_create_project() - #create port + # create port port_handle = self.baseAPI.stc_create_port(project) - #config port + # config port slot = flow['send_port'].split('/')[0] port = flow['send_port'].split('/')[1] - self.baseAPI.stc_config_port_location(port_handle,flow['tester_ip'],slot,port) - #create streamblock + self.baseAPI.stc_config_port_location( + port_handle, flow['tester_ip'], slot, port) + # create streamblock streamblock_handle = self.baseAPI.stc_create_streamblock( - port_name = port_handle, - ExpectedRxPort = '', - vlan_tag = flow['vlan'], - srcMac = flow['src_mac'], - dstMac = flow['dst_mac'], - sourceAddr = flow['src_ip'], - destAddr =flow['dst_ip'] - ) + port_name=port_handle, + ExpectedRxPort='', + vlan_tag=flow['vlan'], + srcMac=flow['src_mac'], + dstMac=flow['dst_mac'], + sourceAddr=flow['src_ip'], + destAddr=flow['dst_ip'] + ) # attach port port_list = [port_handle] self.baseAPI.stc_attach_ports(port_list) - #start streamblock + # start streamblock streamblock_list = [streamblock_handle] flag = self.baseAPI.stc_streamblock_start(streamblock_list) return str(streamblock_list).strip('[]') - except : + except: print("[ERROR]create stream block and send packet failed.") return False - def mac_learning(self,flowA,flowB): + def mac_learning(self, flowA, flowB): try: #import pdb - #pdb.set_trace() + # pdb.set_trace() flowA = eval(flowA) flowB = eval(flowB) port_list = [] streamblock_list = [] - #stc init action + # stc init action self.baseAPI.stc_perform(' ResetConfig -config system1') self.baseAPI.stc_init() - #create project + # create project project = self.baseAPI.stc_create_project() - #create port and config port - for flow in [ flowA,flowB ]: + # create port and config port + for flow in [flowA, flowB]: flow['port_handle'] = self.baseAPI.stc_create_port(project) tmp_test_ip = flow['tester_ip'] tmp_slot = flow['send_port'].split('/')[0] tmp_port = flow['send_port'].split('/')[1] - self.baseAPI.stc_config_port_location(flow['port_handle'],tmp_test_ip,tmp_slot,tmp_port) - #create streamblock - flow['streamblock'] = self.baseAPI.stc_create_streamblock(port_name = flow['port_handle'], - ExpectedRxPort = '', - vlan_tag = flow['vlan'], - srcMac = flow['src_mac'], - dstMac = flow['dst_mac'], - sourceAddr = flow['src_ip'], - destAddr =flow['dst_ip']) - #create port and stream block list + self.baseAPI.stc_config_port_location( + flow['port_handle'], tmp_test_ip, tmp_slot, tmp_port) + # create streamblock + flow['streamblock'] = self.baseAPI.stc_create_streamblock( + port_name=flow['port_handle'], + ExpectedRxPort='', + vlan_tag=flow['vlan'], + srcMac=flow['src_mac'], + dstMac=flow['dst_mac'], + sourceAddr=flow['src_ip'], + destAddr=flow['dst_ip']) + # create port and stream block list port_list.append(flow['port_handle']) streamblock_list.append(flow['streamblock']) - #attach port + # attach port self.baseAPI.stc_attach_ports(port_list) - #start streamblock + # start streamblock flag = self.baseAPI.stc_streamblock_start(streamblock_list) # mac learning time.sleep(2) # stop stream block self.baseAPI.stc_streamblock_stop(streamblock_list) # delete streamblock and release port - for flow in [ flowA,flowB ]: + for flow in [flowA, flowB]: tmp_test_ip = flow['tester_ip'] tmp_slot = flow['send_port'].split('/')[0] tmp_port = flow['send_port'].split('/')[1] self.baseAPI.stc_delete(flow['streamblock']) - self.baseAPI.stc_release('%s/%s/%s' %(tmp_test_ip,tmp_slot,tmp_port)) + self.baseAPI.stc_release( + '%s/%s/%s' % + (tmp_test_ip, tmp_slot, tmp_port)) # delete project self.baseAPI.stc_delete('project1') ret = self.baseAPI.stc_perform('ResetConfig -config system1') return True - except : + except: print("[ERROR]mac learning failed") return False - def stop_flow(self,streamblock_list,flow): + def stop_flow(self, streamblock_list, flow): flow = eval(flow) streamblock_list = streamblock_list.strip('\'').split(',') - #stop streamblock list - try : + # stop streamblock list + try: ret = self.baseAPI.stc_streamblock_stop(streamblock_list) - except : + except: print("[ERROR]Stop the streamblock list failed.") return False - #delete streamblock - try : - for streamblock in streamblock_list : + # delete streamblock + try: + for streamblock in streamblock_list: ret = self.baseAPI.stc_delete(streamblock) - except : + except: print("[ERROR]delete stream block.") return False - #release port - try : + # release port + try: slot = flow['send_port'].split('/')[0] port = flow['send_port'].split('/')[1] - ret = self.baseAPI.stc_release('%s/%s/%s' %(flow['tester_ip'],slot,port)) - except : + ret = self.baseAPI.stc_release( + '%s/%s/%s' % + (flow['tester_ip'], slot, port)) + except: print("[ERROR]Release port failed") return False - ##delete project - try : + # delete project + try: ret = self.baseAPI.stc_delete('project1') ret = self.baseAPI.stc_perform('ResetConfig -config system1') return True - except : + except: print("[ERROR]Delete project1 failed.") return False - - def run_rfc2544_throughput(self,forward_init_flows,reverse_init_flows): + + def run_rfc2544_throughput(self, forward_init_flows, reverse_init_flows): #import pdb - #pdb.set_trace() - #rebuild the flows + # pdb.set_trace() + # rebuild the flows forward_init_flows = eval(forward_init_flows) reverse_init_flows = eval(reverse_init_flows) - #stc init action + # stc init action self.baseAPI.stc_perform(' ResetConfig -config system1') self.baseAPI.stc_init() - #create project + # create project project = self.baseAPI.stc_create_project() - #create sequencer - seq_handle = self.baseAPI.stc_create('Sequencer -under %s' %(project)) - #create port handle + # create sequencer + seq_handle = self.baseAPI.stc_create('Sequencer -under %s' % (project)) + # create port handle forward_port_handle = self.baseAPI.stc_create_port(project) reverse_port_handle = self.baseAPI.stc_create_port(project) - #create forward flow streamblock + # create forward flow streamblock for key in forward_init_flows.keys(): forward_init_flows[key]['port_handle'] = forward_port_handle tmp_test_ip = forward_init_flows[key]['tester_ip'] - tmp_slot = forward_init_flows[key]['send_port'].split('/')[0] - tmp_port = forward_init_flows[key]['send_port'].split('/')[1] - self.baseAPI.stc_config_port_location(forward_init_flows[key]['port_handle'],tmp_test_ip,tmp_slot,tmp_port) - #create streamblock - forward_init_flows[key]['streamblock'] = self.baseAPI.stc_create_streamblock(port_name = forward_init_flows[key]['port_handle'], - vlan_tag = forward_init_flows[key]['vlan'], - ExpectedRxPort = reverse_port_handle, - srcMac = forward_init_flows[key]['src_mac'], - dstMac = forward_init_flows[key]['dst_mac'], - sourceAddr = forward_init_flows[key]['src_ip'], - destAddr = forward_init_flows[key]['dst_ip']) - #create reverse flow streamblock + tmp_slot = forward_init_flows[key]['send_port'].split('/')[0] + tmp_port = forward_init_flows[key]['send_port'].split('/')[1] + self.baseAPI.stc_config_port_location( + forward_init_flows[key]['port_handle'], tmp_test_ip, tmp_slot, tmp_port) + # create streamblock + forward_init_flows[key]['streamblock'] = self.baseAPI.stc_create_streamblock( + port_name=forward_init_flows[key]['port_handle'], + vlan_tag=forward_init_flows[key]['vlan'], + ExpectedRxPort=reverse_port_handle, + srcMac=forward_init_flows[key]['src_mac'], + dstMac=forward_init_flows[key]['dst_mac'], + sourceAddr=forward_init_flows[key]['src_ip'], + destAddr=forward_init_flows[key]['dst_ip']) + # create reverse flow streamblock for key in reverse_init_flows.keys(): reverse_init_flows[key]['port_handle'] = reverse_port_handle tmp_test_ip = reverse_init_flows[key]['tester_ip'] - tmp_slot = reverse_init_flows[key]['send_port'].split('/')[0] - tmp_port = reverse_init_flows[key]['send_port'].split('/')[1] - self.baseAPI.stc_config_port_location(reverse_init_flows[key]['port_handle'],tmp_test_ip,tmp_slot,tmp_port) - #create streamblock - reverse_init_flows[key]['streamblock'] = self.baseAPI.stc_create_streamblock(port_name = reverse_init_flows[key]['port_handle'], - vlan_tag = reverse_init_flows[key]['vlan'], - ExpectedRxPort = forward_port_handle, - srcMac = reverse_init_flows[key]['src_mac'], - dstMac = reverse_init_flows[key]['dst_mac'], - sourceAddr = reverse_init_flows[key]['src_ip'], - destAddr = reverse_init_flows[key]['dst_ip']) - #Create the RFC 2544 throughput test - throughput_config = self.baseAPI.stc_create('Rfc2544ThroughputConfig -under ',project, - '-AcceptableFrameLoss 0.01', - '-NumOfTrials 1', - '-DurationSeconds 60', - '-SearchMode BINARY', - '-RateLowerLimit 1', - '-RateUpperLimit 100', - '-RateInitial 10', - '-UseExistingStreamBlocks True', - '-EnableLearning False', - '-FrameSizeIterationMode CUSTOM', - '-CustomFrameSizeList "70 128 256 512 1024 1280 1518"', - '-LatencyType LIFO', - '-EnableJitterMeasurement TRUE' - ) + tmp_slot = reverse_init_flows[key]['send_port'].split('/')[0] + tmp_port = reverse_init_flows[key]['send_port'].split('/')[1] + self.baseAPI.stc_config_port_location( + reverse_init_flows[key]['port_handle'], tmp_test_ip, tmp_slot, tmp_port) + # create streamblock + reverse_init_flows[key]['streamblock'] = self.baseAPI.stc_create_streamblock( + port_name=reverse_init_flows[key]['port_handle'], + vlan_tag=reverse_init_flows[key]['vlan'], + ExpectedRxPort=forward_port_handle, + srcMac=reverse_init_flows[key]['src_mac'], + dstMac=reverse_init_flows[key]['dst_mac'], + sourceAddr=reverse_init_flows[key]['src_ip'], + destAddr=reverse_init_flows[key]['dst_ip']) + # Create the RFC 2544 throughput test + throughput_config = self.baseAPI.stc_create( + 'Rfc2544ThroughputConfig -under ', + project, + '-AcceptableFrameLoss 0.01', + '-NumOfTrials 1', + '-DurationSeconds 60', + '-SearchMode BINARY', + '-RateLowerLimit 1', + '-RateUpperLimit 100', + '-RateInitial 10', + '-UseExistingStreamBlocks True', + '-EnableLearning False', + '-FrameSizeIterationMode CUSTOM', + '-CustomFrameSizeList "70 128 256 512 1024 1280 1518"', + '-LatencyType LIFO', + '-EnableJitterMeasurement TRUE') #import pdb - #pdb.set_trace() + # pdb.set_trace() # list streamblocks streamblock_list = '" ' for key in forward_init_flows.keys(): - streamblock_list = streamblock_list+forward_init_flows[key]['streamblock']+' ' + streamblock_list = streamblock_list + \ + forward_init_flows[key]['streamblock'] + ' ' for key in reverse_init_flows.keys(): - streamblock_list = streamblock_list+reverse_init_flows[key]['streamblock']+' ' - streamblock_list = streamblock_list+'"' + streamblock_list = streamblock_list + \ + reverse_init_flows[key]['streamblock'] + ' ' + streamblock_list = streamblock_list + '"' - throughput_sbProfile= self.baseAPI.stc_create('Rfc2544StreamBlockProfile -under '+throughput_config+' -Active TRUE -LocalActive TRUE') - self.baseAPI.stc_config(throughput_sbProfile,'-StreamBlockList '+streamblock_list) - self.baseAPI.stc_perform('ExpandBenchmarkConfigCommand','-config ',throughput_config) + throughput_sbProfile = self.baseAPI.stc_create( + 'Rfc2544StreamBlockProfile -under ' + + throughput_config + + ' -Active TRUE -LocalActive TRUE') + self.baseAPI.stc_config( + throughput_sbProfile, + '-StreamBlockList ' + + streamblock_list) + self.baseAPI.stc_perform( + 'ExpandBenchmarkConfigCommand', + '-config ', + throughput_config) - #attach the port before testing - port_list = [ forward_port_handle,reverse_port_handle] + # attach the port before testing + port_list = [forward_port_handle, reverse_port_handle] self.baseAPI.stc_attach_ports(port_list) - #stc apply and begin to sequence test + # stc apply and begin to sequence test self.baseAPI.stc_apply() self.baseAPI.stc_perform("SequencerStart") - #wait until complete + # wait until complete self.baseAPI.stc_waituntilcomplete() - - #get result db - resultsdb = self.baseAPI.stc_get("system1.project.TestResultSetting", "-CurrentResultFileName") - results_dict = self.baseAPI.stc_perform('QueryResult','-DatabaseConnectionString',resultsdb,'-ResultPath RFC2544ThroughputTestResultDetailedSummaryView') - #print results_dict - return True,results_dict - def run_rfc2544_frameloss(self,forward_init_flows,reverse_init_flows): + # get result db + resultsdb = self.baseAPI.stc_get( + "system1.project.TestResultSetting", + "-CurrentResultFileName") + results_dict = self.baseAPI.stc_perform( + 'QueryResult', + '-DatabaseConnectionString', + resultsdb, + '-ResultPath RFC2544ThroughputTestResultDetailedSummaryView') + # print results_dict + return True, results_dict + + def run_rfc2544_frameloss(self, forward_init_flows, reverse_init_flows): #import pdb - #pdb.set_trace() - #rebuild the flows + # pdb.set_trace() + # rebuild the flows forward_init_flows = eval(forward_init_flows) reverse_init_flows = eval(reverse_init_flows) - #stc init action + # stc init action self.baseAPI.stc_perform(' ResetConfig -config system1') self.baseAPI.stc_init() - #create project + # create project project = self.baseAPI.stc_create_project() - #create sequencer - seq_handle = self.baseAPI.stc_create('Sequencer -under %s' %(project)) - #create port handle + # create sequencer + seq_handle = self.baseAPI.stc_create('Sequencer -under %s' % (project)) + # create port handle forward_port_handle = self.baseAPI.stc_create_port(project) reverse_port_handle = self.baseAPI.stc_create_port(project) - #create forward flow streamblock + # create forward flow streamblock for key in forward_init_flows.keys(): forward_init_flows[key]['port_handle'] = forward_port_handle tmp_test_ip = forward_init_flows[key]['tester_ip'] - tmp_slot = forward_init_flows[key]['send_port'].split('/')[0] - tmp_port = forward_init_flows[key]['send_port'].split('/')[1] - self.baseAPI.stc_config_port_location(forward_init_flows[key]['port_handle'],tmp_test_ip,tmp_slot,tmp_port) - #create streamblock - forward_init_flows[key]['streamblock'] = self.baseAPI.stc_create_streamblock(port_name = forward_init_flows[key]['port_handle'], - vlan_tag = forward_init_flows[key]['vlan'], - ExpectedRxPort = reverse_port_handle, - srcMac = forward_init_flows[key]['src_mac'], - dstMac = forward_init_flows[key]['dst_mac'], - sourceAddr = forward_init_flows[key]['src_ip'], - destAddr = forward_init_flows[key]['dst_ip']) - #create reverse flow streamblock + tmp_slot = forward_init_flows[key]['send_port'].split('/')[0] + tmp_port = forward_init_flows[key]['send_port'].split('/')[1] + self.baseAPI.stc_config_port_location( + forward_init_flows[key]['port_handle'], tmp_test_ip, tmp_slot, tmp_port) + # create streamblock + forward_init_flows[key]['streamblock'] = self.baseAPI.stc_create_streamblock( + port_name=forward_init_flows[key]['port_handle'], + vlan_tag=forward_init_flows[key]['vlan'], + ExpectedRxPort=reverse_port_handle, + srcMac=forward_init_flows[key]['src_mac'], + dstMac=forward_init_flows[key]['dst_mac'], + sourceAddr=forward_init_flows[key]['src_ip'], + destAddr=forward_init_flows[key]['dst_ip']) + # create reverse flow streamblock for key in reverse_init_flows.keys(): reverse_init_flows[key]['port_handle'] = reverse_port_handle tmp_test_ip = reverse_init_flows[key]['tester_ip'] - tmp_slot = reverse_init_flows[key]['send_port'].split('/')[0] - tmp_port = reverse_init_flows[key]['send_port'].split('/')[1] - self.baseAPI.stc_config_port_location(reverse_init_flows[key]['port_handle'],tmp_test_ip,tmp_slot,tmp_port) - #create streamblock - reverse_init_flows[key]['streamblock'] = self.baseAPI.stc_create_streamblock(port_name = reverse_init_flows[key]['port_handle'], - vlan_tag = reverse_init_flows[key]['vlan'], - ExpectedRxPort = forward_port_handle, - srcMac = reverse_init_flows[key]['src_mac'], - dstMac = reverse_init_flows[key]['dst_mac'], - sourceAddr = reverse_init_flows[key]['src_ip'], - destAddr = reverse_init_flows[key]['dst_ip']) - #Create the RFC 2544 frameloss test - frameloss_config = self.baseAPI.stc_create('Rfc2544FrameLossConfig -under ',project, - '-NumOfTrials 1 ', - '-DurationSeconds 60 ', - '-LoadUnits PERCENT_LINE_RATE ', - '-LoadType CUSTOM ' - '-CustomLoadList 100 ' - '-UseExistingStreamBlocks True ', - '-EnableLearning False ', - '-FrameSizeIterationMode CUSTOM ', - '-CustomFrameSizeList "70 128 256 512 1024 1280 1518"', - '-LatencyType LIFO', - '-EnableJitterMeasurement TRUE' - ) + tmp_slot = reverse_init_flows[key]['send_port'].split('/')[0] + tmp_port = reverse_init_flows[key]['send_port'].split('/')[1] + self.baseAPI.stc_config_port_location( + reverse_init_flows[key]['port_handle'], tmp_test_ip, tmp_slot, tmp_port) + # create streamblock + reverse_init_flows[key]['streamblock'] = self.baseAPI.stc_create_streamblock( + port_name=reverse_init_flows[key]['port_handle'], + vlan_tag=reverse_init_flows[key]['vlan'], + ExpectedRxPort=forward_port_handle, + srcMac=reverse_init_flows[key]['src_mac'], + dstMac=reverse_init_flows[key]['dst_mac'], + sourceAddr=reverse_init_flows[key]['src_ip'], + destAddr=reverse_init_flows[key]['dst_ip']) + # Create the RFC 2544 frameloss test + frameloss_config = self.baseAPI.stc_create( + 'Rfc2544FrameLossConfig -under ', + project, + '-NumOfTrials 1 ', + '-DurationSeconds 60 ', + '-LoadUnits PERCENT_LINE_RATE ', + '-LoadType CUSTOM ' + '-CustomLoadList 100 ' + '-UseExistingStreamBlocks True ', + '-EnableLearning False ', + '-FrameSizeIterationMode CUSTOM ', + '-CustomFrameSizeList "70 128 256 512 1024 1280 1518"', + '-LatencyType LIFO', + '-EnableJitterMeasurement TRUE') #import pdb - #pdb.set_trace() + # pdb.set_trace() # list streamblocks streamblock_list = '" ' for key in forward_init_flows.keys(): - streamblock_list = streamblock_list+forward_init_flows[key]['streamblock']+' ' + streamblock_list = streamblock_list + \ + forward_init_flows[key]['streamblock'] + ' ' for key in reverse_init_flows.keys(): - streamblock_list = streamblock_list+reverse_init_flows[key]['streamblock']+' ' - streamblock_list = streamblock_list+'"' + streamblock_list = streamblock_list + \ + reverse_init_flows[key]['streamblock'] + ' ' + streamblock_list = streamblock_list + '"' - frameloss_sbProfile= self.baseAPI.stc_create('Rfc2544StreamBlockProfile -under '+frameloss_config+' -Active TRUE -LocalActive TRUE') - self.baseAPI.stc_config(frameloss_sbProfile,'-StreamBlockList '+streamblock_list) - self.baseAPI.stc_perform('ExpandBenchmarkConfigCommand','-config ',frameloss_config) + frameloss_sbProfile = self.baseAPI.stc_create( + 'Rfc2544StreamBlockProfile -under ' + + frameloss_config + + ' -Active TRUE -LocalActive TRUE') + self.baseAPI.stc_config( + frameloss_sbProfile, + '-StreamBlockList ' + + streamblock_list) + self.baseAPI.stc_perform( + 'ExpandBenchmarkConfigCommand', + '-config ', + frameloss_config) - #attach the port before testing - port_list = [ forward_port_handle,reverse_port_handle] + # attach the port before testing + port_list = [forward_port_handle, reverse_port_handle] self.baseAPI.stc_attach_ports(port_list) - #stc apply and begin to sequence test + # stc apply and begin to sequence test self.baseAPI.stc_apply() self.baseAPI.stc_perform("SequencerStart") - #wait until complete + # wait until complete self.baseAPI.stc_waituntilcomplete() - #get result db - resultsdb = self.baseAPI.stc_get("system1.project.TestResultSetting", "-CurrentResultFileName") - results_dict = self.baseAPI.stc_perform('QueryResult','-DatabaseConnectionString',resultsdb,'-ResultPath RFC2544FrameLossTestResultDetailedSummaryView') + # get result db + resultsdb = self.baseAPI.stc_get( + "system1.project.TestResultSetting", + "-CurrentResultFileName") + results_dict = self.baseAPI.stc_perform( + 'QueryResult', + '-DatabaseConnectionString', + resultsdb, + '-ResultPath RFC2544FrameLossTestResultDetailedSummaryView') #import pdb - #pdb.set_trace() - return True,results_dict + # pdb.set_trace() + return True, results_dict - def run_rfc2544_latency(self,forward_init_flows,reverse_init_flows): + def run_rfc2544_latency(self, forward_init_flows, reverse_init_flows): pass - |