#yyds干货盘点# Socket 编程与测试详细介绍

Posted 2021-11-21 测试玉米君

在Python文档中，socket相关的内容记录在这些页面：

• socket --- 底层网络接口
• socketserver --- 用于简化网络服务端编写的类
• 套接字编程方法

此外还有些资料可供参考：

• Socket application programming interface overview

1. socket 协议的概要

计算机网络：计算机之间的通信，输入输出， IO

文件IO：

1. 打开文件
2. 对文件内容读取
3. 对文件内容写入

进程间通信：

1. 建立管道，conn1 和conn2 ,分别给不同的进程
2. conn1 输入内容，conn2可以输出内容
3. conn2输入内容，conn1 可以输出内容

分布式程序：部署在不同计算机的程序

计算机之间的通信：

1. 建立socket连接，两个计算机分别充当 server和client
2. serves输入内容，client输出内容
3. client输入内容，server输出内容

---

计算机之间的连接并不容易：各个计算机可能使用不同软件、硬件、连接方式、

计算机通信标准：OSI七层模型

核心思想：各层依赖下一层，但是不必关系下一层的细节，各司其职。分工合作

实际的应用：TCP / IP 协议簇，没有完全安装OSI 将模型分为7层，而是4层

OSI 是理论，是知道思想

TCP/IP 是现实，是具体的工具

---

所以在创建socket的时候，有一些选项来说明本次使用的协议具体是什么

常用的两个：

	SOCK_STREAM	SOCK_DGRAM	SOCK_RAW
AF_UNIX	UNIX DOMAIN SOCKET	UNIX DOMAIN SOCKET
AF_INET	IPv4 + TCP	IPv4 + UDP
AF_INET6	IPv6 + TCP	IPv6 + UDP
AF_PACKET			自定义协议

目前 TCP(IPv4) 是主流， SOCK_STREAM + AF_INET

2. 创建和使用Socket

socket.socket 类

- 接收数据的方法
- 发送数据的方法
- 监听端口的方法
- ……

socket函数

• 创建客户端
• 创建服务端
• 创建客户端 + 服务端（已连接）
• ……

2.1 创建socket函数

2.1.1 基本函数

本质就是 socket.socket类的实例化

import socket

s1 = socket.socket(
    type=socket.SOCK_STREAM,
    family=socket.AF_INET,
)  # TCP+ ipv4  实现跨计算机通信

s2 = socket.socket(
    type=socket.SOCK_STREAM,
    family=socket.AF_UNIX,
)  # unix socket 同计算机跨进程通信

2.1.2 便捷函数

1. 创建服务端socket

   1. 创建socket
   2. 绑定地址（ip+port）
   3. 监听地址

   import socket
   
   server_address = ("127.0.0.1", 9001)
   server = socket.create_server(
      server_address, family=socket.AF_INET
   )  # 可以指定IPV4 不该改变类型 TCP
   
   server_address_2 = ("::1", 9002)
   server_2 = socket.create_server(
      server_address_2, family=socket.AF_INET6
   )  #  或者 IPV6，不该改变类型 TCP

2. 创建客户端socket

   1. 创建socket

   2. 绑定地址 （ip+port）

   3. 想指定的服务端发起连接

      client = socket.create_connection(
        server_address, source_address=("127.0.0.1", 9005)
      )  # 为空的会自动分配

3. 创建已连接的socket

   1. 创建两个socket，其中作为server监听端口，另一个作为client连接端口

   2. 连接成功后，把两个socket返回

      s1, s2 = socket.socketpair()
      # 如果是linux 使用AF_UNIX（同计算机跨进行通信）
      # 如果不是linux 使用AF_INTE（同计算机跨进行通信，地址127.0.0.1）

​

2.2 socket对象

socket提供了一些列的方法，完成网络连接 、 数据通信，常用的：

• 网络连接
  • 服务端
  • bind 绑定地址 （IP+PORT）
  • listen 监听端口，允许其他socket发起连接
  • accpet 接收连接，生成新的socket，完成数据收发
  • close 关闭
  • 客户端
  • bind 绑定地址 （IP+PORT）
  • connect 连接指定地址
  • close 关闭
• 数据通信（IO）
  • sendall 发送数据
  • recv 接收数据

2.2.1 配置日志

import logging.config

import yaml

with open("logging.conf.yaml") as f:
    config = yaml.safe_load(f)
    logging.config.dictConfig(config)

logger_server = logging.getLogger("server")
logger_client = logging.getLogger("client")

2.2.2 创建server端

server = socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM)
logger_server.info("socket创建")

server.bind(server_address)
logger_server.info(f"绑定了地址 {server_address}")

server.listen()
logger_server.info(f"监听成功，服务器已就绪")

# ---
# 2.使用便捷函数，快速配置
server = socket.create_server(server_address)  # python3.8+ 
logger_server.info(f"监听{server_address}成功，服务器已就绪")

2.2.3 创建客户端

client = socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM)
logger_client.info(f"socket已创建,准备连接 {server_address=}")
client.connect(server_address)
logger_client.info("连接成功")

# ---
client = socket.create_connection(server_address)
logger_client.info("连接成功")

2.2.4 通信

传递的是二进制数据流，字节流

conn, address = server.accept()  # 服务端接收请求
logger_server.info(f"新的客户端进入： {address=}")

data = conn.recv(1024)
logger_server.info(f"收到数据 {data=}")

conn.sendall(data)  # 发送数据
conn.close()

# ----

data = client.recv(1024)
logger_client.info(f"收到数据 {data=}")

client.close()

2.2.5 测试驱动开发

测试驱动开发 TDD

1. 编写测试用例
2. 执行测试用例 （测试失败）（红）
3. 编写业务代码
4. 执行测试用例 （测试通过）（绿）
5. 重构代码
6. 执行测试用例 （测试通过）（绿）

步骤：

1. 分解被测模块
   1. server
      1. 连接
      2. 通信
      3. 业务功能
   2. client
      1. 连接
      2. 通信
2. 了解被测对象
   1. 获取socket属性、状态
   2. 改变socket属性、状态
   3. 甚至mock
3. 编写测试用例
   1. 使用测试框架
   2. 必要的话，自定义断言

---

驱动模块 ---> 测试模块

测试用例中的client ---> 业务中的server，

业务中的server -----> 业务中client

---

文件结构：

• server.py
• test_server.py # unittest 默认发现规则
• server_test.py # 需要修改发现规则：python -m unittest discover -p *_test.py

3. 内置的socket server

server比较专业、复杂，处理很多事务的细节。

socketserver 预置常见的socketserver：

• TCPServer
• ThreadingTCPServer
• ForkingTCPServer

1. 能处理客户端中断的异常，不会导致server退出
2. 停供了并发的支持
3. 将网络连接、数据处理 分开

StreamRequestHandler

1. 封装工作细节
2. 将IO,封装成了两个（读、写）like file
3. 提供了和文件读写相同方法，完成数据收发
   1. rfile 只读
   2. wfile 只写

4. 基于TDD的socket聊天室

1. ThreadingTCPServer 使用
2. socket服务测试

4.1 需求

多人同时在线，每个人都可以发言，发言内容可以被每个人收到

4.2 协议

• 传输协议
  • 使用TCP进行传输
  • 使用换行符作"\\n"为结束标记
• 数据格式
  • name 昵称
  • msg 内容
  • time 时间
  • {"name": "张三", "msg":"你好", "time": "2020-11-1"} (使用json字符串)

【客户端】 --{\'name\':\'张三\'}-- {"name": "张三", "msg":"你好", "time": "2020-11-1"} ---{"name": "张三", }\\n ----------0101010101------------------------------------>>>>>>【服务端】

4.3 编写socket客户端

1. 建立socket，断开socket
2. 发送消息能力
   1. 确定要发送的内容
   2. 组装固定格式的字典
   3. 讲字段转字符串
   4. 字符串 + ”\\n"
   5. 字符串 转字节流
   6. 借助socket发送字节流
3. 接收消息能力
   1. 从socket读取字节流
   2. 字节流转字符串
   3. 去掉结束标记
   4. 字符串转字典
   5. 返回字典

4.4 编写测试用例

1. 服务端是否就绪
   1. 多线程
   2. 多进程
   3. 子进程
2. 服务器能否关闭
   1. 多进程
3. 测试用例
   1. 服务器可用
   2. 服务器多人可用
   3. 服务器可以回复消息
   4. 服务器可以对多个用户广播消息
   5. 服务器在用户掉线后，依然正常

4.5 编写服务端

    import socketserver

    class MyUDPHandler(socketserver.StreamRequestHandler):
        """
        This class works similar to the TCP handler class, except that
        self.request consists of a pair of data and client socket, and since
        there is no connection the client address must be given explicitly
        when sending data back via sendto().
        """

        def handle(self):
            while True:
                data = self.rfile.readline()
                print(f"收到客户端数据 {data=}")

    with socketserver.TCPServer(("127.0.0.1", 9002), MyUDPHandler) as server:
        server.serve_forever()