mirror of
https://github.com/PyCQA/flake8.git
synced 2026-03-30 18:56:53 +00:00
e14d0e6352
It seems likely that the multiprocessing module on Windows is not capable of serializing an object with the structure that we have and preserving the attributes we dynamically set on plugins (like the FlakesChecker). To avoid issues like this with all plugins (although we have only found this on Windows with the FlakesChecker), let's try serializing the Checkers PluginTypeManager to a dictionary so that the only object that a Queue is really trying to serialize/deserialize is the FlakesChecker itself. Related to #179
66 lines
2.3 KiB
Python
66 lines
2.3 KiB
Python
"""Tests for the Manager object for FileCheckers."""
|
|
import errno
|
|
|
|
import mock
|
|
import pytest
|
|
|
|
from flake8 import checker
|
|
|
|
|
|
def style_guide_mock(**kwargs):
|
|
"""Create a mock StyleGuide object."""
|
|
kwargs.setdefault('diff', False)
|
|
kwargs.setdefault('jobs', '4')
|
|
style_guide = mock.Mock()
|
|
style_guide.options = mock.Mock(**kwargs)
|
|
return style_guide
|
|
|
|
|
|
def test_oserrors_cause_serial_fall_back():
|
|
"""Verify that OSErrors will cause the Manager to fallback to serial."""
|
|
err = OSError(errno.ENOSPC, 'Ominous message about spaceeeeee')
|
|
style_guide = style_guide_mock()
|
|
with mock.patch('multiprocessing.Queue', side_effect=err):
|
|
manager = checker.Manager(style_guide, [], [])
|
|
assert manager.using_multiprocessing is False
|
|
|
|
|
|
@mock.patch('flake8.utils.is_windows', return_value=False)
|
|
def test_oserrors_are_reraised(is_windows):
|
|
"""Verify that OSErrors will cause the Manager to fallback to serial."""
|
|
err = OSError(errno.EAGAIN, 'Ominous message')
|
|
style_guide = style_guide_mock()
|
|
with mock.patch('multiprocessing.Queue', side_effect=err):
|
|
with pytest.raises(OSError):
|
|
checker.Manager(style_guide, [], [])
|
|
|
|
|
|
def test_multiprocessing_is_disabled():
|
|
"""Verify not being able to import multiprocessing forces jobs to 0."""
|
|
style_guide = style_guide_mock()
|
|
with mock.patch('flake8.checker.multiprocessing', None):
|
|
manager = checker.Manager(style_guide, [], [])
|
|
assert manager.jobs == 0
|
|
|
|
|
|
def test_make_checkers():
|
|
"""Verify that we create a list of FileChecker instances."""
|
|
style_guide = style_guide_mock()
|
|
files = ['file1', 'file2']
|
|
checkplugins = mock.Mock()
|
|
checkplugins.to_dictionary.return_value = {
|
|
'ast_plugins': [],
|
|
'logical_line_plugins': [],
|
|
'physical_line_plugins': [],
|
|
}
|
|
with mock.patch('flake8.checker.multiprocessing', None):
|
|
manager = checker.Manager(style_guide, files, checkplugins)
|
|
|
|
with mock.patch('flake8.utils.filenames_from') as filenames_from:
|
|
filenames_from.side_effect = [['file1'], ['file2']]
|
|
with mock.patch('flake8.utils.fnmatch', return_value=True):
|
|
with mock.patch('flake8.processor.FileProcessor'):
|
|
manager.make_checkers()
|
|
|
|
for file_checker in manager.checkers:
|
|
assert file_checker.filename in files
|