mirror of
https://github.com/espressif/esp-idf.git
synced 2024-10-05 20:47:46 -04:00
252 lines
9.8 KiB
Python
Executable File
252 lines
9.8 KiB
Python
Executable File
import random
|
|
import socket
|
|
import threading
|
|
|
|
from NativeLog import NativeLog
|
|
|
|
# from NativeLog import NativeLog
|
|
|
|
# make sure target do not listen on this port
|
|
ERROR_PORT = 62685
|
|
|
|
|
|
class Utility(object):
|
|
METHOD_RESULT = {"C_01": ("ESTABLISHED", "ESTABLISHED"), # target TCP peer state, PC TCP peer state
|
|
"C_02": ("SYNC_SENT", "CLOSED"),
|
|
"C_03": ("CLOSED", "CLOSED"),
|
|
"C_04": ("SYN_RCVD", "ESTABLISHED"),
|
|
"C_05": ("ESTABLISHED", "ESTABLISHED"),
|
|
"C_06": ("CLOSED", "CLOSED"),
|
|
"C_07": ("CLOSED", "CLOSED"),
|
|
"C_08": ("CLOSED", "CLOSED"),
|
|
"D_01": ("TIME_WAIT", "CLOSED"),
|
|
"D_02": ("TIME_WAIT", "TIME_WAIT"),
|
|
"D_03": ("FIN_WAIT_2", "CLOSE_WAIT"),
|
|
"D_04": ("FIN_WAIT_1", "CLOSE_WAIT"),
|
|
"D_05": ("CLOSED", "TIME_WAIT"),
|
|
"D_06": ("CLOSED", "CLOSED"),
|
|
"D_07": ("CLOSE_WAIT", "FIN_WAIT2"),
|
|
"D_08": ("TIME_WAIT", "CLOSED"), }
|
|
|
|
SOC_CLOSED_STATE = ("FIN_WAIT_1", "FIN_WAIT_2", "CLOSING", "TIME_WAIT", "LAST_ACK", "CLOSED")
|
|
SOC_CREATED_STATE = ("SYNC_RCVD", "ESTABLISHED")
|
|
SOC_SEND_DATA_STATE = ("ESTABLISHED", "CLOSE_WAIT")
|
|
SOC_ESTABLISHED_STATE = ("ESTABLISHED", )
|
|
|
|
def __init__(self, tc_action, pc_server_port, target_server_port, pc_ip, target_ip):
|
|
self.tc_action = tc_action
|
|
self.pc_server_port = pc_server_port
|
|
self.target_server_port = target_server_port
|
|
self.pc_ip = pc_ip
|
|
self.target_ip = target_ip
|
|
self.pc_close_wait_socket_list = []
|
|
self.sync_lock = threading.Lock()
|
|
pass
|
|
|
|
# create a tcp socket, return True or False
|
|
def __create_tcp_socket(self, connection):
|
|
connection_handler = connection["Connection handler"]
|
|
connection["Target port"] = random.randint(10000, 60000)
|
|
connection_handler.add_checkers("BIND:(\d+),OK,%s,%s"
|
|
% (self.target_ip, connection["Target port"]))
|
|
self.tc_action.send_ssc_command("soc -B -t TCP -i %s -p %s" % (self.target_ip, connection["Target port"]))
|
|
serial_result, tcp_result = connection_handler.get_checker_results()
|
|
if serial_result is not None:
|
|
connection["Target socket id"] = serial_result.group(1)
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
# target do connect, return True or False
|
|
def __target_do_connect(self, connection, dest_ip, dest_port, timeout=20):
|
|
connection_handler = connection["Connection handler"]
|
|
connection_handler.add_checkers("CONNECT:%s,OK" % connection["Target socket id"],
|
|
connection["Target port"])
|
|
self.tc_action.send_ssc_command("soc -C -s %s -i %s -p %s"
|
|
% (connection["Target socket id"], dest_ip, dest_port))
|
|
serial_result, tcp_result = connection_handler.get_checker_results(timeout)
|
|
if serial_result is not None and tcp_result is not None:
|
|
connection["PC socket"] = tcp_result
|
|
return True
|
|
else:
|
|
return False
|
|
pass
|
|
|
|
# pc do connect, return True or False
|
|
def __pc_do_connect(self, connection, dest_ip, dest_port, timeout=20):
|
|
connection_handler = connection["Connection handler"]
|
|
sock = socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM)
|
|
while True:
|
|
connection["PC port"] = random.randint(10000, 60000)
|
|
try:
|
|
sock.bind((self.pc_ip, connection["PC port"]))
|
|
break
|
|
except socket.error, e:
|
|
if e.errno == 10048: # socket port reuse
|
|
continue
|
|
sock.settimeout(timeout)
|
|
connection["PC socket"] = sock
|
|
connection_handler.add_checkers("ACCEPT:(\d+),\d+,%s,%s"
|
|
% (self.pc_ip, connection["PC port"]))
|
|
try:
|
|
sock.connect((dest_ip, dest_port))
|
|
except socket.error:
|
|
pass
|
|
serial_result, tcp_result = connection_handler.get_checker_results()
|
|
if serial_result is not None:
|
|
connection["Target socket id"] = serial_result.group(1)
|
|
return True
|
|
else:
|
|
return False
|
|
pass
|
|
|
|
def connect_c_01(self, connection):
|
|
if self.__create_tcp_socket(connection) is True:
|
|
return self.__target_do_connect(connection, self.pc_ip, self.pc_server_port)
|
|
else:
|
|
return False
|
|
|
|
def connect_c_02(self, connection):
|
|
if self.__create_tcp_socket(connection) is True:
|
|
return not self.__target_do_connect(connection, self.pc_ip, ERROR_PORT, timeout=5)
|
|
else:
|
|
return False
|
|
|
|
def connect_c_03(self, connection):
|
|
return False
|
|
|
|
def connect_c_04(self, connection):
|
|
return False
|
|
|
|
def connect_c_05(self, connection):
|
|
return self.__pc_do_connect(connection, self.target_ip, self.target_server_port)
|
|
|
|
def connect_c_06(self, connection):
|
|
return False
|
|
|
|
def connect_c_07(self, connection):
|
|
return not self.__pc_do_connect(connection, self.target_ip, ERROR_PORT)
|
|
|
|
def connect_c_08(self, connection):
|
|
return False
|
|
|
|
def __target_socket_close(self, connection):
|
|
connection_handler = connection["Connection handler"]
|
|
if connection["Target socket id"] is not None:
|
|
connection_handler.add_checkers("CLOSE:%s" % connection["Target socket id"])
|
|
self.tc_action.send_ssc_command("soc -T -s %s" % connection["Target socket id"])
|
|
serial_result, tcp_result = connection_handler.get_checker_results()
|
|
connection["Target socket id"] = None
|
|
else:
|
|
serial_result = None
|
|
return True if serial_result is not None else False
|
|
|
|
@staticmethod
|
|
def __pc_socket_close(connection):
|
|
connection_handler = connection["Connection handler"]
|
|
if connection["PC socket"] is not None:
|
|
connection_handler.add_checkers("CLOSED:%s" % connection["Target socket id"])
|
|
connection["PC socket"].close()
|
|
serial_result, tcp_result = connection_handler.get_checker_results()
|
|
connection["PC socket"] = None
|
|
else:
|
|
serial_result = None
|
|
return True if serial_result is not None else False
|
|
|
|
def close_d_01(self, connection):
|
|
connection["PC socket"] = None
|
|
return self.__target_socket_close(connection)
|
|
|
|
def close_d_02(self, connection):
|
|
pass
|
|
|
|
def close_d_03(self, connection):
|
|
with self.sync_lock:
|
|
self.pc_close_wait_socket_list.append(connection["PC socket"])
|
|
return self.__target_socket_close(connection)
|
|
pass
|
|
|
|
def close_d_04(self, connection):
|
|
pass
|
|
|
|
def close_d_05(self, connection):
|
|
return self.__pc_socket_close(connection)
|
|
|
|
def close_d_06(self, connection):
|
|
# target send data to PC, PC don't recv and close socket
|
|
connection_handler = connection["Connection handler"]
|
|
connection_handler.add_checkers("SEND:%s,OK" % connection["Target socket id"])
|
|
self.tc_action.send_ssc_command("soc -S -s %s -l 100" % connection["Target socket id"])
|
|
serial_result, tcp_result = connection_handler.get_checker_results()
|
|
if serial_result is None:
|
|
return False
|
|
return self.__pc_socket_close(connection)
|
|
|
|
def close_d_07(self, connection):
|
|
# PC shutdown WR
|
|
result = False
|
|
try:
|
|
connection["PC socket"].shutdown(socket.SHUT_WR)
|
|
result = True
|
|
except StandardError:
|
|
pass
|
|
return result
|
|
|
|
def close_d_08(self, connection):
|
|
pass
|
|
|
|
def close_connection(self, connection):
|
|
self.__target_socket_close(connection)
|
|
pass
|
|
|
|
TCP_ACTION_DICT = {"C_01": connect_c_01,
|
|
"C_02": connect_c_02,
|
|
"C_03": connect_c_03,
|
|
"C_04": connect_c_04,
|
|
"C_05": connect_c_05,
|
|
"C_06": connect_c_06,
|
|
"C_07": connect_c_07,
|
|
"C_08": connect_c_08,
|
|
"D_01": close_d_01,
|
|
"D_02": close_d_02,
|
|
"D_03": close_d_03,
|
|
"D_04": close_d_04,
|
|
"D_05": close_d_05,
|
|
"D_06": close_d_06,
|
|
"D_07": close_d_07,
|
|
"D_08": close_d_08,
|
|
}
|
|
|
|
def get_method_destination_state(self, method):
|
|
return self.METHOD_RESULT[method]
|
|
|
|
def execute_tcp_method(self, method, connection):
|
|
if method in self.METHOD_RESULT:
|
|
result = self.TCP_ACTION_DICT[method](self, connection)
|
|
if result is True:
|
|
state = self.get_method_destination_state(method)
|
|
connection["Target state"] = state[0]
|
|
connection["PC state"] = state[1]
|
|
else:
|
|
NativeLog.add_prompt_trace("[TCPConnection] tcp method %s fail, connection is %s"
|
|
% (method, connection))
|
|
NativeLog.add_trace_info("[TCPConnection][data cache][check item %s] %s"
|
|
% (connection["Connection handler"].serial_check_item_id,
|
|
connection["Connection handler"].data_cache))
|
|
else:
|
|
raise StandardError("Not TCP connection method")
|
|
return result
|
|
|
|
def is_established_state(self, connection):
|
|
return True if connection["Target state"] in self.SOC_CREATED_STATE else False
|
|
|
|
def is_closed_state(self, connection):
|
|
return True if connection["Target state"] in self.SOC_CLOSED_STATE else False
|
|
|
|
|
|
def main():
|
|
pass
|
|
|
|
if __name__ == '__main__':
|
|
main()
|