1 # Copyright 2015 Spirent Communications.
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
16 @author Spirent Comunications
18 This test automates the RFC2544 throughput test using the Spirent
19 TestCenter command sequencer. This test supports Python 2.7.
22 from __future__ import print_function
28 if not os.path.exists(path):
32 print("Failed to create directory %s: %s" % (path, str(e)))
36 def write_query_results_to_csv(results_path, csv_results_file_prefix, query_results):
37 create_dir(results_path)
38 file = os.path.join(results_path, csv_results_file_prefix + ".csv")
39 with open(file, "wb") as f:
40 f.write(query_results["Columns"].replace(" ", ",") + "\n")
41 for row in query_results["Output"].replace("} {", ",").replace("{", "").replace("}", "").split(","):
42 f.write(row.replace(" ", ",") + "\n")
45 def destroy_session(stc, lab_server_addr, test_session_name):
46 stc.perform("CSServerConnectCommand",
49 cs_server = stc.get("system1", "children-csserver")
51 cs_test_sessions = stc.get(cs_server, "children-cstestsession").split()
53 for session in cs_test_sessions:
54 name = stc.get(session, "Name")
55 if test_session_name in name:
56 stc.perform("CSDestroyTestSession",
61 def create_session(stc, lab_server_addr, test_session_name):
62 destroy_session(stc, lab_server_addr, test_session_name)
63 stc.perform("CSTestSessionConnect",
65 CreateNewTestSession="TRUE",
66 TestSessionName=test_session_name)
69 def positive_int(value):
72 raise argparse.ArgumentTypeError("%s is an invalid positive int value" % value)
76 def percent_float(value):
78 if pvalue < 0.0 or pvalue > 100.0:
79 raise argparse.ArgumentTypeError("%s not in range [0.0, 100.0]" % pvalue)
84 parser = argparse.ArgumentParser()
86 requirednamed = parser.add_argument_group("required named arguments")
87 requirednamed.add_argument("--lab_server_addr",
89 help="The IP address of the Spirent Lab Server",
90 dest="lab_server_addr")
91 requirednamed.add_argument("--license_server_addr",
93 help="The IP address of the Spirent License Server",
94 dest="license_server_addr")
95 requirednamed.add_argument("--east_chassis_addr",
97 help="The TestCenter chassis IP address to use for the east test port",
98 dest="east_chassis_addr")
99 requirednamed.add_argument("--east_slot_num",
102 help="The TestCenter slot number to use for the east test port",
103 dest="east_slot_num")
104 requirednamed.add_argument("--east_port_num",
107 help="The TestCenter port number to use for the east test port",
108 dest="east_port_num")
109 requirednamed.add_argument("--west_chassis_addr",
111 help="The TestCenter chassis IP address to use for the west test port",
112 dest="west_chassis_addr")
113 requirednamed.add_argument("--west_slot_num",
116 help="The TestCenter slot number to use for the west test port",
117 dest="west_slot_num")
118 requirednamed.add_argument("--west_port_num",
121 help="The TestCenter port number to use for the west test port",
122 dest="west_port_num")
123 # Optional parameters
124 optionalnamed = parser.add_argument_group("optional named arguments")
125 optionalnamed.add_argument("--test_session_name",
127 default="RFC2544 East-West Throughput",
128 help="The friendly name to identify the Spirent Lab Server test session",
129 dest="test_session_name")
130 optionalnamed.add_argument("--results_dir",
133 help="The directory to copy results to",
135 optionalnamed.add_argument("--csv_results_file_prefix",
137 default="Rfc2544Tput",
138 help="The prefix for the CSV results files",
139 dest="csv_results_file_prefix")
140 optionalnamed.add_argument("--num_trials",
144 help="The number of trials to execute during the test",
146 optionalnamed.add_argument("--trial_duration_sec",
150 help="The duration of each trial executed during the test",
151 dest="trial_duration_sec")
152 optionalnamed.add_argument("--traffic_pattern",
154 choices=["BACKBONE", "MESH", "PAIR"],
156 help="The traffic pattern between endpoints",
157 dest="traffic_pattern")
158 optionalnamed.add_argument("--search_mode",
160 choices=["COMBO", "STEP", "BINARY"],
162 help="The search mode used to find the throughput rate",
164 optionalnamed.add_argument("--learning_mode",
166 choices=["AUTO", "L2_LEARNING", "L3_LEARNING", "NONE"],
168 help="The learning mode used during the test, default = 'NONE'",
169 dest="learning_mode")
170 optionalnamed.add_argument("--rate_lower_limit_pct",
174 help="The minimum percent line rate that will be used during the test",
175 dest="rate_lower_limit_pct")
176 optionalnamed.add_argument("--rate_upper_limit_pct",
180 help="The maximum percent line rate that will be used during the test",
181 dest="rate_upper_limit_pct")
182 optionalnamed.add_argument("--rate_initial_pct",
186 help="If Search Mode is BINARY, the percent line rate that will be used at the start of the test",
187 dest="rate_initial_pct")
188 optionalnamed.add_argument("--rate_step_pct",
192 help="If SearchMode is STEP, the percent load increase per step",
193 dest="rate_step_pct")
194 optionalnamed.add_argument("--resolution_pct",
198 help="The minimum percentage of load adjustment between iterations",
199 dest="resolution_pct")
200 optionalnamed.add_argument("--frame_size_list",
201 type=lambda s: [int(item) for item in s.split(',')],
204 help="A comma-delimited list of frame sizes",
205 dest="frame_size_list")
206 optionalnamed.add_argument("--acceptable_frame_loss_pct",
210 help="The maximum acceptable frame loss percent in any iteration",
211 dest="acceptable_frame_loss_pct")
212 optionalnamed.add_argument("--east_intf_addr",
214 default="192.85.1.3",
215 help="The address to assign to the first emulated device interface on the first east port",
216 dest="east_intf_addr")
217 optionalnamed.add_argument("--east_intf_gateway_addr",
219 default="192.85.1.53",
220 help="The gateway address to assign to the first emulated device interface on the first east port",
221 dest="east_intf_gateway_addr")
222 optionalnamed.add_argument("--west_intf_addr",
224 default="192.85.1.53",
225 help="The address to assign to the first emulated device interface on the first west port",
226 dest="west_intf_addr")
227 optionalnamed.add_argument("--west_intf_gateway_addr",
229 default="192.85.1.53",
230 help="The gateway address to assign to the first emulated device interface on the first west port",
231 dest="west_intf_gateway_addr")
232 parser.add_argument("-v",
236 help="More output during operation when present",
239 args = parser.parse_args()
242 print("Make sure results directory exists")
243 create_dir(args.results_dir)
245 # Load Spirent Test Center library
246 from StcPython import StcPython
249 tx_port_loc = "//%s/%s/%s" % (args.east_chassis_addr,
252 rx_port_loc = "//%s/%s/%s" % (args.west_chassis_addr,
256 # This line will write the TestCenter commands to a log file
257 stc.config("automationoptions", logto="test-op", loglevel="DEBUG")
259 # Retrieve and display the current API version.
261 print ("SpirentTestCenter system version: %s" % stc.get("system1", "version"))
263 create_session(stc, args.lab_server_addr, args.test_session_name)
269 print("Bring up license server")
270 license_mgr = stc.get("system1", "children-licenseservermanager")
272 print("license_mgr = %s" % license_mgr)
273 stc.create("LicenseServer", under=license_mgr, server=args.license_server_addr)
275 # Create the root project object
277 print ("Creating project ...")
278 project = stc.get("System1", "children-Project")
282 print ("Creating ports ...")
283 east_chassis_port = stc.create("port",
285 location=tx_port_loc,
286 useDefaultHost="False")
287 west_chassis_port = stc.create("port",
289 location=rx_port_loc,
290 useDefaultHost="False")
293 print ("Creating devices...")
294 # Create emulated genparam for east port
295 east_device_gen_params = stc.create("EmulatedDeviceGenParams",
297 Port=east_chassis_port)
298 # Create the DeviceGenEthIIIfParams object
299 stc.create("DeviceGenEthIIIfParams",
300 under=east_device_gen_params)
301 # Configuring Ipv4 interfaces
302 stc.create("DeviceGenIpv4IfParams",
303 under=east_device_gen_params,
304 Addr=args.east_intf_addr,
305 Gateway=args.east_intf_gateway_addr)
306 # Create Devices using the Device Wizard
307 device_gen_config = stc.perform("DeviceGenConfigExpand",
309 GenParams=east_device_gen_params)
310 # Append to the device list
311 device_list.append(device_gen_config['ReturnList'])
313 # Create emulated genparam for west port
314 west_device_gen_params = stc.create("EmulatedDeviceGenParams",
316 Port=west_chassis_port)
317 # Create the DeviceGenEthIIIfParams object
318 stc.create("DeviceGenEthIIIfParams",
319 under=west_device_gen_params)
320 # Configuring Ipv4 interfaces
321 stc.create("DeviceGenIpv4IfParams",
322 under=west_device_gen_params,
323 Addr=args.west_intf_addr,
324 Gateway=args.west_intf_gateway_addr)
325 # Create Devices using the Device Wizard
326 device_gen_config = stc.perform("DeviceGenConfigExpand",
328 GenParams=west_device_gen_params)
329 # Append to the device list
330 device_list.append(device_gen_config['ReturnList'])
335 # Create the RFC 2544 throughput test
337 print ("Set up the RFC2544 throughput test...")
338 stc.perform("Rfc2544SetupThroughputTestCommand",
339 AcceptableFrameLoss=args.acceptable_frame_loss_pct,
340 Duration=args.trial_duration_sec,
341 FrameSizeList=args.frame_size_list,
342 LearningMode=args.learning_mode,
343 NumOfTrials=args.num_trials,
344 RateInitial=args.rate_initial_pct,
345 RateLowerLimit=args.rate_lower_limit_pct,
346 RateStep=args.rate_step_pct,
347 RateUpperLimit=args.rate_upper_limit_pct,
348 Resolution=args.resolution_pct,
349 SearchMode=args.search_mode,
350 TrafficPattern=args.traffic_pattern)
352 # Save the configuration
353 stc.perform("SaveToTcc", Filename="2544.tcc")
355 # Connect to the hardware...
356 stc.perform("AttachPorts", portList=stc.get("system1.project", "children-port"), autoConnect="TRUE")
358 # Apply configuration.
360 print("Apply configuration...")
364 print("Starting the sequencer...")
365 stc.perform("SequencerStart")
367 # Wait for sequencer to finish
368 print("Starting test... Please wait for the test to complete...")
369 stc.waitUntilComplete()
370 print("The test has completed... Saving results...")
372 # Determine what the results database filename is...
373 lab_server_resultsdb = stc.get("system1.project.TestResultSetting", "CurrentResultFileName")
376 print("The lab server results database is %s" % lab_server_resultsdb)
378 stc.perform("CSSynchronizeFiles",
379 DefaultDownloadDir=args.results_dir)
381 resultsdb = args.results_dir + lab_server_resultsdb.split("/Results")[1]
383 print("The local summary DB file has been saved to %s" % resultsdb)
385 # The returns the "RFC2544ThroughputTestResultDetailedSummaryView" table view from the results database.
386 # There are other views available.
387 resultsdict = stc.perform("QueryResult",
388 DatabaseConnectionString=resultsdb,
389 ResultPath="RFC2544ThroughputTestResultDetailedSummaryView")
391 print("resultsdict[\"Columns\"]: %s" % resultsdict["Columns"])
392 print("resultsdict[\"Output\"]: %s" % resultsdict["Output"])
395 print("Result paths: %s" % stc.perform("GetTestResultSettingPaths"))
397 # Write results to csv
399 print("Writing CSV file to results directory %s" % args.results_dir)
400 write_query_results_to_csv(args.results_dir, args.csv_results_file_prefix, resultsdict)
403 print("Exception: %s" % str(e))
404 # destroy_session(stc, args.lab_server_addr, args.test_session_name)
408 print("Disconnect from lab server")
409 stc.perform("CSTestSessionDisconnect")
412 print("Destroy session on lab server")
413 # destroy_session(stc, args.lab_server_addr, args.test_session_name)
414 print("Test complete!")
417 if __name__ == "__main__":