TCP协议如何测试？
TCP协议测试简介TCP（Transmission Control Protocol传输控制协议）是一种面向连接的、可靠的、基于字节流的传输层通信协议。
在Python中一般用socket库来创建tcp协议传输。
在部分测试中，我们可能需要模拟tcp服务端或者客户端进行收发消息，其中以模拟客户端居多。
代码示例：服务端：
import socket
class TcpServer(object):
def __init__(self, port):
self.server_port = port
self.server_ip = socket.gethostbyname(socket.gethostname())
def test_tcp_server(self, listen_num=10):
"""
tcp服务
:param listen_num: 最大等待连接数
:return:
"""
# 建立服务端的套接字对象，第一个参数是ipv4协议，第二个参数是tcp协议
logger.info('开始启动tcp_server')
tcp_server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 给程序绑定端口号
tcp_server_socket.bind((self.server_ip, self.server_port))
# 设置监听服务，等待客户端向服务端发送消
tcp_server_socket.listen(listen_num)
logger.info('tcp_server启动完成')
while True:
conn, client_addr = tcp_server_socket.accept()
t = threading.Thread(target=self.get_socket_connect, args=(conn, client_addr,))
t.start()
def get_socket_connect(self, conn, client_addr):
threading_n = threading.active_count()
logger.info('现在存在的线程数为:%s' % threading_n)
logger.info('连接上了客户端%s:%s' % (client_addr, client_addr))
while True:
try:
# 建立连接，接收消息
if conn:
result = conn.recv(1024)
if result == b'':
break
else:
logger.info(len(result))
logger.info(result)
s = self.analysis_bytes(result)
logger.info(s)
logger.info(type(s))
# logger.info(self)
conn.sendall('收到'.encode())
except Exception as e:
logger.error(e)
break
@staticmethod
def analysis_bytes(value):
try:
result = value.decode('utf-8')
except:
try:
result = value.decode('GBK')
except:
result = value
return result
def close_socket(self):
tcp_server_socket.close()
TcpServer(port=server_port).test_tcp_server()
客户端
客户端如果会关闭再起，那么就不要绑定端口，一般也不需要绑定端口
import time
from time import sleep
from TCP_TEST.read_config import send_num, connect_server_ip, connect_server_port, wait_time, send_type, send_msg_type
import random
from common.common import create_folder
import threading
import struct
class TcpClient(object):
"""
tcp协议方法
"""
def __init__(self):
self.connect_server_ip = connect_server_ip
self.connect_server_port = connect_server_port
self.send_client_num = send_num
def get_tcp_connect(self):
tcp_client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# port = random.randint(3300, 3310)
# print(port)
# tcp_client.bind(('', port))
tcp_client.connect((self.connect_server_ip, self.connect_server_port))
return tcp_client
def test_client(self, lock, logger, bytes_lists, tcp_client):
while True:
if send_type is False:
# lock.acquire()
# self.send_client_num -= 1
# lock.release()
if self.send_client_num < 0:
break
# logger.info('原始字节流组个数：%s' % len(bytes_lists))
if send_msg_type is True:
bytes_list = random.choice(bytes_lists)
if send_type is False:
lock.acquire()
self.send_client_num -= 1
lock.release()
if self.send_client_num < 0:
break
self.send_msg(logger, bytes_list, tcp_client)
else:
for bytes_list in bytes_lists:
if send_type is False:
lock.acquire()
self.send_client_num -= 1
lock.release()
if self.send_client_num < 0:
break
self.send_msg(logger, bytes_list, tcp_client)
def client_recv(self, tcp_client, logger):
"""
接收消息
:param tcp_client:
:param logger:
:return:
"""
recv_num = self.send_client_num
tcp_client.recv(1024)
tcp_client.sendall("/n")
while True:
result = tcp_client.recv(312000)
logger.info("接收的消息为:%s" % str(result[:20]))
filename = "test_" + str(time.time() * 1000) + str(recv_num) + '.dat'
create_folder("../data/kddata")
with open("../data/kddata/" + filename, 'ab') as f:
f.write(result)
f.close()
recv_num -= 1
if recv_num <= 0:
break
@staticmethod
def send_msg(logger, bytes_list, tcp_client):
threading_name = threading.currentThread().name
for bytes_array in bytes_list:
logger.info(len(bytes_array))
logger.info('线程%s 发送消息内容为:%s...' % (threading_name, bytes_array[:24]))
tcp_client.sendall(bytes_array)
sleep(wait_time)
@staticmethod
def close_tcp_client(tcp_client):
"""
关闭客户端
:param tcp_client:
:return:
"""
tcp_client.close()
页:
[1]
