From a39e4416cd78fc3b147695465c4de1c896b3face Mon Sep 17 00:00:00 2001 From: Dan Klein Date: Tue, 27 Oct 2015 10:00:18 +0200 Subject: more hltapi progress connect working --- .../trex_control_plane/client/trex_hltapi.py | 72 +++++++- .../client/trex_stateless_client.py | 185 ++++++++++++++++----- .../client_utils/jsonrpc_client.py | 27 +-- 3 files changed, 227 insertions(+), 57 deletions(-) (limited to 'scripts/automation') diff --git a/scripts/automation/trex_control_plane/client/trex_hltapi.py b/scripts/automation/trex_control_plane/client/trex_hltapi.py index 7453d8ec..6194d376 100755 --- a/scripts/automation/trex_control_plane/client/trex_hltapi.py +++ b/scripts/automation/trex_control_plane/client/trex_hltapi.py @@ -4,14 +4,54 @@ import trex_root_path from client_utils.packet_builder import CTRexPktBuilder from trex_stateless_client import CTRexStatelessClient -print "done!" class CTRexHltApi(object): def __init__(self): + self.trex_client = None + self.connected = False + + pass + + # ----- session functions ----- # + + def connect(self, device, port_list, username, port=5050, reset=False, break_locks=False): + ret_dict = {"status": 0} + self.trex_client = CTRexStatelessClient(username, device, port) + res_ok, msg = self.trex_client.connect() + if not res_ok: + self.trex_client = None + ret_dict.update({"log": msg}) + return ret_dict + # arrived here, connection successfully created with server + # next, try acquiring ports of TRex + port_list = self.parse_port_list(port_list) + response = self.trex_client.acquire(port_list, force=break_locks) + res_ok, log = CTRexHltApi.process_response(port_list, response) + if not res_ok: + self.trex_client.disconnect() + self.trex_client = None + ret_dict.update({"log": log}) + # TODO: should revert taken ports?! + return ret_dict + # arrived here, all desired ports were successfully acquired + print log + if reset: + # remove all port traffic configuration from TRex + response = self.trex_client.remove_all_streams(port_list) + res_ok, log = CTRexHltApi.process_response(port_list, response) + if not res_ok: + self.trex_client.disconnect() + self.trex_client = None + ret_dict.update({"log": log}) + return ret_dict + print log + ret_dict.update({"status": 1}) + self.trex_client.disconnect() + pass - def connect(self, device, port_list, username, reset=False, break_locks=False): + def cleanup_session(self, port_list, maintain_lock=False): pass def interface_config(self, port_handle, mode="config"): @@ -41,8 +81,32 @@ class CTRexHltApi(object): def get_stream_stats(self, port_handle): return self.traffic_stats(port_handle, mode="stream") - - + # ----- internal functions ----- # + @staticmethod + def process_response(port_list, response): + if isinstance(port_list, list): + res_ok, response = response + log = CTRexHltApi.join_batch_response(response) + else: + res_ok = response.success + log = str(response) + return res_ok, log + + @staticmethod + def parse_port_list(port_list): + if isinstance(port_list, str): + return [int(port) + for port in port_list.split()] + elif isinstance(port_list, list): + return [int(port) + for port in port_list] + else: + return port_list + + @staticmethod + def join_batch_response(responses): + return "\n".join([str(response) + for response in responses]) if __name__ == "__main__": diff --git a/scripts/automation/trex_control_plane/client/trex_stateless_client.py b/scripts/automation/trex_control_plane/client/trex_stateless_client.py index b976db66..97d3ec0a 100755 --- a/scripts/automation/trex_control_plane/client/trex_stateless_client.py +++ b/scripts/automation/trex_control_plane/client/trex_stateless_client.py @@ -13,10 +13,19 @@ from common.trex_stats import * from common.trex_streams import * from collections import namedtuple +RpcCmdData = namedtuple('RpcCmdData', ['method', 'params']) + +class RpcResponseStatus(namedtuple('RpcResponseStatus', ['success', 'id', 'msg'])): + __slots__ = () + def __str__(self): + return "{id:^3} - {msg} ({stat})".format(id=self.id, + msg=self.msg, + stat="success" if self.success else "fail") + +# RpcResponseStatus = namedtuple('RpcResponseStatus', ['success', 'id', 'msg']) class CTRexStatelessClient(object): """docstring for CTRexStatelessClient""" - RpcCmdData = namedtuple('RpcCmdData', ['method', 'params']) def __init__(self, username, server="localhost", port=5050, virtual=False): super(CTRexStatelessClient, self).__init__() @@ -26,48 +35,69 @@ class CTRexStatelessClient(object): self._active_ports = set() self._stats = CTRexStatsManager("port", "stream") self._system_info = None + self._server_version = None + self.__err_log = None # ----- decorator methods ----- # def force_status(owned=True, active_and_owned=False): def wrapper(func): def wrapper_f(self, *args, **kwargs): port_ids = kwargs.get("port_id") + if not port_ids: + port_ids = args[0] if isinstance(port_ids, int): # make sure port_ids is a list port_ids = [port_ids] bad_ids = set() for port_id in port_ids: - port_owned = self._conn_handler.get(kwargs.get(port_id)) + port_owned = self._conn_handler.get(port_id) if owned and not port_owned: - bad_ids.add(port_ids) + bad_ids.add(port_id) elif active_and_owned: # stronger condition than just owned, hence gets precedence if port_owned and port_id in self._active_ports: continue else: - bad_ids.add(port_ids) + bad_ids.add(port_id) else: continue if bad_ids: # Some port IDs are not according to desires status raise RuntimeError("The requested method ('{0}') cannot be invoked since port IDs {1} are not" - "at allowed stated".format(func.__name__)) + "at allowed stated".format(func.__name__, list(bad_ids))) else: - func(self, *args, **kwargs) + return func(self, *args, **kwargs) return wrapper_f return wrapper @property def system_info(self): if not self._system_info: - self._system_info = self.get_system_info() - return self._system_info + rc, info = self.get_system_info() + if rc: + self._system_info = info + else: + self.__err_log = info + return self._system_info if self._system_info else "Unknown" + + @property + def server_version(self): + if not self._server_version: + rc, ver_info = self.get_version() + if rc: + self._server_version = ver_info + else: + self.__err_log = ver_info + return self._server_version if self._server_version else "Unknown" # ----- user-access methods ----- # def connect(self): - self.comm_link.connect() + rc, err = self.comm_link.connect() + if not rc: + return rc, err + return self._init_sync() def disconnect(self): - self.comm_link.disconnect() + return self.comm_link.disconnect() def ping(self): return self.transmit("ping") @@ -90,18 +120,19 @@ class CTRexStatelessClient(object): if isinstance(port_id, list) or isinstance(port_id, set): # handle as batch mode port_ids = set(port_id) # convert to set to avoid duplications - commands = [self.RpcCmdData("acquire", {"port_id": p_id, "user": self.user, "force": force}) + commands = [RpcCmdData("acquire", {"port_id": p_id, "user": self.user, "force": force}) for p_id in port_ids] rc, resp_list = self.transmit_batch(commands) if rc: - self._process_batch_result(commands, resp_list, self._handle_acquire_response) + return self._process_batch_result(commands, resp_list, self._handle_acquire_response) else: params = {"port_id": port_id, "user": self.user, "force": force} - command = self.RpcCmdData("acquire", params) - self._handle_acquire_response(command, self.transmit(command.method, command.params)) - return self._conn_handler.get(port_id) + command = RpcCmdData("acquire", params) + return self._handle_acquire_response(command, + self.transmit(command.method, command.params), + self.default_success_test) @force_status(owned=True) def release(self, port_id=None): @@ -110,7 +141,7 @@ class CTRexStatelessClient(object): if isinstance(port_id, list) or isinstance(port_id, set): # handle as batch mode port_ids = set(port_id) # convert to set to avoid duplications - commands = [self.RpcCmdData("release", {"handler": self._conn_handler.get(p_id), "port_id": p_id}) + commands = [RpcCmdData("release", {"handler": self._conn_handler.get(p_id), "port_id": p_id}) for p_id in port_ids] rc, resp_list = self.transmit_batch(commands) if rc: @@ -119,7 +150,7 @@ class CTRexStatelessClient(object): self._conn_handler.pop(port_id) params = {"handler": self._conn_handler.get(port_id), "port_id": port_id} - command = self.RpcCmdData("release", params) + command = RpcCmdData("release", params) self._handle_release_response(command, self.transmit(command.method, command.params)) return @@ -143,6 +174,28 @@ class CTRexStatelessClient(object): "stream_id": stream_id} return self.transmit("remove_stream", params) + @force_status(owned=True) + def remove_all_streams(self, port_id=None): + if not self._is_ports_valid(port_id): + raise ValueError("Provided illegal port id input") + if isinstance(port_id, list) or isinstance(port_id, set): + # handle as batch mode + port_ids = set(port_id) # convert to set to avoid duplications + commands = [RpcCmdData("remove_all_streams", {"port_id": p_id, "handler": self._conn_handler.get(p_id)}) + for p_id in port_ids] + rc, resp_list = self.transmit_batch(commands) + if rc: + return self._process_batch_result(commands, resp_list, self._handle_remove_streams_response, + success_test=self.ack_success_test) + else: + params = {"port_id": port_id, + "handler": self._conn_handler.get(port_id)} + command = RpcCmdData("remove_all_streams", params) + return self._handle_remove_streams_response(command, + self.transmit(command.method, command.params), + self.ack_success_test) + pass + @force_status(owned=True, active_and_owned=True) def get_stream_id_list(self, port_id=None): if not self._is_ports_valid(port_id): @@ -167,7 +220,7 @@ class CTRexStatelessClient(object): if isinstance(port_id, list) or isinstance(port_id, set): # handle as batch mode port_ids = set(port_id) # convert to set to avoid duplications - commands = [self.RpcCmdData("start_traffic", {"handler": self._conn_handler.get(p_id), "port_id": p_id}) + commands = [RpcCmdData("start_traffic", {"handler": self._conn_handler.get(p_id), "port_id": p_id}) for p_id in port_ids] rc, resp_list = self.transmit_batch(commands) if rc: @@ -175,7 +228,7 @@ class CTRexStatelessClient(object): else: params = {"handler": self._conn_handler.get(port_id), "port_id": port_id} - command = self.RpcCmdData("start_traffic", params) + command = RpcCmdData("start_traffic", params) self._handle_start_traffic_response(command, self.transmit(command.method, command.params)) return @@ -186,7 +239,7 @@ class CTRexStatelessClient(object): if isinstance(port_id, list) or isinstance(port_id, set): # handle as batch mode port_ids = set(port_id) # convert to set to avoid duplications - commands = [self.RpcCmdData("stop_traffic", {"handler": self._conn_handler.get(p_id), "port_id": p_id}) + commands = [RpcCmdData("stop_traffic", {"handler": self._conn_handler.get(p_id), "port_id": p_id}) for p_id in port_ids] rc, resp_list = self.transmit_batch(commands) if rc: @@ -194,12 +247,12 @@ class CTRexStatelessClient(object): else: params = {"handler": self._conn_handler.get(port_id), "port_id": port_id} - command = self.RpcCmdData("stop_traffic", params) + command = RpcCmdData("stop_traffic", params) self._handle_start_traffic_response(command, self.transmit(command.method, command.params)) return def get_global_stats(self): - command = self.RpcCmdData("get_global_stats", {}) + command = RpcCmdData("get_global_stats", {}) return self._handle_get_global_stats_response(command, self.transmit(command.method, command.params)) # return self.transmit("get_global_stats") @@ -210,7 +263,7 @@ class CTRexStatelessClient(object): if isinstance(port_id, list) or isinstance(port_id, set): # handle as batch mode port_ids = set(port_id) # convert to set to avoid duplications - commands = [self.RpcCmdData("get_port_stats", {"handler": self._conn_handler.get(p_id), "port_id": p_id}) + commands = [RpcCmdData("get_port_stats", {"handler": self._conn_handler.get(p_id), "port_id": p_id}) for p_id in port_ids] rc, resp_list = self.transmit_batch(commands) if rc: @@ -218,7 +271,7 @@ class CTRexStatelessClient(object): else: params = {"handler": self._conn_handler.get(port_id), "port_id": port_id} - command = self.RpcCmdData("get_port_stats", params) + command = RpcCmdData("get_port_stats", params) return self._handle_get_port_stats_response(command, self.transmit(command.method, command.params)) @force_status(owned=True, active_and_owned=True) @@ -228,7 +281,7 @@ class CTRexStatelessClient(object): if isinstance(port_id, list) or isinstance(port_id, set): # handle as batch mode port_ids = set(port_id) # convert to set to avoid duplications - commands = [self.RpcCmdData("get_stream_stats", {"handler": self._conn_handler.get(p_id), "port_id": p_id}) + commands = [RpcCmdData("get_stream_stats", {"handler": self._conn_handler.get(p_id), "port_id": p_id}) for p_id in port_ids] rc, resp_list = self.transmit_batch(commands) if rc: @@ -236,10 +289,18 @@ class CTRexStatelessClient(object): else: params = {"handler": self._conn_handler.get(port_id), "port_id": port_id} - command = self.RpcCmdData("get_stream_stats", params) + command = RpcCmdData("get_stream_stats", params) return self._handle_get_stream_stats_response(command, self.transmit(command.method, command.params)) # ----- internal methods ----- # + def _init_sync(self): + # get server version and system info + err = False + if self.server_version == "Unknown" or self.system_info == "Unknown": + self.disconnect() + return False, self.__err_log + return True, "" + def transmit(self, method_name, params={}): return self.comm_link.transmit(method_name, params) @@ -265,16 +326,46 @@ class CTRexStatelessClient(object): else: return False + @staticmethod + def ack_success_test(result_obj): + if result_obj.success and result_obj.data == "ACK": + return True + else: + return False + + # ----- handler internal methods ----- # - def _handle_acquire_response(self, request, response): - if response.success: - self._conn_handler[request.get("port_id")] = response.data + def _handle_general_response(self, request, response, msg, success_test=None): + port_id = request.params.get("port_id") + if not success_test: + success_test = self.default_success_test + if success_test(response): + self._conn_handler[port_id] = response.data + return RpcResponseStatus(True, port_id, msg) + else: + return RpcResponseStatus(False, port_id, response.data) + - def _handle_release_response(self, request, response): + def _handle_acquire_response(self, request, response, success_test): + port_id = request.params.get("port_id") + if success_test(response): + self._conn_handler[port_id] = response.data + return RpcResponseStatus(True, port_id, "Acquired") + else: + return RpcResponseStatus(False, port_id, response.data) + + def _handle_remove_streams_response(self, request, response, success_test): + port_id = request.params.get("port_id") + if success_test(response): + return RpcResponseStatus(True, port_id, "Removed") + else: + return RpcResponseStatus(False, port_id, response.data) + + def _handle_release_response(self, request, response, success_test): if response.success: del self._conn_handler[request.get("port_id")] - def _handle_start_traffic_response(self, request, response): + def _handle_start_traffic_response(self, request, response, success_test): if response.success: self._active_ports.add(request.get("port_id")) @@ -282,19 +373,19 @@ class CTRexStatelessClient(object): if response.success: self._active_ports.remove(request.get("port_id")) - def _handle_get_global_stats_response(self, request, response): + def _handle_get_global_stats_response(self, request, response, success_test): if response.success: return CGlobalStats(**response.success) else: return False - def _handle_get_port_stats_response(self, request, response): + def _handle_get_port_stats_response(self, request, response, success_test): if response.success: return CPortStats(**response.success) else: return False - def _handle_get_stream_stats_response(self, request, response): + def _handle_get_stream_stats_response(self, request, response, success_test): if response.success: return CStreamStats(**response.success) else: @@ -311,13 +402,21 @@ class CTRexStatelessClient(object): return False def _process_batch_result(self, req_list, resp_list, handler_func=None, success_test=default_success_test): + res_ok = True + responses = [] + if isinstance(success_test, staticmethod): + success_test = success_test.__func__ for i, response in enumerate(resp_list): - # testing each result with success test so that a conclusion report could be deployed in future. - if success_test(response): - # run handler method with its params - handler_func(req_list[i], response) - else: - continue # TODO: mark in this case somehow the bad result + # run handler method with its params + processed_response = handler_func(req_list[i], response, success_test) + responses.append(processed_response) + if not processed_response.success: + res_ok = False + # else: + # res_ok = False # TODO: mark in this case somehow the bad result + # print res_ok + # print responses + return res_ok, responses # ------ private classes ------ # @@ -332,7 +431,11 @@ class CTRexStatelessClient(object): def connect(self): if not self.virtual: - self.rpc_link.connect() + return self.rpc_link.connect() + + def disconnect(self): + if not self.virtual: + self.rpc_link.disconnect() def transmit(self, method_name, params={}): if self.virtual: diff --git a/scripts/automation/trex_control_plane/client_utils/jsonrpc_client.py b/scripts/automation/trex_control_plane/client_utils/jsonrpc_client.py index 1c5dbe7e..fe94e5ef 100755 --- a/scripts/automation/trex_control_plane/client_utils/jsonrpc_client.py +++ b/scripts/automation/trex_control_plane/client_utils/jsonrpc_client.py @@ -6,6 +6,9 @@ import json import general_utils import re from time import sleep +from collections import namedtuple + +CmdResponse = namedtuple('CmdResponse', ['success', 'data']) class bcolors: BLUE = '\033[94m' @@ -23,12 +26,12 @@ class BatchMessage(object): self.rpc_client = rpc_client self.batch_list = [] - def add (self, method_name, params = {}): + def add (self, method_name, params={}): id, msg = self.rpc_client.create_jsonrpc_v2(method_name, params, encode = False) self.batch_list.append(msg) - def invoke (self, block = False): + def invoke(self, block = False): if not self.rpc_client.connected: return False, "Not connected to server" @@ -36,9 +39,9 @@ class BatchMessage(object): rc, resp_list = self.rpc_client.send_raw_msg(msg, block = False) if len(self.batch_list) == 1: - return True, [(rc, resp_list)] + return CmdResponse(True, [CmdResponse(rc, resp_list)]) else: - return rc, resp_list + return CmdResponse(rc, resp_list) # JSON RPC v2.0 client @@ -47,7 +50,7 @@ class JsonRpcClient(object): def __init__ (self, default_server, default_port): self.verbose = False self.connected = False - + # default values self.port = default_port self.server = default_server @@ -73,7 +76,7 @@ class JsonRpcClient(object): # float pretty_str = re.sub(r'([ ]*:[ ]+)(\-?[1-9][0-9]*\.[0-9]+)',r'\1{0}\2{1}'.format(bcolors.MAGENTA, bcolors.ENDC), pretty_str) # strings - + pretty_str = re.sub(r'([ ]*:[ ]+)("[^"]*")',r'\1{0}\2{1}'.format(bcolors.RED, bcolors.ENDC), pretty_str) pretty_str = re.sub(r"('[^']*')", r'{0}\1{1}'.format(bcolors.MAGENTA, bcolors.RED), pretty_str) except : @@ -115,7 +118,7 @@ class JsonRpcClient(object): return self.send_raw_msg(msg, block) - + # low level send of string message def send_raw_msg (self, msg, block = False): self.verbose_msg("Sending Request To Server:\n\n" + self.pretty_json(msg) + "\n") @@ -127,7 +130,7 @@ class JsonRpcClient(object): self.socket.send(msg, flags = zmq.NOBLOCK) except zmq.error.ZMQError as e: self.disconnect() - return False, "Failed To Get Send Message" + return CmdResponse(False, "Failed To Get Send Message") got_response = False @@ -145,7 +148,7 @@ class JsonRpcClient(object): if not got_response: self.disconnect() - return False, "Failed To Get Server Response" + return CmdResponse(False, "Failed To Get Server Response") self.verbose_msg("Server Response:\n\n" + self.pretty_json(response) + "\n") @@ -159,13 +162,13 @@ class JsonRpcClient(object): for single_response in response_json: rc, msg = self.process_single_response(single_response) - rc_list.append( (rc, msg) ) + rc_list.append( CmdResponse(rc, msg) ) - return True, rc_list + return CmdResponse(True, rc_list) else: rc, msg = self.process_single_response(response_json) - return rc, msg + return CmdResponse(rc, msg) def process_single_response (self, response_json): -- cgit 1.2.3-korg