mirror of
https://github.com/PyCQA/flake8.git
synced 2026-04-02 03:46:53 +00:00
e8f43e1243
~40% improvement over status quo (perf measurements are best-of-5) ### before ```console $ time flake8 /dev/null real 0m0.337s user 0m0.212s sys 0m0.028s ``` ### after ```console $ time flake8 /dev/null real 0m0.197s user 0m0.182s sys 0m0.012s ```
78 lines
2.7 KiB
Python
78 lines
2.7 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 _parallel_checker_manager():
|
|
"""Call Manager.run() and return the number of calls to `run_serial`."""
|
|
style_guide = style_guide_mock()
|
|
manager = checker.Manager(style_guide, [], [])
|
|
# multiple checkers is needed for parallel mode
|
|
manager.checkers = [mock.Mock(), mock.Mock()]
|
|
return manager
|
|
|
|
|
|
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')
|
|
with mock.patch('_multiprocessing.SemLock', side_effect=err):
|
|
manager = _parallel_checker_manager()
|
|
with mock.patch.object(manager, 'run_serial') as serial:
|
|
manager.run()
|
|
assert serial.call_count == 1
|
|
|
|
|
|
@mock.patch('flake8.utils.is_windows', return_value=False)
|
|
def test_oserrors_are_reraised(is_windows):
|
|
"""Verify that unexpected OSErrors will cause the Manager to reraise."""
|
|
err = OSError(errno.EAGAIN, 'Ominous message')
|
|
with mock.patch('_multiprocessing.SemLock', side_effect=err):
|
|
manager = _parallel_checker_manager()
|
|
with mock.patch.object(manager, 'run_serial') as serial:
|
|
with pytest.raises(OSError):
|
|
manager.run()
|
|
assert serial.call_count == 0
|
|
|
|
|
|
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
|