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

如何测试另一个线程内部发起的URL调用是否执行

这个问题我之前也碰到过——requests-mock的默认行为是只捕获当前线程发起的HTTP请求,而你的SimpleCounter逻辑跑在独立的子线程里,所以主线程里设置的mock根本抓不到子线程的调用。给你两种解决方案,第一种是更推荐的测试友好型设计,第二种是不用改原有代码的快速解决方式:

方案一:依赖注入(推荐,更优雅的测试设计)

核心思路是把硬编码的requests.get抽成可传入的依赖,这样测试时可以替换成一个mock函数,直接验证它的调用情况,完全不需要处理HTTP请求的mock。

修改后的待测试代码:

import threading
import time
import requests

class SimpleCounter(threading.Thread):
    def __init__(self, alarm_func=None):
        super().__init__()
        self.must_stop = False
        self.value = 0
        # 默认使用真实的requests.get,测试时传入mock函数
        self.alarm_func = alarm_func or requests.get

    def increment(self):
        self.value += 1

    def run(self):
        limit = 5
        while not self.must_stop:
            if self.value > limit:
                self.alarm_func("http://localhost/alarm.cgi")
            time.sleep(1)

对应的测试代码(用unittest.mock就能搞定,不需要额外的HTTP mock库):

from unittest.mock import Mock
import time

def test_simplecounter_alarm_triggered():
    # 创建一个mock函数用来替代requests.get
    mock_alarm = Mock()
    counter = SimpleCounter(alarm_func=mock_alarm)
    counter.start()
    
    # 把计数器加到超过阈值
    for _ in range(6):
        counter.increment()
    
    # 等待线程执行告警请求(线程里有1秒sleep,所以多等一点时间)
    time.sleep(1.5)
    # 停止线程并等待它结束
    counter.must_stop = True
    counter.join()
    
    # 验证mock函数被正确调用
    mock_alarm.assert_called_once_with("http://localhost/alarm.cgi")

这种方式的好处:

  • 解耦了业务逻辑和HTTP请求的依赖,代码灵活性更高
  • 测试不需要依赖网络或HTTP mock,执行速度更快、更可靠

方案二:让requests-mock捕获子线程请求

如果你不想修改原有业务代码,也可以手动把requests-mock的适配器挂载到requests的全局会话上,让它能捕获所有线程的请求:

测试代码示例:

import requests_mock
import time

def test_simplecounterapp():
    counter = SimpleCounter()
    
    with requests_mock.Mocker() as m:
        # 注册要mock的URL
        m.get("http://localhost/alarm.cgi", text="ok")
        # 手动将mock适配器挂载到requests的全局适配器上
        requests.adapters.DEFAULT_ADAPTERS["http://"] = m.adapter
        requests.adapters.DEFAULT_ADAPTERS["https://"] = m.adapter
        
        counter.start()
        for _ in range(6):
            counter.increment()
        
        # 等待线程执行请求
        time.sleep(1.5)
        # 停止线程并等待结束
        counter.must_stop = True
        counter.join()
        
        # 验证请求被调用
        assert m.called
        assert m.call_count == 1
        assert m.last_request.url == "http://localhost/alarm.cgi"

⚠️ 注意:这种方式需要注意清理全局适配器,避免影响其他测试。如果用pytest,可以在测试前后加清理逻辑,把DEFAULT_ADAPTERS恢复成原来的状态。

另外,测试时一定要加time.sleep等待线程执行请求,否则线程可能还没跑到请求代码,测试就已经结束了,会导致断言失败。

内容的提问来源于stack exchange,提问作者JohnLoopM

相关产品推荐
方舟 Agent Plan

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

最近更新时间:2026.05.20 10:16:28