如何用Pytest测试依赖sys.stdin/stdout的多线程递归Python代码？

你遇到的核心问题是线程隔离导致Mock无法生效，加上特殊栈/递归设置引发的底层内存问题。下面给你几个针对性的解决思路，按优先级排序：

1. 优先剥离线程依赖，直接测试核心逻辑 #

如果你的main函数只是为了规避主线程栈限制才放到新线程里，那测试时完全可以绕过线程启动——毕竟线程在这里不是业务逻辑的一部分，只是解决递归深度的手段。

比如原代码结构可能是这样：

import sys
import threading

def main():
    spam = SpamClass()
    data = spam.read()
    result = spam.process(data)
    spam.write(result)

if __name__ == "__main__":
    sys.setrecursionlimit(10**6)
    threading.stack_size(2**27)
    threading.Thread(target=main).start()

测试时直接调用核心逻辑，同时提前设置递归和栈参数，用capsys或io.StringIOMock输入输出：

import sys
import io
import threading
import pytest
from your_module import main

def test_core_logic(capsys):
    # 保存原始设置，测试后恢复
    original_recursion_limit = sys.getrecursionlimit()
    original_stack_size = threading.stack_size()
    original_stdin = sys.stdin

    sys.setrecursionlimit(10**6)
    threading.stack_size(2**27)
    sys.stdin = io.StringIO("test input content")

    try:
        # 直接调用main，无需启动线程
        main()
        # 捕获输出并和样本对比
        captured = capsys.readouterr()
        with open("sample_output.txt", "r") as f:
            expected = f.read()
        assert captured.out.strip() == expected.strip()
    finally:
        # 恢复所有全局状态，避免污染其他测试
        sys.setrecursionlimit(original_recursion_limit)
        threading.stack_size(original_stack_size)
        sys.stdin = original_stdin

这个方案最简洁，能彻底避开线程带来的Mock隔离问题。

2. 必须保留线程时，在启动前完成Mock #

新线程会继承主线程启动时的sys对象状态，所以要在启动线程前完成stdin/stdout的替换，同时手动捕获线程输出（capsys默认只捕获主线程输出）：

import sys
import io
import threading
from your_module import main

def test_with_thread():
    original_recursion_limit = sys.getrecursionlimit()
    original_stack_size = threading.stack_size()
    original_stdin = sys.stdin
    original_stdout = sys.stdout

    sys.setrecursionlimit(10**6)
    threading.stack_size(2**27)

    # 准备Mock输入输出对象
    mock_stdin = io.StringIO("test input content")
    mock_stdout = io.StringIO()
    sys.stdin = mock_stdin
    sys.stdout = mock_stdout

    try:
        # 启动线程并等待执行完成
        thread = threading.Thread(target=main)
        thread.start()
        thread.join()

        # 获取输出并对比样本
        output = mock_stdout.getvalue()
        with open("sample_output.txt", "r") as f:
            expected = f.read()
        assert output.strip() == expected.strip()
    finally:
        # 恢复全局状态
        sys.setrecursionlimit(original_recursion_limit)
        threading.stack_size(original_stack_size)
        sys.stdin = original_stdin
        sys.stdout = original_stdout

3. 解决SIGBUS（地址未对齐）问题 #

你用monkeypatch替换SpamClass.read时出现SIGBUS，大概率是两个原因：

• SpamClass.read可能调用了底层C扩展代码，依赖特定的内存对齐规则，替换成Python方法后破坏了这个规则；
• 你设置的threading.stack_size(2**27)（134MB）可能超出系统允许的栈大小上限，导致栈地址不对齐。

解决思路：

• 不要替换read方法，直接Mocksys.stdin的底层buffer（如果是二进制输入用io.BytesIO，确保编码和原代码一致）；
• 调整栈大小为系统支持的对齐值，比如2**26（64MB），大部分系统对栈大小的要求是2的幂且不超过128MB；
• 如果SpamClass是自定义类，检查read方法中是否有直接操作内存的代码（比如ctypes调用），确保内存访问是对齐的。

示例调整栈大小的测试代码：

def test_adjust_stack_size():
    original_stack_size = threading.stack_size()
    # 使用更小的、符合系统要求的栈大小
    threading.stack_size(2**26)
    try:
        # 后续测试逻辑...
    finally:
        threading.stack_size(original_stack_size)

额外注意事项 #

• 所有修改全局状态（递归限制、栈大小、sys对象）的操作，必须在finally块中恢复原始值，否则会影响其他测试用例；
• 若处理二进制输入输出，一定要用io.BytesIO而非io.StringIO，避免编码不匹配；
• 测试时临时提高递归限制后，务必记得恢复，防止其他测试出现意外的递归深度错误。

内容的提问来源于stack exchange，提问作者curlew77