pre-commit/flake8
1
0
Fork 0
mirror of https://github.com/PyCQA/flake8.git synced 2026-03-29 10:36:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix mypy errors

Browse source
This commit is contained in:
Anthony Sottile 2021-04-07 08:28:05 -07:00
parent fcf56ac93e
commit 3f10c04fd0
8 changed files with 92 additions and 76 deletions
Download patch file Download diff file Expand all files Collapse all files

16
src/flake8/checker.py
View file

@ -40,11 +40,7 @@ SERIAL_RETRY_ERRNOS = {
def _multiprocessing_is_fork(): # type () -> bool
"""Class state is only preserved when using the `fork` strategy."""
return (
multiprocessing
# https://github.com/python/typeshed/pull/3415
and multiprocessing.get_start_method() == "fork" # type: ignore
)
return multiprocessing and multiprocessing.get_start_method() == "fork"
class Manager:
@ -396,7 +392,7 @@ class FileChecker:
# If we're recovering from a problem in _make_processor, we will not
# have this attribute.
if hasattr(self, "processor"):
if hasattr(self, "processor") and self.processor is not None:
line = self.processor.noqa_line_for(line_number)
else:
line = None
@ -407,6 +403,7 @@ class FileChecker:
def run_check(self, plugin, **arguments):
"""Run the check in a single plugin."""
LOG.debug("Running %r with %r", plugin, arguments)
assert self.processor is not None
try:
self.processor.keyword_arguments_for(
plugin["parameters"], arguments
@ -467,6 +464,7 @@ class FileChecker:
def run_ast_checks(self) -> None:
"""Run all checks expecting an abstract syntax tree."""
assert self.processor is not None
try:
ast = self.processor.build_ast()
except (ValueError, SyntaxError, TypeError) as e:
@ -494,6 +492,7 @@ class FileChecker:
def run_logical_checks(self):
"""Run all checks expecting a logical line."""
assert self.processor is not None
comments, logical_line, mapping = self.processor.build_logical_line()
if not mapping:
return
@ -522,6 +521,7 @@ class FileChecker:
A single physical check may return multiple errors.
"""
assert self.processor is not None
for plugin in self.checks["physical_line_plugins"]:
self.processor.update_checker_state_for(plugin)
result = self.run_check(plugin, physical_line=physical_line)
@ -554,6 +554,7 @@ class FileChecker:
Instead of using this directly, you should use
:meth:`flake8.checker.FileChecker.run_checks`.
"""
assert self.processor is not None
parens = 0
statistics = self.statistics
file_processor = self.processor
@ -577,6 +578,7 @@ class FileChecker:
def run_checks(self):
"""Run checks against the file."""
assert self.processor is not None
try:
self.process_tokens()
self.run_ast_checks()
@ -594,6 +596,7 @@ class FileChecker:
def handle_newline(self, token_type):
"""Handle the logic when encountering a newline token."""
assert self.processor is not None
if token_type == tokenize.NEWLINE:
self.run_logical_checks()
self.processor.reset_blank_before()
@ -608,6 +611,7 @@ class FileChecker:
self, token: processor._Token, prev_physical: str
) -> None:
"""Run physical checks if and only if it is at the end of the line."""
assert self.processor is not None
# a newline token ends a single physical line.
if processor.is_eol_token(token):
# if the file does not end with a newline, the NEWLINE

37
src/flake8/main/application.py
View file

@ -44,7 +44,7 @@ class Application:
#: The timestamp when the Application instance was instantiated.
self.start_time = time.time()
#: The timestamp when the Application finished reported errors.
self.end_time: float = None
self.end_time: Optional[float] = None
#: The name of the program being run
self.program = program
#: The version of the program being run
@ -63,24 +63,26 @@ class Application:
options.register_default_options(self.option_manager)
#: The instance of :class:`flake8.plugins.manager.Checkers`
self.check_plugins: plugin_manager.Checkers = None
self.check_plugins: Optional[plugin_manager.Checkers] = None
#: The instance of :class:`flake8.plugins.manager.ReportFormatters`
self.formatting_plugins: plugin_manager.ReportFormatters = None
self.formatting_plugins: Optional[
plugin_manager.ReportFormatters
] = None
#: The user-selected formatter from :attr:`formatting_plugins`
self.formatter: BaseFormatter = None
self.formatter: Optional[BaseFormatter] = None
#: The :class:`flake8.style_guide.StyleGuideManager` built from the
#: user's options
self.guide: style_guide.StyleGuideManager = None
self.guide: Optional[style_guide.StyleGuideManager] = None
#: The :class:`flake8.checker.Manager` that will handle running all of
#: the checks selected by the user.
self.file_checker_manager: checker.Manager = None
self.file_checker_manager: Optional[checker.Manager] = None
#: The user-supplied options parsed into an instance of
#: :class:`argparse.Namespace`
self.options: argparse.Namespace = None
self.options: Optional[argparse.Namespace] = None
#: The left over arguments that were not parsed by
#: :attr:`option_manager`
self.args: List[str] = None
self.args: Optional[List[str]] = None
#: The number of errors, warnings, and other messages after running
#: flake8 and taking into account ignored errors and lines.
self.result_count = 0
@ -128,6 +130,7 @@ class Application:
This should be the last thing called on the application instance. It
will check certain options and exit appropriately.
"""
assert self.options is not None
if self.options.count:
print(self.result_count)
@ -162,8 +165,10 @@ class Application:
def register_plugin_options(self) -> None:
"""Register options provided by plugins to our option manager."""
assert self.check_plugins is not None
self.check_plugins.register_options(self.option_manager)
self.check_plugins.register_plugin_versions(self.option_manager)
assert self.formatting_plugins is not None
self.formatting_plugins.register_options(self.option_manager)
def parse_configuration_and_cli(
@ -190,15 +195,18 @@ class Application:
if not self.parsed_diff:
self.exit()
assert self.check_plugins is not None
self.check_plugins.provide_options(
self.option_manager, self.options, self.args
)
assert self.formatting_plugins is not None
self.formatting_plugins.provide_options(
self.option_manager, self.options, self.args
)
def formatter_for(self, formatter_plugin_name):
"""Retrieve the formatter class by plugin name."""
assert self.formatting_plugins is not None
default_formatter = self.formatting_plugins["default"]
formatter_plugin = self.formatting_plugins.get(formatter_plugin_name)
if formatter_plugin is None:
@ -214,6 +222,7 @@ class Application:
self, formatter_class: Optional[Type["BaseFormatter"]] = None
) -> None:
"""Initialize a formatter based on the parsed options."""
assert self.options is not None
format_plugin = self.options.format
if 1 <= self.options.quiet < 2:
format_plugin = "quiet-filename"
@ -227,6 +236,8 @@ class Application:
def make_guide(self) -> None:
"""Initialize our StyleGuide."""
assert self.formatter is not None
assert self.options is not None
self.guide = style_guide.StyleGuideManager(
self.options, self.formatter
)
@ -252,6 +263,7 @@ class Application:
:param list files:
List of filenames to process
"""
assert self.file_checker_manager is not None
if self.running_against_diff:
files = sorted(self.parsed_diff)
self.file_checker_manager.start(files)
@ -267,9 +279,12 @@ class Application:
def report_benchmarks(self):
"""Aggregate, calculate, and report benchmarks for this run."""
assert self.options is not None
if not self.options.benchmark:
return
assert self.file_checker_manager is not None
assert self.end_time is not None
time_elapsed = self.end_time - self.start_time
statistics = [("seconds elapsed", time_elapsed)]
add_statistic = statistics.append
@ -280,6 +295,7 @@ class Application:
per_second_description = f"{statistic} processed per second"
add_statistic((per_second_description, int(value / time_elapsed)))
assert self.formatter is not None
self.formatter.show_benchmarks(statistics)
def report_errors(self) -> None:
@ -289,6 +305,7 @@ class Application:
number of errors, warnings, and other messages found.
"""
LOG.info("Reporting errors")
assert self.file_checker_manager is not None
results = self.file_checker_manager.report()
self.total_result_count, self.result_count = results
LOG.info(
@ -299,9 +316,12 @@ class Application:
def report_statistics(self):
"""Aggregate and report statistics from this run."""
assert self.options is not None
if not self.options.statistics:
return
assert self.formatter is not None
assert self.guide is not None
self.formatter.show_statistics(self.guide.stats)
def initialize(self, argv: List[str]) -> None:
@ -332,6 +352,7 @@ class Application:
def report(self):
"""Report errors, statistics, and benchmarks."""
assert self.formatter is not None
self.formatter.start()
self.report_errors()
self.report_statistics()

4
src/flake8/plugins/manager.py
View file

@ -251,7 +251,9 @@ class PluginManager: # pylint: disable=too-few-public-methods
for plugin_str in local_plugins:
name, _, entry_str = plugin_str.partition("=")
name, entry_str = name.strip(), entry_str.strip()
entry_point = importlib_metadata.EntryPoint(name, entry_str, None)
entry_point = importlib_metadata.EntryPoint(
name, entry_str, self.namespace
)
self._load_plugin_from_entrypoint(entry_point, local=True)
def _load_entrypoint_plugins(self):

27
src/flake8/style_guide.py
View file

@ -54,20 +54,19 @@ def find_noqa(physical_line: str) -> Optional[Match[str]]:
return defaults.NOQA_INLINE_REGEXP.search(physical_line)
_Violation = collections.namedtuple(
"Violation",
[
"code",
"filename",
"line_number",
"column_number",
"text",
"physical_line",
],
)
class Violation(_Violation):
class Violation(
collections.namedtuple(
"Violation",
[
"code",
"filename",
"line_number",
"column_number",
"text",
"physical_line",
],
)
):
"""Class representing a violation reported by Flake8."""
def is_inline_ignored(self, disable_noqa: bool) -> bool:

2
src/flake8/utils.py
View file

@ -57,7 +57,7 @@ def parse_comma_separated_list(
return [item for item in item_gen if item]
_Token = collections.namedtuple("Token", ("tp", "src"))
_Token = collections.namedtuple("_Token", ("tp", "src"))
_CODE, _FILE, _COLON, _COMMA, _WS = "code", "file", "colon", "comma", "ws"
_EOF = "eof"
_FILE_LIST_TOKEN_TYPES = [

4
tests/integration/test_plugins.py
View file

@ -41,7 +41,9 @@ def test_enable_local_plugin_from_config():
app = application.Application()
app.initialize(['flake8', '--config', LOCAL_PLUGIN_CONFIG])
assert app.check_plugins is not None
assert app.check_plugins['XE'].plugin is ExtensionTestPlugin
assert app.formatting_plugins is not None
assert app.formatting_plugins['XR'].plugin is ReportTestPlugin
@ -51,6 +53,7 @@ def test_local_plugin_can_add_option():
app.initialize(
['flake8', '--config', LOCAL_PLUGIN_CONFIG, '--anopt', 'foo'])
assert app.options is not None
assert app.options.anopt == 'foo'
@ -59,4 +62,5 @@ def test_enable_local_plugin_at_non_installed_path():
app = application.Application()
app.initialize(['flake8', '--config', LOCAL_PLUGIN_PATH_CONFIG])
assert app.check_plugins is not None
assert app.check_plugins['XE'].plugin.name == 'ExtensionTestPlugin2'

70
tests/unit/test_exceptions.py
View file

@ -1,48 +1,34 @@
"""Tests for the flake8.exceptions module."""
import pickle
import pytest
from flake8 import exceptions
class _ExceptionTest:
def test_pickleable(self):
"""Test that the exception is round-trip pickleable."""
for proto in range(pickle.HIGHEST_PROTOCOL + 1):
new_err = pickle.loads(pickle.dumps(self.err, protocol=proto))
assert str(self.err) == str(new_err)
orig_e = self.err.original_exception
new_e = new_err.original_exception
assert (type(orig_e), orig_e.args) == (type(new_e), new_e.args)
class TestFailedToLoadPlugin(_ExceptionTest):
"""Tests for the FailedToLoadPlugin exception."""
err = exceptions.FailedToLoadPlugin(
plugin_name='plugin_name',
exception=ValueError('boom!'),
)
class TestInvalidSyntax(_ExceptionTest):
"""Tests for the InvalidSyntax exception."""
err = exceptions.InvalidSyntax(exception=ValueError('Unexpected token: $'))
class TestPluginRequestedUnknownParameters(_ExceptionTest):
"""Tests for the PluginRequestedUnknownParameters exception."""
err = exceptions.PluginRequestedUnknownParameters(
plugin={'plugin_name': 'plugin_name'},
exception=ValueError('boom!'),
)
class TestPluginExecutionFailed(_ExceptionTest):
"""Tests for the PluginExecutionFailed exception."""
err = exceptions.PluginExecutionFailed(
plugin={'plugin_name': 'plugin_name'},
exception=ValueError('boom!'),
)
@pytest.mark.parametrize(
'err',
(
exceptions.FailedToLoadPlugin(
plugin_name='plugin_name',
exception=ValueError('boom!'),
),
exceptions.InvalidSyntax(exception=ValueError('Unexpected token: $')),
exceptions.PluginRequestedUnknownParameters(
plugin={'plugin_name': 'plugin_name'},
exception=ValueError('boom!'),
),
exceptions.PluginExecutionFailed(
plugin={'plugin_name': 'plugin_name'},
exception=ValueError('boom!'),
)
),
)
def test_pickleable(err):
"""Ensure that our exceptions can cross pickle boundaries."""
for proto in range(pickle.HIGHEST_PROTOCOL + 1):
new_err = pickle.loads(pickle.dumps(err, protocol=proto))
assert str(err) == str(new_err)
orig_e = err.original_exception
new_e = new_err.original_exception
assert (type(orig_e), orig_e.args) == (type(new_e), new_e.args)

8
tests/unit/test_plugin_manager.py
View file

@ -18,8 +18,8 @@ def test_calls_entrypoints_creates_plugins_automaticaly(entry_points_mck):
"""Verify that we create Plugins on instantiation."""
entry_points_mck.return_value = {
'testing.entrypoints': [
importlib_metadata.EntryPoint('T100', '', None),
importlib_metadata.EntryPoint('T200', '', None),
importlib_metadata.EntryPoint('T100', '', 'testing.entrypoints'),
importlib_metadata.EntryPoint('T200', '', 'testing.entrypoints'),
],
}
plugin_mgr = manager.PluginManager(namespace='testing.entrypoints')
@ -36,8 +36,8 @@ def test_handles_mapping_functions_across_plugins(entry_points_mck):
"""Verify we can use the PluginManager call functions on all plugins."""
entry_points_mck.return_value = {
'testing.entrypoints': [
importlib_metadata.EntryPoint('T100', '', None),
importlib_metadata.EntryPoint('T200', '', None),
importlib_metadata.EntryPoint('T100', '', 'testing.entrypoints'),
importlib_metadata.EntryPoint('T200', '', 'testing.entrypoints'),
],
}
plugin_mgr = manager.PluginManager(namespace='testing.entrypoints')