# limitations under the License.
"""Host Stack util library."""
+import json
from time import sleep
from robot.api import logger
f"socket-name {vpp_echo_attributes[u'vpp_api_socket']} " \
f"{vpp_echo_attributes[u'json_output']} " \
f"uri {proto}://{addr}/{port} " \
+ f"nthreads {vpp_echo_attributes[u'nthreads']} " \
+ f"mq-size {vpp_echo_attributes[u'mq_size']} " \
f"nclients {vpp_echo_attributes[u'nclients']} " \
f"quic-streams {vpp_echo_attributes[u'quic_streams']} " \
f"time {vpp_echo_attributes[u'time']} " \
sleep(1)
@staticmethod
- def analyze_hoststack_test_program_output(node, role, nsim_attr,
- program):
+ def analyze_hoststack_test_program_output(
+ node, role, nsim_attr, program):
"""Gather HostStack test program output and check for errors.
+ The [defer_fail] return bool is used instead of failing immediately
+ to allow the analysis of both the client and server instances of
+ the test program for debugging a test failure. When [defer_fail]
+ is true, then the string returned is debug output instead of
+ JSON formatted test program results.
+
:param node: DUT node.
:param role: Role (client|server) of test program.
:param nsim_attr: Network Simulation Attributes.
:type role: str
:type nsim_attr: dict
:type program: dict
- :returns: tuple of no results bool and test program results.
+ :returns: tuple of [defer_fail] bool and either JSON formatted hoststack
+ test program output or failure debug output.
:rtype: bool, str
:raises RuntimeError: If node subtype is not a DUT.
"""
program_stdout, program_stderr = \
HoststackUtil.get_hoststack_test_program_logs(node, program)
- no_results = False
env_vars = f"{program[u'env_vars']} " if u"env_vars" in program else u""
program_cmd = f"{env_vars}{program_name} {program[u'args']}"
test_results = f"Test Results of '{program_cmd}':\n"
- if nsim_attr[u"output_feature_enable"] or \
- nsim_attr[u"cross_connect_feature_enable"]:
- if nsim_attr[u"output_feature_enable"]:
+ if nsim_attr[u"output_nsim_enable"] or \
+ nsim_attr[u"xc_nsim_enable"]:
+ if nsim_attr[u"output_nsim_enable"]:
feature_name = u"output"
else:
feature_name = u"cross-connect"
f"NSIM({feature_name}): delay " \
f"{nsim_attr[u'delay_in_usec']} usecs, " \
f"avg-pkt-size {nsim_attr[u'average_packet_size']}, " \
- f"bandwidth {nsim_attr[u'bandwidth_in_bits_per_second']} " \
+ f"bandwidth {nsim_attr[u'bw_in_bits_per_second']} " \
f"bits/sec, pkt-drop-rate {nsim_attr[u'packets_per_drop']} " \
f"pkts/drop\n"
- if u"error" in program_stderr.lower():
- test_results += f"ERROR DETECTED:\n{program_stderr}"
- raise RuntimeError(test_results)
- if program_stdout:
- bad_test_results = False
- if program == u"vpp_echo" and u"JSON stats" not in program_stdout:
- test_results += u"Invalid test data output!\n"
- bad_test_results = True
- test_results += program_stdout
- if bad_test_results:
- raise RuntimeError(test_results)
- else:
- no_results = True
- test_results += f"\nNo {program} test data retrieved!\n"
- cmd = u"ls -l /tmp/*.log"
- ls_stdout, _ = exec_cmd_no_error(node, cmd, sudo=True)
- test_results += f"{ls_stdout}\n"
-
# TODO: Incorporate show error stats into results analysis
- host = node[u"host"]
test_results += \
- f"\n{role} VPP 'show errors' on host {host}:\n" \
+ f"\n{role} VPP 'show errors' on host {node[u'host']}:\n" \
f"{PapiSocketExecutor.run_cli_cmd(node, u'show error')}\n"
- return no_results, test_results
+ if u"error" in program_stderr.lower():
+ test_results += f"ERROR DETECTED:\n{program_stderr}"
+ return (True, test_results)
+ if not program_stdout:
+ test_results += f"\nNo {program} test data retrieved!\n"
+ ls_stdout, _ = exec_cmd_no_error(node, u"ls -l /tmp/*.log",
+ sudo=True)
+ test_results += f"{ls_stdout}\n"
+ return (True, test_results)
+ if program[u"name"] == u"vpp_echo":
+ if u"JSON stats" in program_stdout and \
+ u'"has_failed": "0"' in program_stdout:
+ json_start = program_stdout.find(u"{")
+ #TODO: Fix parsing once vpp_echo produces valid
+ # JSON output. Truncate for now.
+ json_end = program_stdout.find(u',\n "closing"')
+ json_results = f"{program_stdout[json_start:json_end]}\n}}"
+ program_json = json.loads(json_results)
+ else:
+ test_results += u"Invalid test data output!\n" + program_stdout
+ return (True, test_results)
+ elif program[u"name"] == u"iperf3":
+ test_results += program_stdout
+ iperf3_json = json.loads(program_stdout)
+ program_json = iperf3_json[u"intervals"][0][u"sum"]
+ else:
+ test_results += u"Unknown HostStack Test Program!\n" + \
+ program_stdout
+ return (True, program_stdout)
+ return (False, json.dumps(program_json))
@staticmethod
- def no_hoststack_test_program_results(server_no_results, client_no_results):
- """Return True if no HostStack test program output was gathered.
+ def hoststack_test_program_defer_fail(server_defer_fail, client_defer_fail):
+ """Return True if either HostStack test program fail was deferred.
- :param server_no_results: server no results value.
- :param client_no_results: client no results value.
- :type server_no_results: bool
- :type client_no_results: bool
+ :param server_defer_fail: server no results value.
+ :param client_defer_fail: client no results value.
+ :type server_defer_fail: bool
+ :type client_defer_fail: bool
:rtype: bool
"""
- return server_no_results and client_no_results
+ return server_defer_fail and client_defer_fail
Code Review / csit.git / blobdiff