标签：协议 socket self send TCP server client 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[0], client_addr[1]))
        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[11:])
                        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()

标签：协议,socket,self,send,TCP,server,client,tcp,测试
来源： https://www.cnblogs.com/caodingzheng/p/16675965.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。