pre-commit/flake8
1
0
Fork 0
mirror of https://github.com/PyCQA/flake8.git synced 2026-04-07 05:26:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

enable multiprocessing on other platforms

Browse source
This commit is contained in:
Anthony Sottile 2022-10-26 20:17:35 -07:00
parent ebbb57d63c
commit 0d667a7329
7 changed files with 175 additions and 317 deletions
Download patch file Download diff file Expand all files Collapse all files

17
tests/integration/test_checker.py
View file

@ -266,17 +266,12 @@ def test_report_order(results, expected_order):
# tuples to create the expected result lists from the indexes
expected_results = [results[index] for index in expected_order]
file_checker = mock.Mock(spec=["results", "display_name"])
file_checker.results = results
file_checker.display_name = "placeholder"
style_guide = mock.MagicMock(spec=["options", "processing_file"])
# Create a placeholder manager without arguments or plugins
# Just add one custom file checker which just provides the results
manager = checker.Manager(style_guide, finder.Checkers([], [], []))
manager.checkers = manager._all_checkers = [file_checker]
manager = checker.Manager(style_guide, finder.Checkers([], [], []), [])
manager.results = [("placeholder", results, {})]
# _handle_results is the first place which gets the sorted result
# Should something non-private be mocked instead?
handler = mock.Mock(side_effect=count_side_effect)
@ -295,9 +290,9 @@ def test_acquire_when_multiprocessing_pool_can_initialize():
This simulates the behaviour on most common platforms.
"""
with mock.patch("multiprocessing.Pool") as pool:
result = checker._try_initialize_processpool(2)
result = checker._try_initialize_processpool(2, [])
pool.assert_called_once_with(2, checker._pool_init)
pool.assert_called_once_with(2, checker._mp_init, initargs=([],))
assert result is pool.return_value
@ -314,9 +309,9 @@ def test_acquire_when_multiprocessing_pool_can_not_initialize():
https://github.com/python/cpython/blob/4e02981de0952f54bf87967f8e10d169d6946b40/Lib/multiprocessing/synchronize.py#L30-L33
"""
with mock.patch("multiprocessing.Pool", side_effect=ImportError) as pool:
result = checker._try_initialize_processpool(2)
result = checker._try_initialize_processpool(2, [])
pool.assert_called_once_with(2, checker._pool_init)
pool.assert_called_once_with(2, checker._mp_init, initargs=([],))
assert result is None