From 0a51002b002c42363a6d0c33ee0cff3b4c9e0e58 Mon Sep 17 00:00:00 2001 From: Lev Vereshchagin Date: Mon, 12 Oct 2026 12:41:29 +0300 Subject: [PATCH 1/2] Re-enable COP001 with static module resolution --- AGENTS.md | 6 + README.md | 18 ++ .../checks/disabled/__init__.py | 0 .../disabled/module_import_many_names.py | 90 ------- .../checks/module_import_many_names.py | 182 +++++++++++++++ .../checks/name_length.py | 22 +- .../plugin.py | 16 +- tests/test_module_imports.py | 221 ++++++++++++++++++ tests/test_plugin.py | 5 +- 9 files changed, 452 insertions(+), 108 deletions(-) create mode 100644 AGENTS.md delete mode 100644 src/community_of_python_flake8_plugin/checks/disabled/__init__.py delete mode 100644 src/community_of_python_flake8_plugin/checks/disabled/module_import_many_names.py create mode 100644 src/community_of_python_flake8_plugin/checks/module_import_many_names.py create mode 100644 tests/test_module_imports.py diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..2ec97a6 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,6 @@ +# Import analysis + +Import checks must inspect AST and filesystem data without executing inspected +packages, import hooks, extension modules, or `.pth` files. Do not use runtime +imports or `find_spec` to resolve application dependencies. Leave unresolved or +dynamic imports unchecked rather than guessing their type. diff --git a/README.md b/README.md index 5e0c8cd..94d8dbb 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,24 @@ This plugin implements the following code style checks: - **COP016**: Add `*` or `/` when defining more than two regular arguments - **COP017**: Avoid reassigning variables not annotated with `Mutable` +### COP001: import modules instead of many symbols + +Use a module import when importing more than two statically identifiable symbols +from the same module. Aliases still count; imported submodules do not. Standard +library imports are handled exclusively by COP002, preserving its exceptions +for `__future__` and `collections.abc`. + +COP001 reads Python source and stub files without executing inspected packages, +import hooks, native extensions, or `.pth` files. Search roots include the current +directory, the linted file's project directory and its `src` directory, and +existing interpreter search paths. Console-script and `python -m flake8` +invocations use the same project roots. + +Detection is deliberately conservative: unavailable dependencies, dynamic +exports or package paths, re-exports, and opaque native modules are left +unchecked. Relative imports, imports from `.settings`, and files declaring +`__all__` retain their existing exemptions. Split imports are checked separately. + ### COP017: immutable variables by default Variables are treated as immutable: reassigning a name with `=` in the same scope is a violation. When reassignment is intended, allow it explicitly with the `Mutable` annotation on the first binding: diff --git a/src/community_of_python_flake8_plugin/checks/disabled/__init__.py b/src/community_of_python_flake8_plugin/checks/disabled/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/src/community_of_python_flake8_plugin/checks/disabled/module_import_many_names.py b/src/community_of_python_flake8_plugin/checks/disabled/module_import_many_names.py deleted file mode 100644 index 003b949..0000000 --- a/src/community_of_python_flake8_plugin/checks/disabled/module_import_many_names.py +++ /dev/null @@ -1,90 +0,0 @@ -"""Check is disabled because it doesn't work when flake8 invoked directly, not as python module. - -(.venv/bin/flake8 doesn't work, python -m flake8 works). - - -""" - -from __future__ import annotations -import ast -import importlib.machinery -import typing -from importlib import util as importlib_util - -from community_of_python_flake8_plugin import constants -from community_of_python_flake8_plugin.violation_codes import ViolationCodes -from community_of_python_flake8_plugin.violations import Violation - - -def check_module_has_all_declaration(module_node: ast.Module) -> bool: - for one_statement in module_node.body: - if isinstance(one_statement, ast.Assign) and any( - isinstance(one_target_element, ast.Name) and one_target_element.id == "__all__" - for one_target_element in one_statement.targets - ): - return True - if ( - isinstance(one_statement, ast.AnnAssign) - and isinstance(one_statement.target, ast.Name) - and one_statement.target.id == "__all__" - ): - return True - return False - - -def check_module_path_exists(module_name: str) -> bool: - try: - if "." not in module_name: - return importlib_util.find_spec(module_name) is not None - - parent, _, _child = module_name.rpartition(".") - parent_spec: typing.Final = importlib_util.find_spec(parent) # does not execute parent - if parent_spec is None: - return False - - # Parent must be a package (have places to search for submodules) - locations: typing.Final = parent_spec.submodule_search_locations - if locations is None: - return False - - # Find child without importing parent - return importlib.machinery.PathFinder.find_spec(module_name, list(locations)) is not None - except ModuleNotFoundError: - return False - - -@typing.final -class ModuleImportManyNamesCheck(ast.NodeVisitor): - def __init__(self, syntax_tree: ast.AST) -> None: - self.violations: list[Violation] = [] - self.contains_all_declaration: typing.Final[bool] = ( - check_module_has_all_declaration(syntax_tree) if isinstance(syntax_tree, ast.Module) else False - ) - - def visit_ImportFrom(self, ast_node: ast.ImportFrom) -> None: - if ast_node.module and ast_node.level == 0: - self.validate_import_size(ast_node) - self.generic_visit(ast_node) - - def validate_import_size(self, ast_node: ast.ImportFrom) -> None: - if len(ast_node.names) <= constants.MAX_IMPORT_NAMES: - return - if self.contains_all_declaration: - return - - module_name: typing.Final = ast_node.module - if module_name is not None and module_name.endswith(".settings"): - return - - if not any( - check_module_path_exists(f"{module_name}.{one_alias_element.name}") - for one_alias_element in ast_node.names - if isinstance(one_alias_element, ast.alias) and module_name is not None - ): - self.violations.append( - Violation( - line_number=ast_node.lineno, - column_number=ast_node.col_offset, - violation_code=ViolationCodes.MODULE_IMPORT_MANY_NAMES, - ) - ) diff --git a/src/community_of_python_flake8_plugin/checks/module_import_many_names.py b/src/community_of_python_flake8_plugin/checks/module_import_many_names.py new file mode 100644 index 0000000..829756d --- /dev/null +++ b/src/community_of_python_flake8_plugin/checks/module_import_many_names.py @@ -0,0 +1,182 @@ +from __future__ import annotations +import ast +import dataclasses +import pathlib +import sys +import tokenize +import typing + +from community_of_python_flake8_plugin import constants +from community_of_python_flake8_plugin.violation_codes import ViolationCodes +from community_of_python_flake8_plugin.violations import Violation + + +if typing.TYPE_CHECKING: + from cop_extensions import Mutable + + +PROJECT_MARKERS: typing.Final = ("pyproject.toml", "setup.cfg", "setup.py", ".git") +SOURCE_SUFFIXES: typing.Final = (".py", ".pyi") +LITERAL_NODES: typing.Final = (ast.Constant, ast.List, ast.Tuple, ast.Set, ast.Dict) + + +def collect_search_roots(filename: str) -> tuple[pathlib.Path, ...]: # noqa: COP006 + current_directory: typing.Final = pathlib.Path.cwd() + source_directory: typing.Final = ( + pathlib.Path(filename).absolute().parent if filename not in {"stdin", "-"} else current_directory + ) + root_candidates: typing.Final = [current_directory, current_directory / "src"] + for one_ancestor in (source_directory, *source_directory.parents): + if any((one_ancestor / one_marker).exists() for one_marker in PROJECT_MARKERS): + root_candidates.extend((one_ancestor, one_ancestor / "src")) + break + root_candidates.extend(pathlib.Path(one_search_path or current_directory) for one_search_path in sys.path) + return tuple(dict.fromkeys(one_root.absolute() for one_root in root_candidates if one_root.is_dir())) + + +@typing.final +@dataclasses.dataclass(kw_only=True, slots=True, frozen=True) +class SourceModule: + source_path: pathlib.Path | None + package_paths: tuple[pathlib.Path, ...] = () + + +def resolve_source_module(*, module_name: str, search_roots: tuple[pathlib.Path, ...]) -> SourceModule | None: + search_locations: Mutable[tuple[pathlib.Path, ...]] = search_roots + resolved_module: Mutable[SourceModule | None] = None + for one_component in module_name.split("."): + resolved_module = resolve_module_component(module_name=one_component, search_roots=search_locations) + if resolved_module is None: + return None + search_locations = resolved_module.package_paths + return resolved_module + + +def resolve_module_component(*, module_name: str, search_roots: tuple[pathlib.Path, ...]) -> SourceModule | None: + namespace_paths: typing.Final = [] + try: + for one_root in search_roots: + package_directory = one_root / module_name + # Native modules are opaque; never load them or fall through to another package. + if any(one_path.suffix in {".so", ".pyd"} for one_path in one_root.glob(f"{module_name}.*")) or any( + one_path.suffix in {".so", ".pyd"} for one_path in package_directory.glob("__init__.*") + ): + return None + for one_suffix in SOURCE_SUFFIXES: + initializer_path = package_directory / f"__init__{one_suffix}" + if initializer_path.is_file(): + if collect_symbol_names(initializer_path) is None: + return None + return SourceModule(source_path=initializer_path, package_paths=(package_directory,)) + for one_suffix in SOURCE_SUFFIXES: + module_path = one_root / f"{module_name}{one_suffix}" + if module_path.is_file(): + return SourceModule(source_path=module_path) + if package_directory.is_dir(): + namespace_paths.append(package_directory) + except OSError: + return None + return SourceModule(source_path=None, package_paths=tuple(namespace_paths)) if namespace_paths else None + + +def collect_symbol_names(source_path: pathlib.Path) -> frozenset[str] | None: + try: + with tokenize.open(source_path) as source_file: + source_tree: typing.Final = ast.parse(source_file.read()) + except (OSError, SyntaxError, UnicodeError): + return None + if any( + (isinstance(one_node, (ast.FunctionDef, ast.AsyncFunctionDef)) and one_node.name == "__getattr__") + or (isinstance(one_node, ast.Name) and one_node.id == "__path__") + for one_node in ast.walk(source_tree) + ): + return None + symbol_names: typing.Final = set() + for one_statement in source_tree.body: + if isinstance(one_statement, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)): + symbol_names.add(one_statement.name) + else: + update_assigned_symbols(symbol_names=symbol_names, statement_node=one_statement) + return frozenset(symbol_names) + + +def update_assigned_symbols(*, symbol_names: set[str], statement_node: ast.stmt) -> None: + if isinstance(statement_node, ast.Assign): + for one_target in statement_node.targets: + if isinstance(one_target, ast.Name): + if isinstance(statement_node.value, LITERAL_NODES): + symbol_names.add(one_target.id) + else: + symbol_names.discard(one_target.id) + elif isinstance(statement_node, ast.AnnAssign) and isinstance(statement_node.target, ast.Name): + if isinstance(statement_node.value, LITERAL_NODES): + symbol_names.add(statement_node.target.id) + else: + symbol_names.discard(statement_node.target.id) + elif isinstance(statement_node, (ast.Import, ast.ImportFrom)): + for one_alias in statement_node.names: + symbol_names.discard(one_alias.asname or one_alias.name.split(".")[0]) + + +def check_module_has_all_declaration(module_node: ast.AST) -> bool: + return isinstance(module_node, ast.Module) and any( + ( + isinstance(one_statement, ast.Assign) + and any( + isinstance(one_target, ast.Name) and one_target.id == "__all__" for one_target in one_statement.targets + ) + ) + or ( + isinstance(one_statement, ast.AnnAssign) + and isinstance(one_statement.target, ast.Name) + and one_statement.target.id == "__all__" + ) + for one_statement in module_node.body + ) + + +@typing.final +@dataclasses.dataclass(kw_only=True, slots=True, frozen=True) +class ModuleImportManyNamesCheck(ast.NodeVisitor): + syntax_tree: ast.AST + filename: str = "stdin" # noqa: COP004 + violations: list[Violation] = dataclasses.field(default_factory=list) + + def visit_ImportFrom(self, ast_node: ast.ImportFrom) -> None: + if ( + ast_node.module + and ast_node.level == 0 + and len(ast_node.names) > constants.MAX_IMPORT_NAMES + and ast_node.module.split(".")[0] not in sys.stdlib_module_names + and ast_node.module not in constants.ALLOWED_STDLIB_FROM_IMPORTS + and not ast_node.module.endswith(".settings") + and not check_module_has_all_declaration(self.syntax_tree) + ): + self.validate_import_size(ast_node) + self.generic_visit(ast_node) + + def validate_import_size(self, ast_node: ast.ImportFrom) -> None: + if ast_node.module is None: + return + imported_module: typing.Final = resolve_source_module( + module_name=ast_node.module, search_roots=collect_search_roots(self.filename) + ) + if imported_module is None or imported_module.source_path is None: + return + symbol_names: typing.Final = collect_symbol_names(imported_module.source_path) + if symbol_names is None: + return + ordinary_names: typing.Final = [] + for one_alias in ast_node.names: + if one_alias.name in symbol_names: + ordinary_names.append(one_alias.name) + elif resolve_source_module(module_name=one_alias.name, search_roots=imported_module.package_paths) is None: + return + if len(ordinary_names) > constants.MAX_IMPORT_NAMES: + self.violations.append( + Violation( + line_number=ast_node.lineno, + column_number=ast_node.col_offset, + violation_code=ViolationCodes.MODULE_IMPORT_MANY_NAMES, + ) + ) diff --git a/src/community_of_python_flake8_plugin/checks/name_length.py b/src/community_of_python_flake8_plugin/checks/name_length.py index f1746b1..0509fc6 100644 --- a/src/community_of_python_flake8_plugin/checks/name_length.py +++ b/src/community_of_python_flake8_plugin/checks/name_length.py @@ -3,12 +3,8 @@ import dataclasses import typing +from community_of_python_flake8_plugin import utils from community_of_python_flake8_plugin.constants import FINAL_CLASS_EXCLUDED_BASES, MIN_NAME_LENGTH -from community_of_python_flake8_plugin.utils import ( - check_inherits_from_bases, - find_parent_class_definition, - find_parent_function_definition, -) from community_of_python_flake8_plugin.violation_codes import ViolationCodes from community_of_python_flake8_plugin.violations import Violation @@ -58,7 +54,9 @@ class COP004NameLengthCheck(ast.NodeVisitor): def visit_AnnAssign(self, ast_node: ast.AnnAssign) -> None: if isinstance(ast_node.target, ast.Name): self.validate_name_length( - ast_node.target.id, ast_node, parent_class=find_parent_class_definition(self.syntax_tree, ast_node) + ast_node.target.id, + ast_node, + parent_class=utils.find_parent_class_definition(self.syntax_tree, ast_node), ) self.generic_visit(ast_node) @@ -66,17 +64,17 @@ def visit_Assign(self, ast_node: ast.Assign) -> None: for one_target in ast_node.targets: if isinstance(one_target, ast.Name): self.validate_name_length( - one_target.id, ast_node, parent_class=find_parent_class_definition(self.syntax_tree, ast_node) + one_target.id, ast_node, parent_class=utils.find_parent_class_definition(self.syntax_tree, ast_node) ) self.generic_visit(ast_node) def visit_FunctionDef(self, ast_node: ast.FunctionDef) -> None: - self.validate_function_name(ast_node, find_parent_class_definition(self.syntax_tree, ast_node)) + self.validate_function_name(ast_node, utils.find_parent_class_definition(self.syntax_tree, ast_node)) self.validate_function_args(ast_node) self.generic_visit(ast_node) def visit_AsyncFunctionDef(self, ast_node: ast.AsyncFunctionDef) -> None: - self.validate_function_name(ast_node, find_parent_class_definition(self.syntax_tree, ast_node)) + self.validate_function_name(ast_node, utils.find_parent_class_definition(self.syntax_tree, ast_node)) self.validate_function_args(ast_node) self.generic_visit(ast_node) @@ -206,13 +204,13 @@ def validate_name_length(self, identifier: str, ast_node: ast.stmt, *, parent_cl if ( parent_class and isinstance(ast_node, (ast.AnnAssign, ast.Assign)) - and check_inherits_from_bases(parent_class, FINAL_CLASS_EXCLUDED_BASES) + and utils.check_inherits_from_bases(parent_class, FINAL_CLASS_EXCLUDED_BASES) ): return if len(identifier) < MIN_NAME_LENGTH: # Determine if this is an attribute (inside a class but not in a method) or variable - parent_function: typing.Final = find_parent_function_definition(self.syntax_tree, ast_node) + parent_function: typing.Final = utils.find_parent_function_definition(self.syntax_tree, ast_node) # It's an attribute only if it's in a class but NOT in a function/method is_attribute: typing.Final = parent_class is not None and parent_function is None @@ -234,7 +232,7 @@ def validate_function_name( return if check_is_ignored_name(ast_node.name): return - if parent_class and check_inherits_from_bases(parent_class, FINAL_CLASS_EXCLUDED_BASES): + if parent_class and utils.check_inherits_from_bases(parent_class, FINAL_CLASS_EXCLUDED_BASES): return if check_is_pytest_fixture(ast_node): return diff --git a/src/community_of_python_flake8_plugin/plugin.py b/src/community_of_python_flake8_plugin/plugin.py index ec366bc..05e168a 100644 --- a/src/community_of_python_flake8_plugin/plugin.py +++ b/src/community_of_python_flake8_plugin/plugin.py @@ -6,6 +6,7 @@ import typing import community_of_python_flake8_plugin.checks as checks_module +from community_of_python_flake8_plugin.checks import module_import_many_names if typing.TYPE_CHECKING: @@ -16,7 +17,8 @@ class PluginCheckProtocol(typing.Protocol): - violations: list[Violation] + @property + def violations(self) -> list[Violation]: ... def __init__(self, *, syntax_tree: ast.AST) -> None: ... def visit(self, node: ast.AST) -> None: ... # noqa: COP007,COP006,COP012 @@ -27,8 +29,9 @@ class CommunityOfPythonFlake8Plugin: name: typing.Final[str] = str(pathlib.Path(__file__).parent.name) # noqa: COP004 version: typing.Final[str] = importlib.metadata.version(name) # noqa: COP004 - def __init__(self, tree: ast.AST) -> None: # noqa: COP006 + def __init__(self, tree: ast.AST, filename: str = "stdin") -> None: # noqa: COP006 self.ast_syntax_tree: typing.Final[ast.AST] = tree + self.source_filename: typing.Final = filename def run(self) -> Iterable[tuple[int, int, str, type[object]]]: # noqa: COP007 for one_check_instance in self._collect_checks(): @@ -48,7 +51,14 @@ def _collect_checks(self) -> list[PluginCheckProtocol]: for one_attribute_name in dir(imported_module): attribute = getattr(imported_module, one_attribute_name) if isinstance(attribute, type) and one_attribute_name.endswith("Check") and hasattr(attribute, "visit"): - check_instance = attribute(syntax_tree=self.ast_syntax_tree) + check_constructor = typing.cast("type[PluginCheckProtocol]", attribute) + check_instance: PluginCheckProtocol = ( + module_import_many_names.ModuleImportManyNamesCheck( + syntax_tree=self.ast_syntax_tree, filename=self.source_filename + ) + if attribute is module_import_many_names.ModuleImportManyNamesCheck + else check_constructor(syntax_tree=self.ast_syntax_tree) + ) check_instance.visit(self.ast_syntax_tree) checks_collection.append(check_instance) return checks_collection diff --git a/tests/test_module_imports.py b/tests/test_module_imports.py new file mode 100644 index 0000000..16438ca --- /dev/null +++ b/tests/test_module_imports.py @@ -0,0 +1,221 @@ +from __future__ import annotations +import ast +import importlib.machinery +import importlib.util +import os +import pathlib +import subprocess +import sys +import typing + +import pytest + +from community_of_python_flake8_plugin.checks import module_import_many_names +from community_of_python_flake8_plugin.plugin import CommunityOfPythonFlake8Plugin + + +SYMBOL_SOURCE: typing.Final = "class FirstClass: pass\nclass SecondClass: pass\nclass ThirdClass: pass\n" + + +def collect_import_codes(*, source_code: str, filename: pathlib.Path) -> list[str]: # noqa: COP006 + return [ + one_result[2].split()[0] + for one_result in CommunityOfPythonFlake8Plugin(ast.parse(source_code), str(filename)).run() + if one_result[2].startswith("COP001") + ] + + +@pytest.mark.parametrize( + ("source_code", "expected_codes"), + [ + ("from fixture_module import FirstClass, SecondClass, ThirdClass", ["COP001"]), + ("from fixture_module import FirstClass as RenamedClass, SecondClass, ThirdClass", ["COP001"]), + ("from fixture_module import FirstClass, SecondClass", []), + ("from fixture_module import FirstClass, SecondClass, UnknownClass", []), + ("from unavailable_package import alpha, beta, gamma", []), + ("from .fixture_module import FirstClass, SecondClass, ThirdClass", []), + ("from fixture_module import FirstClass, SecondClass, ThirdClass\n__all__ = ['unrelated_export']", []), + ("from fixture_module import FirstClass, SecondClass, ThirdClass\n__all__: list[str] = []", []), + ], +) +def test_classify_source_symbols( + tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch, source_code: str, expected_codes: list[str] +) -> None: + monkeypatch.chdir(tmp_path) + (tmp_path / "fixture_module.py").write_text(SYMBOL_SOURCE) + assert collect_import_codes(source_code=source_code, filename=tmp_path / "consumer.py") == expected_codes + + +@pytest.mark.parametrize( + "source_code", + [ + "from importlib import resources, metadata, util", + "from collections.abc import Iterable, Iterator, Mapping", + "from __future__ import annotations, division, print_function", + "from os import name, environ, getcwd", + "from xml.etree.ElementTree import Element, SubElement, tostring", + ], +) +def test_preserve_stdlib_imports(source_code: str, tmp_path: pathlib.Path) -> None: + assert collect_import_codes(source_code=source_code, filename=tmp_path / "consumer.py") == [] + + +@pytest.mark.parametrize("initializer", ["", "raise RuntimeError('must not execute')\n"]) +def test_skip_module_imports(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch, initializer: str) -> None: # noqa: COP006 + monkeypatch.chdir(tmp_path) + package_directory: typing.Final = tmp_path / "fixture_package" + package_directory.mkdir() + (package_directory / "__init__.py").write_text(initializer) + for one_module in ("alpha", "beta", "gamma"): + (package_directory / f"{one_module}.py").write_text("raise RuntimeError('must not execute')\n") + assert ( + collect_import_codes( + source_code="from fixture_package import alpha, beta, gamma", filename=tmp_path / "consumer.py" + ) + == [] + ) + + +def test_detect_mixed_imports(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(tmp_path) + package_directory: typing.Final = tmp_path / "fixture_package" + package_directory.mkdir() + (package_directory / "__init__.py").write_text(SYMBOL_SOURCE) + (package_directory / "submodule.py").write_text("") + assert collect_import_codes( + source_code="from fixture_package import submodule, FirstClass, SecondClass, ThirdClass", + filename=tmp_path / "consumer.py", + ) == ["COP001"] + + +def test_never_execute_package_code(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(tmp_path) + package_directory: typing.Final = tmp_path / "fixture_package" + package_directory.mkdir() + (package_directory / "__init__.py").write_text( + "from pathlib import Path\n" + "Path(__file__).with_name('executed').touch()\n" + "raise RuntimeError('must not execute')\n" + ) + (package_directory / "definitions.py").write_text(SYMBOL_SOURCE + "raise RuntimeError('must not execute')\n") + + def reject_runtime_lookup(*arguments: object, **parameters: object) -> typing.NoReturn: # noqa: ARG001 + raise AssertionError("runtime import lookup must not be used") + + monkeypatch.setattr(importlib.util, "find_spec", reject_runtime_lookup) + monkeypatch.setattr(importlib.machinery.PathFinder, "find_spec", reject_runtime_lookup) + check_instance: typing.Final = module_import_many_names.ModuleImportManyNamesCheck( + syntax_tree=ast.parse("from fixture_package.definitions import FirstClass, SecondClass, ThirdClass") + ) + check_instance.visit(check_instance.syntax_tree) + assert [one_violation.violation_code.code for one_violation in check_instance.violations] == ["COP001"] + assert not (package_directory / "executed").exists() + assert "fixture_package" not in sys.modules + + +@pytest.mark.parametrize( + "source_code", + [ + SYMBOL_SOURCE + "\ndef __getattr__(attribute_name): return None\n", + SYMBOL_SOURCE + "\n__path__ = calculate_search_path()\n", + SYMBOL_SOURCE + "\nFirstClass = calculate_dynamic_export()\n", + SYMBOL_SOURCE + "\nfrom dynamic_package import export as FirstClass\n", + "not valid python!", + ], +) +def test_skip_dynamic_or_invalid_modules( + tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch, source_code: str +) -> None: + monkeypatch.chdir(tmp_path) + (tmp_path / "fixture_module.py").write_text(source_code) + assert ( + collect_import_codes( + source_code="from fixture_module import FirstClass, SecondClass, ThirdClass", + filename=tmp_path / "consumer.py", + ) + == [] + ) + + +@pytest.mark.parametrize("source_suffix", [".py", ".pyi"]) +def test_detect_installed_module(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch, source_suffix: str) -> None: + installed_directory: typing.Final = tmp_path / "site-packages" + installed_directory.mkdir() + (installed_directory / f"fixture_module{source_suffix}").write_text(SYMBOL_SOURCE) + monkeypatch.syspath_prepend(str(installed_directory)) + assert collect_import_codes( + source_code="from fixture_module import FirstClass, SecondClass, ThirdClass", filename=tmp_path / "consumer.py" + ) == ["COP001"] + + +def test_skip_native_module(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(tmp_path) + (tmp_path / "fixture_module.so").write_bytes(b"not an executable module") + (tmp_path / "fixture_module.py").write_text(SYMBOL_SOURCE) + assert ( + collect_import_codes( + source_code="from fixture_module import FirstClass, SecondClass, ThirdClass", + filename=tmp_path / "consumer.py", + ) + == [] + ) + + +def test_skip_dynamic_parent_paths(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(tmp_path) + package_directory: typing.Final = tmp_path / "fixture_package" + package_directory.mkdir() + (package_directory / "__init__.py").write_text("__path__ = calculate_dynamic_path()\n") + (package_directory / "definitions.py").write_text(SYMBOL_SOURCE) + assert ( + collect_import_codes( + source_code="from fixture_package.definitions import FirstClass, SecondClass, ThirdClass", + filename=tmp_path / "consumer.py", + ) + == [] + ) + + +def test_never_evaluate_pth_files(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(tmp_path) + (tmp_path / "fixture_module.py").write_text(SYMBOL_SOURCE) + (tmp_path / "fixture.pth").write_text("import pathlib; pathlib.Path('executed').touch()\n") + assert collect_import_codes( + source_code="from fixture_module import FirstClass, SecondClass, ThirdClass", filename=tmp_path / "consumer.py" + ) == ["COP001"] + assert not (tmp_path / "executed").exists() + + +@pytest.mark.parametrize("source_layout", ["flat", "src", "namespace"]) +def test_compare_flake8_entrypoints(tmp_path: pathlib.Path, source_layout: str) -> None: + project_directory: typing.Final = tmp_path / "project" + project_directory.mkdir() + (project_directory / "pyproject.toml").write_text("[project]\nname = 'fixture-project'\nversion = '0.0.0'\n") + package_directory: typing.Final = ( + project_directory / "src" / "fixture_package" + if source_layout == "src" + else project_directory / "fixture_package" + ) + package_directory.mkdir(parents=True) + if source_layout != "namespace": + (package_directory / "__init__.py").write_text("raise RuntimeError('must not execute')\n") + (package_directory / "definitions.py").write_text(SYMBOL_SOURCE) + consumer_path: typing.Final = package_directory / "consumer.py" + consumer_path.write_text("from fixture_package.definitions import FirstClass, SecondClass, ThirdClass\n") + console_path: typing.Final = pathlib.Path(sys.executable).parent / "flake8" + runtime_environment: typing.Final = dict(os.environ, PYTHONDONTWRITEBYTECODE="1") + command_outputs: typing.Final = [] + # Run outside the project so filename-based roots are exercised as well. + for one_command in ([str(console_path)], [sys.executable, "-m", "flake8"]): + completed_process = subprocess.run( # noqa: S603 + [*one_command, "--isolated", "--select=COP001", "--jobs=1", str(consumer_path)], + cwd=tmp_path, + env=runtime_environment, + check=False, + capture_output=True, + text=True, + ) + assert completed_process.returncode == 1, completed_process.stderr + assert "COP001" in completed_process.stdout + command_outputs.append(completed_process.stdout) + assert command_outputs[0] == command_outputs[1] diff --git a/tests/test_plugin.py b/tests/test_plugin.py index 3debfd9..d3e41a8 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -55,12 +55,11 @@ def test_import_stdlib_validations(input_source: str, expected_output: list[str] ) == sorted(expected_output) -@pytest.mark.skip("disabled") @pytest.mark.parametrize( ("input_source", "expected_output"), [ - # COP001: Use module import when importing more than two names - ("from x import a, b, c", ["COP001"]), + # COP001: Unavailable modules cannot be classified statically + ("from x import a, b, c", []), # Unresolved modules are left unchecked. # COP001: Importing more than two names from third party (but settings exempt) ("from my_project.settings import A, B, C", []), # No violation: importlib resources submodules are allowed From f2df2015e0347f21b132c01559a71faa38dfb65e Mon Sep 17 00:00:00 2001 From: Lev Vereshchagin Date: Mon, 12 Oct 2026 14:04:10 +0300 Subject: [PATCH 2/2] Make COP001 opt-in within the existing extension --- AGENTS.md | 3 ++ README.md | 9 ++++ pyproject.toml | 1 + .../plugin.py | 14 ++++- tests/test_module_imports.py | 51 ++++++++++++++++--- 5 files changed, 69 insertions(+), 9 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 2ec97a6..5582acb 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -4,3 +4,6 @@ Import checks must inspect AST and filesystem data without executing inspected packages, import hooks, extension modules, or `.pth` files. Do not use runtime imports or `find_spec` to resolve application dependencies. Leave unresolved or dynamic imports unchecked rather than guessing their type. + +COP001 must remain off by default. Repositories opt in explicitly through +Flake8's `extend-select = ["COP001"]` setting. Keep one COP extension. diff --git a/README.md b/README.md index 94d8dbb..f6a3ad0 100644 --- a/README.md +++ b/README.md @@ -26,6 +26,15 @@ This plugin implements the following code style checks: ### COP001: import modules instead of many symbols +COP001 is off by default, including when selecting the `COP` prefix. Enable it +explicitly with `--extend-select=COP001`, `--select=COP001`, or in your Flake8 configuration: + +```toml +[tool.flake8] +select = ["COP"] +extend-select = ["COP001"] +``` + Use a module import when importing more than two statically identifiable symbols from the same module. Aliases still count; imported submodules do not. Standard library imports are handled exclusively by COP002, preserving its exceptions diff --git a/pyproject.toml b/pyproject.toml index ec544f6..d79aebb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -71,6 +71,7 @@ lines-after-imports = 2 strict = true [tool.flake8] +extend-select = ["COP001"] select = ["COP"] exclude = [".venv"] show-source = true diff --git a/src/community_of_python_flake8_plugin/plugin.py b/src/community_of_python_flake8_plugin/plugin.py index 05e168a..1dd2397 100644 --- a/src/community_of_python_flake8_plugin/plugin.py +++ b/src/community_of_python_flake8_plugin/plugin.py @@ -10,6 +10,7 @@ if typing.TYPE_CHECKING: + import argparse import ast from collections.abc import Iterable @@ -28,11 +29,19 @@ def visit(self, node: ast.AST) -> None: ... # noqa: COP007,COP006,COP012 class CommunityOfPythonFlake8Plugin: name: typing.Final[str] = str(pathlib.Path(__file__).parent.name) # noqa: COP004 version: typing.Final[str] = importlib.metadata.version(name) # noqa: COP004 + cop001_enabled: typing.ClassVar[bool] = False def __init__(self, tree: ast.AST, filename: str = "stdin") -> None: # noqa: COP006 self.ast_syntax_tree: typing.Final[ast.AST] = tree self.source_filename: typing.Final = filename + @classmethod + def parse_options(cls, parsed_options: argparse.Namespace) -> None: + cls.cop001_enabled = "COP001" in ( + *(parsed_options.select or []), + *(parsed_options.extend_select or []), + ) + def run(self) -> Iterable[tuple[int, int, str, type[object]]]: # noqa: COP007 for one_check_instance in self._collect_checks(): for one_violation in one_check_instance.violations: @@ -46,18 +55,19 @@ def run(self) -> Iterable[tuple[int, int, str, type[object]]]: # noqa: COP007 def _collect_checks(self) -> list[PluginCheckProtocol]: checks_collection: typing.Final = [] for _, one_module_name, _ in pkgutil.iter_modules(checks_module.__path__): + if one_module_name == "module_import_many_names" and not self.cop001_enabled: + continue imported_module = importlib.import_module(f"{checks_module.__name__}.{one_module_name}") for one_attribute_name in dir(imported_module): attribute = getattr(imported_module, one_attribute_name) if isinstance(attribute, type) and one_attribute_name.endswith("Check") and hasattr(attribute, "visit"): - check_constructor = typing.cast("type[PluginCheckProtocol]", attribute) check_instance: PluginCheckProtocol = ( module_import_many_names.ModuleImportManyNamesCheck( syntax_tree=self.ast_syntax_tree, filename=self.source_filename ) if attribute is module_import_many_names.ModuleImportManyNamesCheck - else check_constructor(syntax_tree=self.ast_syntax_tree) + else typing.cast("type[PluginCheckProtocol]", attribute)(syntax_tree=self.ast_syntax_tree) ) check_instance.visit(self.ast_syntax_tree) checks_collection.append(check_instance) diff --git a/tests/test_module_imports.py b/tests/test_module_imports.py index 16438ca..a563468 100644 --- a/tests/test_module_imports.py +++ b/tests/test_module_imports.py @@ -11,18 +11,17 @@ import pytest from community_of_python_flake8_plugin.checks import module_import_many_names -from community_of_python_flake8_plugin.plugin import CommunityOfPythonFlake8Plugin SYMBOL_SOURCE: typing.Final = "class FirstClass: pass\nclass SecondClass: pass\nclass ThirdClass: pass\n" def collect_import_codes(*, source_code: str, filename: pathlib.Path) -> list[str]: # noqa: COP006 - return [ - one_result[2].split()[0] - for one_result in CommunityOfPythonFlake8Plugin(ast.parse(source_code), str(filename)).run() - if one_result[2].startswith("COP001") - ] + check_instance: typing.Final = module_import_many_names.ModuleImportManyNamesCheck( + syntax_tree=ast.parse(source_code), filename=str(filename) + ) + check_instance.visit(check_instance.syntax_tree) + return [one_violation.violation_code.code for one_violation in check_instance.violations] @pytest.mark.parametrize( @@ -208,7 +207,14 @@ def test_compare_flake8_entrypoints(tmp_path: pathlib.Path, source_layout: str) # Run outside the project so filename-based roots are exercised as well. for one_command in ([str(console_path)], [sys.executable, "-m", "flake8"]): completed_process = subprocess.run( # noqa: S603 - [*one_command, "--isolated", "--select=COP001", "--jobs=1", str(consumer_path)], + [ + *one_command, + "--isolated", + "--select=COP001", + "--extend-select=COP001", + "--jobs=1", + str(consumer_path), + ], cwd=tmp_path, env=runtime_environment, check=False, @@ -219,3 +225,34 @@ def test_compare_flake8_entrypoints(tmp_path: pathlib.Path, source_layout: str) assert "COP001" in completed_process.stdout command_outputs.append(completed_process.stdout) assert command_outputs[0] == command_outputs[1] + + +@pytest.mark.parametrize("opt_in_mode", ["disabled", "cli", "config", "exact-select"]) +def test_require_explicit_opt_in(tmp_path: pathlib.Path, opt_in_mode: str) -> None: + (tmp_path / "fixture_module.py").write_text(SYMBOL_SOURCE) + consumer_path: typing.Final = tmp_path / "consumer.py" + consumer_path.write_text( + "from fixture_module import FirstClass, SecondClass, ThirdClass\nasync def get_something(): pass\n" + ) + flake8_arguments: typing.Final = [ + "--select=COP001" if opt_in_mode == "exact-select" else "--select=COP", + "--jobs=1", + str(consumer_path), + ] + if opt_in_mode == "config": + (tmp_path / "pyproject.toml").write_text('[tool.flake8]\nselect = ["COP"]\nextend-select = ["COP001"]\n') + else: + flake8_arguments.append("--isolated") + if opt_in_mode == "cli": + flake8_arguments.append("--extend-select=COP001") + completed_process: typing.Final = subprocess.run( # noqa: S603 + [sys.executable, "-m", "flake8", *flake8_arguments], + cwd=tmp_path, + check=False, + capture_output=True, + text=True, + ) + assert completed_process.returncode == 1, completed_process.stderr + if opt_in_mode != "exact-select": + assert "COP010" in completed_process.stdout + assert completed_process.stdout.count("COP001") == (0 if opt_in_mode == "disabled" else 1)