mock.patch 和多处理 [英] mock.patch and multiprocessing

问题描述

我正在努力在多处理环境中使用 mock.patch 而没有多处理 mock.patch 工作正常.文件名:test_mp.py

I'm struggling to use mock.patch in a multiprocessing environment while without multiprocessing mock.patch works fine. Filename: test_mp.py

import multiprocessing
import mock

def inner():
    return sub()

def sub():
    return "abc"

def test_local():
    assert inner()=="abc"

def test_mp():    
    with multiprocessing.Pool() as pool:
        assert pool.apply(inner,args=[])=='abc'

def test_mock():    
    with mock.patch('test_mp.sub', return_value='xxx') as xx:
        assert inner()=="xxx"
        xx.assert_called_once()

def test_mp_mock():    
    with multiprocessing.Pool() as pool:
        with mock.patch('test_mp.sub', return_value='xyz') as xx:
            assert pool.apply(inner,args=[])=='xyz'
            xx.assert_called_once()

• 测试 test_local、test_mock 和 test_mock 成功完成
• 但是 test_mp_mock 失败了
• Test test_local, test_mock and test_mock finishes successfully
• However test_mp_mock fails with

=================================== FAILURES ===================================
_________________________________ test_mp_mock _________________________________

    def test_mp_mock():
        with multiprocessing.Pool() as pool:
            with mock.patch('test_mp.sub', return_value='xyz') as xx:
>               assert pool.apply(inner,args=[])=='xyz'
E               AssertionError: assert 'abc' == 'xyz'
E                 - abc
E                 + xyz

projects/mfhealth/test_mp.py:25: AssertionError

更新:

基于 https://medium.com/uckey/how-mock-patch-decorator-works-in-python-37acd8b78ae 并借助 sharedmock https://github.com/elritsch/python-sharedmock 我能够走得更远，但仍然没有完成.

Based on https://medium.com/uckey/how-mock-patch-decorator-works-in-python-37acd8b78ae and with help of sharedmock https://github.com/elritsch/python-sharedmock I was able to get further, but still not complete.

我为

from sharedmock.mock import SharedMock

def inner2(sm):
    with mock.patch('test_mp.sub', sm) as xx:
        return inner()


def test_mp_smock():
    with multiprocessing.Pool() as pool:
        sm=SharedMock()
        sm.return_value="xyz"
        with mock.patch('test_mp.sub', sm) as xx:
            assert pool.apply(inner2,args=[sm])=='xyz'
            assert xx.call_count == 1

def test_mp_mock2():
    with multiprocessing.Pool() as pool:
        sm=mock.Mock()
        sm.return_value="xyz"
        print(f"before patch {sub}, {locals()}")
        with mock.patch('test_mp.sub', sm) as xx:
            print(f"after patch {sub}")
            assert pool.apply(inner2,args=[sm])=='xyz'
            assert xx.call_count == 1

结果如下:

• test_mp_smock 成功完成.
• test_mp_mock2 失败，出现 _pickle.PicklingError: Can't pickle <class 'mock.mock.Mock'>: it's not the same object as mock.mock.Mock
• test_mp_smock finishes successfully.
• test_mp_mock2 fails with _pickle.PicklingError: Can't pickle <class 'mock.mock.Mock'>: it's not the same object as mock.mock.Mock

test_mp_smock 的主要缺点是必须引入新方法 inner2 以通过 mock.patch 激活补丁.关于如何将补丁从 test_mp_smock 传播到被测代码而不引入包装方法 inner2 的任何想法，因为我无法覆盖.

The main disadvantage of test_mp_smock is that a new method inner2 must be introduced to activate patching via mock.patch. Any idea of how to propagate patching from test_mp_smock to code under test without introducing a wrapper method inner2 because I cannot override.

pool.apply(inner,args=[])

推荐答案

我终于搞定了.

SharedMock 不像 Mock 那样灵活，例如.缺少assert_call_once_with，...，但您可以设置返回值或检查调用次数或其参数.

SharedMock is not that flexible like Mock, eg. missing assert_called_once_with, ..., but you can set returned value or check the number of calls or its arguments.

import multiprocessing
import mock
from sharedmock.mock import SharedMock


def inner():
    return sub(x="xxx")

def sub(x=""):
    return f"abc"

def fun_under_test():
    with multiprocessing.Pool() as pool:
        assert pool.apply(inner,args=[])=='xyz'

def test_final():
    sm=SharedMock()
    sm.return_value="xyz"
    with mock.patch('test_mp.sub', sm) as xx:
        fun_under_test()
        assert xx.call_count == 1 #number of calls of sub function
        assert xx.mock_calls[0][2]['x']=="xxx" # value of parameters ie sub(x="xxx")

这篇关于mock.patch 和多处理的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！