操作を実行する関数の周りにいくつかの単体テストを書き込もうとしていますmap_async()。より具体的には、プロセスの 1 つで例外が発生した場合に、いくつかのファイルがクリーンアップされることを確認したいと思います。以下に示す意図を持つサンプルの擬似コード。
foo.py
def write_chunk(chunk):
... create file from chunk
return created_filename
class Foo:
def write_parallel(chunks):
filenames = set()
try:
pool = Pool(processes=2)
pool.map_async(write_chunk, chunks, callback=filenames.add)
except Exception:
//handle exception
finally:
cleanup_files(filenames)
test_foo.py
@patch("foo.write_chunk")
def test_write_parallel_exception_cleanup(self, mock_write_chunk):
def mock_side_effect(chunk):
if "chunk_1" == chunk:
raise Exception
else:
return chunk
mock_write_chunk.side_effect = mock_side_effect
foo = Foo()
foo.write_parallel({"chunk_1", "chunk_2"})
//assert "chunk_2" cleaned up and exception is thrown.
ただし、テストを実行すると、次の PicklingError: が表示されPicklingError: Can't pickle <class 'mock.MagicMock'>: it's not the same object as mock.MagicMockます。
マップされた関数を自分のモック関数に置き換えるという望ましい結果を実行する方法はありますか?