You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何使用pytest-mock模拟只读方法?——Socket通信客户端与服务端测试问题

如何使用pytest-mock模拟只读方法?——Socket通信客户端与服务端测试问题

嘿,我看了你这个模拟socket的思路,其实已经摸到门道了!针对socket里的只读属性和方法,我们可以用pytest-mock的PropertyMock搞定只读属性,再结合side_effect绑定自定义逻辑来模拟recv这类方法,刚好能解决你不想搭真实连接的测试需求。

先给你补全并优化一下你当前的代码,顺便把核心问题点讲清楚:

from collections import deque
from threading import Lock
import socket
from time import sleep
from typing import Deque

import pytest
from pytest_mock import MockFixture

from comms import Comms


def mocked_socket(mocker: MockFixture):
    # 用spec参数创建mock,让它继承真实socket的结构,避免模拟出不符合规范的方法
    mock_sock = mocker.MagicMock(spec=socket.socket)
    backing_store: Deque[int] = deque()
    lock = Lock()

    # 用PropertyMock模拟只读的timeout属性
    timeout_prop = mocker.PropertyMock(return_value=1.0)
    type(mock_sock).timeout = timeout_prop

    def receiver(n: int):
        with lock:
            bl = len(backing_store)
            if bl == 0 and mock_sock.timeout is not None:
                # 释放锁让其他线程能写入数据,模拟真实socket的超时等待
                lock.release()
                sleep(mock_sock.timeout)
                lock.acquire()
                # 超时后仍无数据,抛出真实的socket超时异常
                if len(backing_store) == 0:
                    raise socket.timeout
            # 从缓存中读取指定长度的数据
            data = bytearray()
            while len(data) < n and backing_store:
                data.append(backing_store.popleft())
            return bytes(data)

    def sender(data: bytes):
        with lock:
            backing_store.extend(data)
            return len(data)

    # 把自定义的收发逻辑绑定到mock的方法上
    mock_sock.recv.side_effect = receiver
    mock_sock.sendall.side_effect = sender

    return mock_sock

然后在测试用例里,你可以这样用这个mock的socket:

def test_comms_send_recv(mocker):
    # 生成我们的mock socket
    mock_sock = mocked_socket(mocker)
    # 替换Comms类里用到的真实socket
    mocker.patch("comms.socket.socket", return_value=mock_sock)
    
    # 正常初始化并测试Comms逻辑
    comms = Comms()
    test_data = b"hello mock socket"
    comms.send(test_data)
    received_data = comms.recv(1024)
    
    assert received_data == test_data

这里几个核心的处理点给你划重点:

  • 针对timeout这类只读属性:必须用PropertyMock来模拟,直接赋值会因为属性只读报错,用type(mock_sock).timeout把属性绑定到mock实例的类型上,才能正确模拟只读行为
  • 锁的处理:你之前的代码里释放锁后没有重新获取,会导致锁状态异常,我补了lock.acquire(),保证线程安全
  • 模拟真实socket行为:超时后要抛出socket.timeout异常,这样你的业务代码里的超时处理逻辑也能被覆盖测试
  • 用spec=socket.socket:能让mock的socket和真实socket的方法签名、属性结构完全对齐,避免测试出现“假通过”的情况

备注:内容来源于stack exchange,提问作者Joku

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.04.15 11:53:07