本文介绍了mock.patch 和多处理的处理方法,对大家解决问题具有一定的参考价值,需要的朋友们下面随着小编来一起学习吧!

问题描述


我正在努力在多处理环境中使用 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_localtest_mocktest_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 和多处理的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持!

09-05 10:34