testsuites/vstf/vstf_scripts/vstf/agent/spirent/spirent.py

   1 ##############################################################################
   2 # Copyright (c) 2015 Huawei Technologies Co.,Ltd and others.
   3 #
   4 # All rights reserved. This program and the accompanying materials
   5 # are made available under the terms of the Apache License, Version 2.0
   6 # which accompanies this distribution, and is available at
   7 # http://www.apache.org/licenses/LICENSE-2.0
   8 ##############################################################################
   9
  10
  11 import Tkinter
  12
  13
  14 def build_cmd(*args):
  15     cmd = ''
  16     for arg in args:
  17         cmd = cmd + str(arg) + ' '
  18     #import pdb
  19     # pdb.set_trace()
  20     return cmd
  21
  22
  23 class stcPython():
  24
  25     def __init__(self):
  26         self.tclsh = Tkinter.Tcl()
  27         self.stcpkg = '/home/Spirent_TestCenter_4.46/Spirent_TestCenter_Application_Linux'
  28         self.tclsh.eval(
  29             "set auto_path [ linsert $auto_path 0 %s ]" %
  30             (self.stcpkg))
  31         self.tclsh.eval("package require SpirentTestCenter")
  32
  33     def build_cmd(self, *args):
  34         cmd = ''
  35         for arg in args:
  36             cmd = cmd + str(arg) + ' '
  37         return cmd
  38
  39     # [ stc base interface ]
  40     def stc_init(self, *args):
  41         cmd = build_cmd('stc::init', *args)
  42         return self.tclsh.eval(cmd)
  43     # stc connect
  44
  45     def stc_connect(self, *args):
  46         cmd = build_cmd('stc::connect', *args)
  47         return self.tclsh.eval(cmd)
  48     # stc disconnect
  49
  50     def stc_disconnect(self, *args):
  51         cmd = build_cmd('stc::disconnect', *args)
  52         return self.tclsh.eval(cmd)
  53     # stc create
  54
  55     def stc_create(self, *args):
  56         cmd = build_cmd('stc::create', *args)
  57         return self.tclsh.eval(cmd)
  58     # stc delete
  59
  60     def stc_delete(self, *args):
  61         cmd = build_cmd('stc::delete', *args)
  62         return self.tclsh.eval(cmd)
  63     # stc config
  64
  65     def stc_config(self, *args):
  66         cmd = build_cmd('stc::config', *args)
  67         return self.tclsh.eval(cmd)
  68     # stc get
  69
  70     def stc_get(self, *args):
  71         cmd = build_cmd('stc::get', *args)
  72         return self.tclsh.eval(cmd)
  73     # stc apply
  74
  75     def stc_apply(self, *args):
  76         cmd = build_cmd('stc::apply', *args)
  77         return self.tclsh.eval(cmd)
  78     # stc perform
  79
  80     def stc_perform(self, *args):
  81         cmd = build_cmd('stc::perform', *args)
  82         return self.tclsh.eval(cmd)
  83     # stc reserve
  84
  85     def stc_reserve(self, *args):
  86         cmd = build_cmd('stc::reserve', *args)
  87         return self.tclsh.eval(cmd)
  88     # stc release
  89
  90     def stc_release(self, *args):
  91         cmd = build_cmd('stc::release', *args)
  92         return self.tclsh.eval(cmd)
  93     # stc subscribe
  94
  95     def stc_subscribe(self, *args):
  96         cmd = build_cmd('stc::subscribe', *args)
  97         return self.tclsh.eval(cmd)
  98     # stc unsubscribe
  99
 100     def stc_unsubscribe(self, *args):
 101         cmd = build_cmd('stc::unsubscribe', *args)
 102         return self.tclsh.eval(cmd)
 103     # stc wait until sequencer complete
 104
 105     def stc_waituntilcomplete(self, *args):
 106         cmd = build_cmd('stc::waituntilcomplete', *args)
 107         return self.tclsh.eval(cmd)
 108     # stc help
 109
 110     def stc_help(self, *args):
 111         cmd = build_cmd('stc::help', *args)
 112         return self.tclsh.eval(cmd)
 113
 114     # [ stc expand interface ]
 115     # get one dict-key's value
 116     # return value
 117     def stc_get_value(self, stc_dict, stc_key):
 118         cmd = stc_dict + ' -' + stc_key
 119         return self.stc_get(cmd)
 120     # create project
 121     # return: project_name
 122
 123     def stc_create_project(self):
 124         return self.stc_create('project')
 125     # create port under project
 126     # return: port name
 127
 128     def stc_create_port(self, project_name):
 129         cmd = 'port -under ' + project_name
 130         return self.stc_create(cmd)
 131     # config port location
 132     # return: None
 133
 134     def stc_config_port_location(self, port_name, chassisAddress, slot, port):
 135         #import pdb
 136         # pdb.set_trace()
 137         cmd = port_name + ' -location //' + chassisAddress + \
 138             '/' + slot + '/' + port + ' -UseDefaultHost False'
 139         return self.stc_config(cmd)
 140     # create streamblock under port
 141     # return: streamblock name
 142
 143     def stc_create_streamblock(
 144             self,
 145             port_name,
 146             vlan_tag,
 147             ExpectedRxPort,
 148             srcMac,
 149             dstMac,
 150             sourceAddr,
 151             destAddr):
 152         #import pdb
 153         # pdb.set_trace()
 154         if vlan_tag is None or vlan_tag == 'None':
 155             frameStruc = '"EthernetII IPv4 Udp"'
 156             if ExpectedRxPort == '':
 157                 return self.stc_create(
 158                     'streamBlock -under ',
 159                     port_name,
 160                     '-frameConfig ',
 161                     frameStruc,
 162                     '-frame "EthernetII.srcMac',
 163                     srcMac,
 164                     'EthernetII.dstMac',
 165                     dstMac,
 166                     'IPv4.1.sourceAddr',
 167                     sourceAddr,
 168                     'IPv4.1.destAddr',
 169                     destAddr,
 170                     '"')
 171             else:
 172                 return self.stc_create(
 173                     'streamBlock -under ',
 174                     port_name,
 175                     '-ExpectedRxPort',
 176                     ExpectedRxPort,
 177                     '-frameConfig ',
 178                     frameStruc,
 179                     '-frame "EthernetII.srcMac',
 180                     srcMac,
 181                     'EthernetII.dstMac',
 182                     dstMac,
 183                     'IPv4.1.sourceAddr',
 184                     sourceAddr,
 185                     'IPv4.1.destAddr',
 186                     destAddr,
 187                     '"')
 188         else:
 189             frameStruc = '"EthernetII Vlan IPv4 Udp"'
 190             if ExpectedRxPort == '':
 191                 return self.stc_create(
 192                     'streamBlock -under ',
 193                     port_name,
 194                     '-frameConfig ' +
 195                     frameStruc,
 196                     '-frame "EthernetII.srcMac',
 197                     srcMac,
 198                     'EthernetII.dstMac',
 199                     dstMac,
 200                     'Vlan.1.id',
 201                     vlan_tag,
 202                     'IPv4.1.sourceAddr',
 203                     sourceAddr,
 204                     'IPv4.1.destAddr',
 205                     destAddr,
 206                     '"')
 207             else:
 208                 return self.stc_create(
 209                     'streamBlock -under ',
 210                     port_name,
 211                     '-ExpectedRxPort',
 212                     ExpectedRxPort,
 213                     '-frameConfig ' +
 214                     frameStruc,
 215                     '-frame "EthernetII.srcMac',
 216                     srcMac,
 217                     'EthernetII.dstMac',
 218                     dstMac,
 219                     'Vlan.1.id',
 220                     vlan_tag,
 221                     'IPv4.1.sourceAddr',
 222                     sourceAddr,
 223                     'IPv4.1.destAddr',
 224                     destAddr,
 225                     '"')
 226     # config streamblock with part arguments
 227     # argument list use args dictionary
 228
 229     def stc_config_streamblock(self, streamblock_name, args_dict):
 230         cmd = ''
 231         for key in args_dict.keys():
 232             temp_cmd = '-' + key + ' ' + str(args_dict[key])
 233             cmd = cmd + temp_cmd
 234         return self.stc_config(streamblock_name, cmd)
 235     # get generator name from port name
 236     # return: generator name
 237
 238     def stc_get_generator(self, port_name):
 239         cmd = port_name + ' -children-generator'
 240         return self.stc_get(cmd)
 241     # config generator with part arguments
 242     # argument list use args dictionary
 243     # return none
 244
 245     def stc_config_generator(self, generator_name, args_dict):
 246         cmd = ''
 247         for key in args_dict.keys():
 248             temp_cmd = '-' + key + ' ' + str(args_dict[key])
 249             cmd = cmd + temp_cmd
 250         return self.stc_config(generator_name, cmd)
 251     # attach port
 252     # return: port's parent project info
 253
 254     def stc_attach_ports(self, portList):
 255         cmd = 'AttachPorts -portList {'
 256         for port in portList:
 257             cmd = cmd + ' ' + port
 258         cmd = cmd + '} -autoConnect TRUE'
 259         return self.stc_perform(cmd)
 260     # config src mac and dst mac
 261     # return: none
 262
 263     def stc_config_ethII(self, ethII, src_mac, dst_mac):
 264         cmd = ethII + ' -srcMac ' + src_mac + ' -dstMac ' + dst_mac
 265         return self.stc_config(cmd)
 266     # config src ip and dst ip
 267     # return: none
 268
 269     def stc_config_ethIII(self, ethIII, src_ip, dst_ip):
 270         cmd = ethIII + ' -sourceAddr ' + src_ip + ' -destAddr ' + dst_ip
 271         return self.stc_config(cmd)
 272     # start streamblock
 273     # return: none
 274
 275     def stc_streamblock_start(self, streamblock_list):
 276         cmd = 'StreamBlockStart -StreamBlockList {'
 277         for streamblock in streamblock_list:
 278             cmd = cmd + ' ' + streamblock
 279         cmd = cmd + ' } -ExecuteSynchronous TRUE'
 280         return self.stc_perform(cmd)
 281     # stop streamblock
 282
 283     def stc_streamblock_stop(self, streamblock_list):
 284         cmd = 'StreamBlockStop -StreamBlockList {'
 285         for streamblock in streamblock_list:
 286             cmd = cmd + ' ' + streamblock
 287         cmd = cmd + ' } -ExecuteSynchronous TRUE'
 288         return self.stc_perform(cmd)
 289     # start generator
 290     # return: none
 291
 292     def stc_generator_start(self, generator_List):
 293         cmd = 'GeneratorStart -generatorList {'
 294         for generator in generator_List:
 295             cmd = cmd + ' ' + generator
 296         cmd = cmd + ' }'
 297         return self.stc_perform(cmd)
 298     # stop generator
 299     # return: none
 300
 301     def stc_generator_stop(self, generator_List):
 302         cmd = 'GeneratorStop -generatorList {'
 303         for generator in generator_List:
 304             cmd = cmd + ' ' + generator
 305         cmd = cmd + ' }'
 306         return self.stc_perform(cmd)
 307     # create rfc2544 throughput test
 308
 309     def stc_setup_rfc2544_throughput(self):
 310         pass
 311     # create rfc2544 frameloss test
 312
 313     def stc_setup_rfc2544_frameloss(self):
 314         pass
 315     # create rfc2544 latency test
 316
 317     def stc_setup_rfc2544_latency(self):
 318         pass
 319     # start Sequence start
 320
 321     def stc_sequence_start(self):
 322         return self.stc_perform('SequencerStart')
 323     # output rfc2544 throughput result
 324
 325     def stc_get_rfc2544_throughput_result(self):
 326         pass
 327     # output rfc2544 frameloss result
 328
 329     def stc_get_rfc2544_frameloss_result(self):
 330         pass
 331     # output rfc2544 latency result
 332
 333     def stc_get_rfc2544_latency_result(self):
 334         pass