ETL作业功能测试中Monkeypatch抽取模块生效问题

问题背景 #

现有ETL仓库结构：

etl_repo  
├── app  
     ├── extract  
         ├── extr_a.py  
         ├── extr_b.py   
     ├── transform  
         ├── trans_a.py  
         ├── trans_b.py  
     ├── load  
         ├── load_a.py  
         ├── load_b.py  
     ├── config.py  
     ├── my_job1.py 
├── tests  
     ├── test_my_job1.py  

生产环境通过python app/my_job1.py执行作业，my_job1.py将核心逻辑封装在main()函数中，通过if __name__ == "__main__"触发调用。当前要做端到端功能测试，用pytest的Monkeypatch替换extract模块的网络抽取函数为静态数据，但遇到核心问题：

• 用subprocess执行生产命令时，子进程无法继承Monkeypatch补丁
• 直接在测试中导入my_job1并调用main()，补丁对my_job1内部调用的extract函数不生效，且要求不修改生产代码添加测试分支逻辑。

测试代码示例：

from app.extract import extr_a
from app import my_job1


@pytest.fixture
def mock_getdata_extra_a(monkeypatch):
    def mock_func(aref):
        with open("staticpickle.pkl", "rb") as file:
            data = pickle.load(file)
        return data[aref]

    monkeypatch.setattr(extr_a, "getdata", mock_func)


def test_my_job1(mock_getdata_extra_a):
    # 此处调用已被补丁生效
    extr_a.getdata("thisisaref")

    # 但my_job1.main()中的extr_a.getdata未被补丁生效
    my_job1.main()

解决方案 #

方案1：直接补丁my_job1模块内的函数引用 #

问题根源是Python的导入机制：my_job1.py中导入的getdata是独立的对象引用，和测试代码中补丁的extr_a模块内的getdata并非同一引用。直接补丁my_job1模块内的getdata即可生效。

修改测试代码：

from app import my_job1
from app.extract import extr_a


@pytest.fixture
def mock_getdata_extra_a(monkeypatch):
    def mock_func(aref):
        with open("staticpickle.pkl", "rb") as file:
            data = pickle.load(file)
        return data[aref]

    # 补丁my_job1模块中导入的getdata引用
    monkeypatch.setattr(my_job1, "getdata", mock_func)
    # 可选：同时补丁extr_a模块，确保其他场景调用也生效
    monkeypatch.setattr(extr_a, "getdata", mock_func)


def test_my_job1(mock_getdata_extra_a):
    my_job1.main()
    # 添加断言验证ETL结果，比如检查加载模块的输出、生成的文件等

方案2：补丁后重新加载my_job1模块 #

如果my_job1在补丁前已被导入并缓存了原函数引用，可在补丁完成后重新加载模块，让它重新导入被补丁后的函数。

修改测试代码：

import importlib
from app.extract import extr_a
from app import my_job1


@pytest.fixture
def mock_getdata_extra_a(monkeypatch):
    def mock_func(aref):
        with open("staticpickle.pkl", "rb") as file:
            data = pickle.load(file)
        return data[aref]

    monkeypatch.setattr(extr_a, "getdata", mock_func)
    # 重新加载my_job1，使其获取补丁后的函数引用
    importlib.reload(my_job1)


def test_my_job1(mock_getdata_extra_a):
    my_job1.main()
    # 添加断言逻辑

方案3：模拟生产执行流程的独立补丁脚本 #

如果需要严格贴近生产环境的执行方式，可编写一个前置补丁的脚本，用subprocess执行该脚本替代直接运行my_job1.py：

1. 在tests目录下创建test_runner.py：

import sys
import pickle
from app.extract import extr_a

# 替换抽取函数为静态数据
def mock_getdata(aref):
    with open("tests/staticpickle.pkl", "rb") as file:
        data = pickle.load(file)
    return data[aref]

extr_a.getdata = mock_getdata

# 导入并执行作业逻辑
sys.path.insert(0, ".")
from app import my_job1
my_job1.main()

2. 在测试中执行该脚本：

import subprocess

def test_my_job1():
    result = subprocess.run(
        ["python", "tests/test_runner.py"],
        capture_output=True,
        text=True
    )
    # 验证执行结果
    assert result.returncode == 0
    # 添加其他断言，比如检查数据库加载结果、生成的输出文件等

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