diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index 6c9244a..86888e4 100644 --- a/.github/workflows/lint.yml +++ b/.github/workflows/lint.yml @@ -47,4 +47,4 @@ jobs: - name: Run mypy for tests shell: bash - run: mypy tests + run: mypy tests --exclude tests/typing diff --git a/.gitignore b/.gitignore index 5d1b9e7..71d6b12 100644 --- a/.gitignore +++ b/.gitignore @@ -21,3 +21,4 @@ uv.lock node_modules mutants CLAUDE.md +AGENTS.md diff --git a/pristan/common_types.py b/pristan/common_types.py index 4026fe6..4a93ca8 100644 --- a/pristan/common_types.py +++ b/pristan/common_types.py @@ -1,13 +1,76 @@ -from typing import Callable, Dict, List, Optional, TypeVar, Union +import sys +from typing import ( + Any, + Callable, + Dict, + Iterator, + List, + Optional, + Protocol, + Tuple, + TypeVar, + Union, + overload, +) -try: - # TODO: ParamSpec appeared with python 3.10, so delete this try-except if python 3.9 deleted from the matrix - from typing import ParamSpec # type: ignore[attr-defined, unused-ignore] -except ImportError: # pragma: no cover - from typing_extensions import ParamSpec # type: ignore[assignment, unused-ignore] +if sys.version_info >= (3, 10): + from typing import ParamSpec # pragma: no cover +else: + from typing_extensions import ParamSpec # pragma: no cover + +SlotParameters = ParamSpec('SlotParameters') -SlotPapameters = ParamSpec('SlotPapameters') PluginResult = TypeVar('PluginResult') +SlotCallResult = TypeVar('SlotCallResult') +SlotCallResultCovariant = TypeVar('SlotCallResultCovariant', covariant=True) # noqa: PLC0105 +PluginResultCovariant = TypeVar('PluginResultCovariant', covariant=True) # noqa: PLC0105 + SlotResult = Optional[Union[List[PluginResult], Dict[str, PluginResult]]] -SlotFunction = Callable[SlotPapameters, Optional[Union[List[PluginResult], Dict[str, PluginResult]]]] # type: ignore[valid-type, misc, unused-ignore] -PluginFunction = Callable[SlotPapameters, PluginResult] # type: ignore[valid-type, misc, unused-ignore] +SlotFunction = Callable[SlotParameters, SlotCallResult] +PluginFunction = Callable[SlotParameters, PluginResult] + + +class PluginProtocol(Protocol[SlotParameters, PluginResultCovariant]): # pragma: no cover + name: str + requested_name: str + + def __call__(self, *args: SlotParameters.args, **kwargs: SlotParameters.kwargs) -> PluginResultCovariant: ... + + +class BaseSlotViewProtocol(Protocol[SlotParameters, SlotCallResultCovariant, PluginResultCovariant]): # pragma: no cover + def __call__(self, *args: SlotParameters.args, **kwargs: SlotParameters.kwargs) -> SlotCallResultCovariant: ... + + def __iter__(self) -> Iterator[PluginProtocol[SlotParameters, PluginResultCovariant]]: ... + + def __bool__(self) -> bool: ... + + def __len__(self) -> int: ... + + +class SlotSelectionProtocol(BaseSlotViewProtocol[SlotParameters, SlotCallResultCovariant, PluginResultCovariant], Protocol[SlotParameters, SlotCallResultCovariant, PluginResultCovariant]): + pass + + +class SlotProtocol(BaseSlotViewProtocol[SlotParameters, SlotCallResultCovariant, PluginResult], Protocol[SlotParameters, SlotCallResultCovariant, PluginResult]): # pragma: no cover + @overload + def plugin(self, plugin_function_or_name: Optional[str] = None, unique: bool = False, engine: Optional[Union[List[str], str]] = None, run_once: bool = False) -> Callable[[Callable[SlotParameters, PluginResult]], Callable[SlotParameters, PluginResult]]: ... + + @overload + def plugin(self, plugin_function_or_name: Callable[SlotParameters, PluginResult], unique: bool = False, engine: Optional[Union[List[str], str]] = None, run_once: bool = False) -> Callable[SlotParameters, PluginResult]: ... + + def keys(self) -> Tuple[str, ...]: ... + + def __getitem__(self, key: str) -> SlotSelectionProtocol[SlotParameters, SlotCallResultCovariant, PluginResult]: ... + + def __contains__(self, item: object) -> bool: ... + + +class SlotDecoratorProtocol(Protocol): # pragma: no cover + @overload + def __call__(self, function: Callable[SlotParameters, List[PluginResult]], /) -> SlotProtocol[SlotParameters, List[PluginResult], PluginResult]: ... + + @overload + def __call__(self, function: Callable[SlotParameters, Dict[str, PluginResult]], /) -> SlotProtocol[SlotParameters, Dict[str, PluginResult], PluginResult]: ... + + @overload + def __call__(self, function: Callable[SlotParameters, None], /) -> SlotProtocol[SlotParameters, None, Any]: ... diff --git a/pristan/components/plugin.py b/pristan/components/plugin.py index 35cb5ce..3e30dba 100644 --- a/pristan/components/plugin.py +++ b/pristan/components/plugin.py @@ -5,7 +5,7 @@ from printo import repred from simtypes import check -from pristan.common_types import PluginFunction, PluginResult, SlotPapameters +from pristan.common_types import PluginFunction, PluginResult, SlotParameters from pristan.components.slot_code_representer import sentinel as return_type_sentinel from pristan.errors import NumberOfCallsError @@ -13,7 +13,7 @@ @repred(positionals=['name']) class Plugin(Generic[PluginResult]): # TODO: consider to delete this "type: ignore" if python 3.9 deleted from the matrix - def __init__(self, name: str, plugin_function: PluginFunction[SlotPapameters, PluginResult], expected_result_type: Union[InnerNoneType, Type[Any]], type_check: bool, unique: bool, run_once: bool = False) -> None: # type: ignore[type-arg, unused-ignore] # noqa: PLR0913 + def __init__(self, name: str, plugin_function: PluginFunction[SlotParameters, PluginResult], expected_result_type: Union[InnerNoneType, Type[Any]], type_check: bool, unique: bool, run_once: bool = False) -> None: # noqa: PLR0913 self.plugin_function = plugin_function self.requested_name = name self.name = name @@ -24,7 +24,7 @@ def __init__(self, name: str, plugin_function: PluginFunction[SlotPapameters, Pl self.call_count = 0 self.lock = Lock() - def __call__(self, *args: SlotPapameters.args, **kwargs: SlotPapameters.kwargs) -> PluginResult: + def __call__(self, *args: SlotParameters.args, **kwargs: SlotParameters.kwargs) -> PluginResult: if self.run_once: with self.lock: if self.call_count: @@ -32,12 +32,12 @@ def __call__(self, *args: SlotPapameters.args, **kwargs: SlotPapameters.kwargs) self.call_count += 1 # TODO: try to delete this "type: ignore" comments if python 3.8 deleted from CI - result = self.plugin_function(*args, **kwargs) # type: ignore[arg-type, unused-ignore] + result = self.plugin_function(*args, **kwargs) # type: ignore[arg-type] if self.type_check and self.expected_result_type is not return_type_sentinel and not check(result, self.expected_result_type, strict=True): # type: ignore[arg-type] - raise TypeError(f'The type {type(result).__name__} of the plugin\'s "{self.name}" return value {result!r} does not match the expected type {self._get_class_name(self.expected_result_type)}.') # type: ignore[union-attr, unused-ignore] + raise TypeError(f'The type {type(result).__name__} of the plugin\'s "{self.name}" return value {result!r} does not match the expected type {self._get_class_name(self.expected_result_type)}.') - return result # type: ignore[no-any-return, unused-ignore] + return result def set_name(self, name: str) -> None: self.name = name diff --git a/pristan/components/plugins_group.py b/pristan/components/plugins_group.py index 7f1f4c9..7a3cf74 100644 --- a/pristan/components/plugins_group.py +++ b/pristan/components/plugins_group.py @@ -15,7 +15,7 @@ class PluginsGroup(Generic[PluginResult]): All collection operations are totally not thread-safe. """ - def __init__(self, caller: 'SlotCaller', plugins: Optional[List[Plugin[PluginResult]]] = None) -> None: # type: ignore[name-defined] # noqa: F821 + def __init__(self, caller: 'SlotCaller[PluginResult]', plugins: Optional[List[Plugin[PluginResult]]] = None) -> None: # type: ignore[name-defined] # noqa: F821 self.caller = caller self.plugins: List[Plugin[PluginResult]] = [] self.plugins_by_requested_names: DefaultDict[str, List[Plugin[PluginResult]]] = defaultdict(list) @@ -57,7 +57,7 @@ def __contains__(self, item: Any) -> bool: raise TypeError('Checking for inclusion is only possible for strings of a valid format or for plugin objects.') - def __getitem__(self, key: str) -> 'CallerWithPlugins': # type: ignore[name-defined] # noqa: F821 + def __getitem__(self, key: str) -> 'CallerWithPlugins[PluginResult]': # type: ignore[name-defined] # noqa: F821 from pristan.components.slot_caller import CallerWithPlugins # noqa: PLC0415 if isinstance(key, str): diff --git a/pristan/components/slot.py b/pristan/components/slot.py index 1674ccc..db30481 100644 --- a/pristan/components/slot.py +++ b/pristan/components/slot.py @@ -1,10 +1,10 @@ try: from importlib_metadata import ( # type: ignore[import-not-found, unused-ignore] - entry_points, # type: ignore[import-not-found, unused-ignore] + entry_points, ) -except ImportError: # type: ignore[assignment, unused-ignore] # pragma: no cover +except ImportError: # pragma: no cover from importlib.metadata import ( # type: ignore[assignment, unused-ignore] - entry_points, # type: ignore[assignment, unused-ignore] + entry_points, ) from threading import RLock @@ -26,9 +26,10 @@ from pristan.common_types import ( PluginFunction, + PluginProtocol, PluginResult, SlotFunction, - SlotPapameters, + SlotParameters, SlotResult, ) from pristan.components.plugin import Plugin @@ -55,7 +56,7 @@ }, ) class Slot(Generic[PluginResult]): - def __init__(self, slot_function: SlotFunction[SlotPapameters, SlotResult[PluginResult]], signature: Optional[str], slot_name: Optional[str], max: Optional[int], type_check: bool, entrypoint_group: str) -> None: # type: ignore[type-arg, unused-ignore] # noqa: PLR0913, A002 + def __init__(self, slot_function: SlotFunction[SlotParameters, SlotResult[PluginResult]], signature: Optional[str], slot_name: Optional[str], max: Optional[int], type_check: bool, entrypoint_group: str) -> None: # noqa: PLR0913, A002 if max is not None and max < 0: raise ValueError('The maximum number of plugins cannot be less than zero.') @@ -78,15 +79,15 @@ def __init__(self, slot_function: SlotFunction[SlotPapameters, SlotResult[Plugin self.plugins: PluginsGroup[PluginResult] = PluginsGroup(self.caller) self.backed_caller = CallerWithPlugins(self.caller, self.plugins.plugins) - self._compare_signatures(self.slot_function, self.slot_function) # type: ignore[arg-type, unused-ignore] + self._compare_signatures(self.slot_function, self.slot_function) # type: ignore[arg-type] self.loaded = False - def __call__(self, *args: SlotPapameters.args, **kwargs: SlotPapameters.kwargs) -> SlotResult[PluginResult]: + def __call__(self, *args: SlotParameters.args, **kwargs: SlotParameters.kwargs) -> SlotResult[PluginResult]: self._load_entrypoints() return self.backed_caller(*args, **kwargs) - def __iter__(self) -> Generator[Plugin[PluginResult], None, None]: + def __iter__(self) -> Generator[PluginProtocol[SlotParameters, PluginResult], None, None]: self._load_entrypoints() yield from self.plugins @@ -101,18 +102,17 @@ def __len__(self) -> int: return len(self.plugins) @overload - def plugin(self, plugin_function_or_name: Optional[str], unique: bool = False, engine: Optional[Union[List[str], str]] = None) -> Callable[[PluginFunction[SlotPapameters, PluginResult]], PluginFunction[SlotPapameters, PluginResult]]: # type: ignore[type-arg, unused-ignore] + def plugin(self, plugin_function_or_name: Optional[str] = None, unique: bool = False, engine: Optional[Union[List[str], str]] = None, run_once: bool = False) -> Callable[[Callable[SlotParameters, PluginResult]], Callable[SlotParameters, PluginResult]]: ... # pragma: no cover @overload - def plugin(self, plugin_function_or_name: PluginFunction[SlotPapameters, PluginResult], unique: bool = False, engine: Optional[Union[List[str], str]] = None) -> PluginFunction[SlotPapameters, PluginResult]: # type: ignore[type-arg, unused-ignore] - ... # pragma: no cover + def plugin(self, plugin_function_or_name: Callable[SlotParameters, PluginResult], unique: bool = False, engine: Optional[Union[List[str], str]] = None, run_once: bool = False) -> Callable[SlotParameters, PluginResult]: ... # pragma: no cover - def plugin(self, plugin_function_or_name: Optional[Union[PluginFunction[SlotPapameters, PluginResult], str]] = None, unique: bool = False, engine: Optional[Union[List[str], str]] = None, run_once: bool = False) -> Union[Callable[[PluginFunction[SlotPapameters, PluginResult]], PluginFunction[SlotPapameters, PluginResult]], PluginFunction[SlotPapameters, PluginResult]]: # type: ignore[type-arg, unused-ignore] + def plugin(self, plugin_function_or_name: Optional[Union[PluginFunction[SlotParameters, PluginResult], str]] = None, unique: bool = False, engine: Optional[Union[List[str], str]] = None, run_once: bool = False) -> Union[Callable[[Callable[SlotParameters, PluginResult]], Callable[SlotParameters, PluginResult]], Callable[SlotParameters, PluginResult]]: if isinstance(plugin_function_or_name, str): if not plugin_function_or_name.isidentifier(): raise ValueError('The plugin name must be a valid Python identifier.') - get_plugin_name: Callable[[PluginFunction[SlotPapameters, PluginResult]], str] = lambda function: plugin_function_or_name # type: ignore[type-arg, unused-ignore] # noqa: E731, ARG005 + get_plugin_name: Callable[[PluginFunction[SlotParameters, PluginResult]], str] = lambda function: plugin_function_or_name # noqa: E731, ARG005 elif callable(plugin_function_or_name): get_plugin_name = lambda function: plugin_function_or_name.__name__ # noqa: E731, ARG005 @@ -123,9 +123,9 @@ def plugin(self, plugin_function_or_name: Optional[Union[PluginFunction[SlotPapa else: raise TypeError('Only a function or plugin name followed by a function can be passed to the decorator.') - def decorator(plugin_function: PluginFunction[SlotPapameters, PluginResult]) -> PluginFunction[SlotPapameters, PluginResult]: # type: ignore[type-arg, unused-ignore] + def decorator(plugin_function: Callable[SlotParameters, PluginResult]) -> Callable[SlotParameters, PluginResult]: # TODO: consider to delete this "type: ignore" if python 3.8 deleted from the matrix - self._compare_signatures(self.slot_function, plugin_function) # type: ignore[arg-type, unused-ignore] + self._compare_signatures(self.slot_function, plugin_function) # type: ignore[arg-type] self._add_plugin(get_plugin_name(plugin_function), plugin_function, unique, engine, run_once) return plugin_function @@ -145,8 +145,8 @@ def _load_entrypoints(self) -> None: point.load() self.loaded = True - def _add_plugin(self, name: str, function: PluginFunction[SlotPapameters, PluginResult], unique: bool, engine: Optional[Union[str, List[str]]], run_once: bool) -> None: # type: ignore[type-arg, unused-ignore] - plugin: Plugin = Plugin(name, function, self.code_representation.returning_type, self.type_check, unique, run_once) # type: ignore[type-arg] + def _add_plugin(self, name: str, function: PluginFunction[SlotParameters, PluginResult], unique: bool, engine: Optional[Union[str, List[str]]], run_once: bool) -> None: + plugin: Plugin[PluginResult] = Plugin(name, function, self.code_representation.returning_type, self.type_check, unique, run_once) with self.lock: if len(self.plugins) == self.max_number_of_plugins: @@ -161,7 +161,7 @@ def _add_plugin(self, name: str, function: PluginFunction[SlotPapameters, Plugin self.plugins.delete_last_by_name(name) raise PrimadonnaPluginError(f'Plugin "{other_plugin.name}" claims to be unique, but there are other plugins with the same name.') - def _compare_signatures(self, slot_function: SlotFunction[SlotPapameters, SlotResult[PluginResult]], plugin_function: PluginFunction[SlotPapameters, PluginResult]) -> None: # type: ignore[type-arg, unused-ignore] + def _compare_signatures(self, slot_function: SlotFunction[SlotParameters, SlotResult[PluginResult]], plugin_function: PluginFunction[SlotParameters, PluginResult]) -> None: if self.signature is not None: PossibleCallMatcher(self.signature).match(plugin_function, raise_exception=True) elif not PossibleCallMatcher.from_callable(slot_function) & PossibleCallMatcher.from_callable(plugin_function): diff --git a/pristan/components/slot_caller.py b/pristan/components/slot_caller.py index 70843f5..b3e2fe0 100644 --- a/pristan/components/slot_caller.py +++ b/pristan/components/slot_caller.py @@ -6,7 +6,7 @@ from pristan.common_types import ( PluginResult, SlotFunction, - SlotPapameters, + SlotParameters, SlotResult, ) from pristan.components.plugin import Plugin @@ -18,13 +18,13 @@ @repred class SlotCaller(Generic[PluginResult]): # TODO: consider to delete this "type: ignore" if python 3.8 deleted from the matrix - def __init__(self, code_representation: SlotCodeRepresenter, slot_name: Optional[str], slot_function: SlotFunction[SlotPapameters, SlotResult[PluginResult]], type_check: bool) -> None: # type: ignore[type-arg, unused-ignore] + def __init__(self, code_representation: SlotCodeRepresenter, slot_name: Optional[str], slot_function: SlotFunction[SlotParameters, SlotResult[PluginResult]], type_check: bool) -> None: self.code_representation = code_representation self.slot_name = slot_name self.slot_function = slot_function self.type_check = type_check - def __call__(self, plugins: Union[PluginsGroup[PluginResult], List[Plugin[PluginResult]]], *args: SlotPapameters.args, **kwargs: SlotPapameters.kwargs) -> SlotResult[PluginResult]: # type: ignore[return] + def __call__(self, plugins: Union[PluginsGroup[PluginResult], List[Plugin[PluginResult]]], *args: SlotParameters.args, **kwargs: SlotParameters.kwargs) -> SlotResult[PluginResult]: # type: ignore[return] if not self.code_representation.is_empty and not plugins: if self.code_representation.returns_list: if self.code_representation.returning_type is return_type_sentinel: @@ -40,7 +40,7 @@ def __call__(self, plugins: Union[PluginsGroup[PluginResult], List[Plugin[Plugin returns_type = self.code_representation.returning_type # TODO: consider to delete this "type: ignore" if python 3.9 deleted from the matrix - result: SlotResult[PluginResult] = Plugin(self.slot_name if self.slot_name is not None else self.slot_function.__name__, self.slot_function, returns_type, self.type_check, False)(*args, **kwargs) # type: ignore[assignment, unused-ignore] + result: SlotResult[PluginResult] = Plugin(self.slot_name if self.slot_name is not None else self.slot_function.__name__, self.slot_function, returns_type, self.type_check, False)(*args, **kwargs) if self.code_representation.returning_type is return_type_sentinel and not self.code_representation.returns_dict and not self.code_representation.returns_list: result = None @@ -63,7 +63,7 @@ def __init__(self, caller: SlotCaller[PluginResult], plugins: List[Plugin[Plugin self.caller = caller self.plugins = plugins - def __call__(self, *args: SlotPapameters.args, **kwargs: SlotPapameters.kwargs) -> SlotResult[PluginResult]: + def __call__(self, *args: SlotParameters.args, **kwargs: SlotParameters.kwargs) -> SlotResult[PluginResult]: return self.caller(self.plugins, *args, **kwargs) def __iter__(self) -> Generator[Plugin[PluginResult], None, None]: diff --git a/pristan/decorators/slot.py b/pristan/decorators/slot.py index 43c604e..ffaa38e 100644 --- a/pristan/decorators/slot.py +++ b/pristan/decorators/slot.py @@ -1,27 +1,37 @@ from functools import partial, wraps -from typing import Callable, Optional, Union, overload - -from pristan.common_types import PluginResult, SlotFunction, SlotPapameters +from typing import Any, Callable, Dict, List, Optional, overload + +from pristan.common_types import ( + PluginResult, + SlotDecoratorProtocol, + SlotParameters, + SlotProtocol, +) from pristan.components.slot import Slot -# TODO: concider to delete all "type: ignore" comments with "unused-ignore"'s in this file if python 3.8 deleted from the matrix @overload -def slot(function: SlotFunction[SlotPapameters, PluginResult], /) -> SlotFunction[SlotPapameters, PluginResult]: ... # type: ignore[type-arg, unused-ignore] # pragma: no branch +def slot(function: Callable[SlotParameters, List[PluginResult]], /) -> SlotProtocol[SlotParameters, List[PluginResult], PluginResult]: ... # pragma: no branch + +@overload +def slot(function: Callable[SlotParameters, Dict[str, PluginResult]], /) -> SlotProtocol[SlotParameters, Dict[str, PluginResult], PluginResult]: ... # pragma: no branch + +@overload +def slot(function: Callable[SlotParameters, None], /) -> SlotProtocol[SlotParameters, None, Any]: ... # pragma: no branch @overload -def slot(*, signature: Optional[str] = None, name: Optional[str] = None, max: Optional[int] = None, type_check: bool = True, entrypoint_group: str = 'pristan') -> Callable[[SlotFunction[SlotPapameters, PluginResult]], SlotFunction[SlotPapameters, PluginResult]]: ... # type: ignore[type-arg, unused-ignore] # pragma: no branch +def slot(*, signature: Optional[str] = None, name: Optional[str] = None, max: Optional[int] = None, type_check: bool = True, entrypoint_group: str = 'pristan') -> SlotDecoratorProtocol: ... # pragma: no branch, PLR0913, A002 @overload -def slot(function: str, *, signature: Optional[str] = None, name: Optional[str] = None, max: Optional[int] = None, type_check: bool = True, entrypoint_group: str = 'pristan') -> Callable[[SlotFunction[SlotPapameters, PluginResult]], SlotFunction[SlotPapameters, PluginResult]]: ... # type: ignore[type-arg, unused-ignore] # pragma: no branch +def slot(function: str, /, *, signature: Optional[str] = None, name: Optional[str] = None, max: Optional[int] = None, type_check: bool = True, entrypoint_group: str = 'pristan') -> SlotDecoratorProtocol: ... # pragma: no branch, PLR0913, A002 -def slot(function: Optional[Union[SlotFunction[SlotPapameters, PluginResult], str]] = None, /, *, signature: Optional[str] = None, name: Optional[str] = None, max: Optional[int] = None, type_check: bool = True, entrypoint_group: str = 'pristan') -> Union[SlotFunction[SlotPapameters, PluginResult], Callable[[SlotFunction[SlotPapameters, PluginResult]], SlotFunction[SlotPapameters, PluginResult]]]: # type: ignore[misc, type-arg, unused-ignore] # noqa: PLR0913, A002 +def slot(function: Optional[object] = None, /, *, signature: Optional[str] = None, name: Optional[str] = None, max: Optional[int] = None, type_check: bool = True, entrypoint_group: str = 'pristan') -> Any: # noqa: PLR0913, A002 if callable(function): - return wraps(function)(Slot(function, signature, name, max, type_check, entrypoint_group)) # type: ignore[arg-type, return-value, unused-ignore] + return wraps(function)(Slot(function, signature, name, max, type_check, entrypoint_group)) if isinstance(function, str): if name is not None and name != function: raise ValueError('You have specified two different names for the slot.') name = function - return partial(slot, signature=signature, name=name, max=max, type_check=type_check, entrypoint_group=entrypoint_group) # type: ignore[arg-type, unused-ignore] + return partial(slot, signature=signature, name=name, max=max, type_check=type_check, entrypoint_group=entrypoint_group) diff --git a/pyproject.toml b/pyproject.toml index 48db921..1492d41 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "pristan" -version = "0.0.12" +version = "0.0.13" authors = [{ name = "Evgeniy Blinov", email = "zheni-b@yandex.ru" }] description = "Function-based plugin system with respect to typing" readme = "README.md" @@ -17,6 +17,7 @@ dependencies = [ 'getsources>=0.0.4', 'packaging>=26.0', 'importlib-metadata; python_version < "3.10"', + 'typing_extensions; python_version < "3.10"', ] classifiers = [ "Operating System :: OS Independent", diff --git a/tests/typing/__init__.py b/tests/typing/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/typing/decorators/__init__.py b/tests/typing/decorators/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/typing/decorators/test_slot.py b/tests/typing/decorators/test_slot.py new file mode 100644 index 0000000..caeaec1 --- /dev/null +++ b/tests/typing/decorators/test_slot.py @@ -0,0 +1,528 @@ +# ruff: noqa: ARG001, B015, B018, F821, F841 +# mypy: warn-unused-ignores + +__test__ = False + +import sys +from typing import Any, Callable, Dict, List + +import pytest + +from pristan import slot +from pristan.common_types import ( + SlotDecoratorProtocol, + SlotProtocol, + SlotSelectionProtocol, +) + + +@pytest.mark.mypy_testing +def test_slot_returns_exact_dict_type_with_typing_dict_without_parentheses(): + @slot + def collect(value: int) -> Dict[str, int]: + return {} + + callable_view: Callable[[int], Dict[str, int]] = collect + slot_view: SlotProtocol[[int], Dict[str, int], int] = collect + + reveal_type(collect(1)) # R: builtins.dict[builtins.str, builtins.int] + reveal_type(collect.keys()) # R: builtins.tuple[builtins.str, ...] + + def consume(payload: Dict[str, int]): + pass + + consume(collect(1)) + callable_view(1) + slot_view(1) + + +@pytest.mark.mypy_testing +def test_slot_returns_exact_list_type_with_typing_list_with_parentheses(): + @slot() + def collect(value: int) -> List[int]: + return [] + + callable_view: Callable[[int], List[int]] = collect + slot_view: SlotProtocol[[int], List[int], int] = collect + + reveal_type(collect(1)) # R: builtins.list[builtins.int] + + def consume(payload: List[int]): + pass + + consume(collect(1)) + callable_view(1) + slot_view(1) + + +@pytest.mark.mypy_testing +def test_slot_returns_exact_list_and_dict_types_with_typing_collections_in_both_decorator_forms(): + @slot + def collect_list(value: int) -> List[int]: + return [] + + @slot() + def collect_dict(value: int) -> Dict[str, int]: + return {} + + list_view: SlotProtocol[[int], List[int], int] = collect_list + dict_view: SlotProtocol[[int], Dict[str, int], int] = collect_dict + + reveal_type(collect_list(1)) # R: builtins.list[builtins.int] + reveal_type(collect_dict(1)) # R: builtins.dict[builtins.str, builtins.int] + + list_view(1) + dict_view(1) + + +@pytest.mark.mypy_testing +def test_slot_without_return_annotation_returns_none_in_both_forms(): + @slot + def notify(value: int): + return None + + @slot() + def notify_too(value: int): + return None + + @notify.plugin + def plugin_without_parentheses(value: int) -> str: + return str(value) + + @notify_too.plugin() + def plugin_with_parentheses(value: int) -> int: + return value + + reveal_type(notify(1)) # R: Any + reveal_type(notify_too(1)) # R: Any + + def consume(payload: Any): + pass + + consume(notify(1)) + consume(notify_too(1)) + plugin_without_parentheses(1) + plugin_with_parentheses(1) + + +@pytest.mark.mypy_testing +def test_slot_configuration_arguments_are_typed(): + @slot('some_another_slot_name') + def slot_with_positional_name(value: int) -> List[int]: + return [] + + @slot(name='some_named_slot') + def slot_with_keyword_name(value: int) -> Dict[str, int]: + return {} + + @slot(signature='..', max=1, type_check=False, entrypoint_group='new_namespace') + def configured_slot(value: int) -> List[int]: + return [] + + reveal_type(slot_with_positional_name(1)) # R: builtins.list[builtins.int] + reveal_type(slot_with_keyword_name(1)) # R: builtins.dict[builtins.str, builtins.int] + reveal_type(configured_slot(1)) # R: builtins.list[builtins.int] + + +@pytest.mark.mypy_testing +def test_plugin_decorator_variants_preserve_callable_types(): + @slot + def collect(value: int) -> List[int]: + return [] + + @collect.plugin + def plugin_without_parentheses(value: int) -> int: + return value + + @collect.plugin() + def plugin_with_parentheses(value: int) -> int: + return value + 1 + + @collect.plugin('another_plugin_name') + def plugin_with_name(value: int) -> int: + return value + 2 + + @collect.plugin(unique=True) + def plugin_with_unique(value: int) -> int: + return value + 3 + + @collect.plugin(engine='>1.0.0') + def plugin_with_engine_string(value: int) -> int: + return value + 4 + + @collect.plugin(engine=['>1.0.0', '<2.0.0']) + def plugin_with_engine_list(value: int) -> int: + return value + 5 + + @collect.plugin(run_once=True) + def plugin_with_run_once(value: int) -> int: + return value + 6 + + callable_1: Callable[[int], int] = plugin_without_parentheses + callable_2: Callable[[int], int] = plugin_with_parentheses + callable_3: Callable[[int], int] = plugin_with_name + callable_4: Callable[[int], int] = plugin_with_unique + callable_5: Callable[[int], int] = plugin_with_engine_string + callable_6: Callable[[int], int] = plugin_with_engine_list + callable_7: Callable[[int], int] = plugin_with_run_once + + reveal_type(collect.keys()) # R: builtins.tuple[builtins.str, ...] + + callable_1(1) + callable_2(1) + callable_3(1) + callable_4(1) + callable_5(1) + callable_6(1) + callable_7(1) + + +@pytest.mark.mypy_testing +def test_slot_selection_has_narrower_public_type(): + @slot + def collect(value: int) -> List[int]: + return [] + + selection = collect['name'] + selection_view: SlotSelectionProtocol[[int], List[int], int] = selection + callable_view: Callable[[int], List[int]] = selection + + reveal_type(selection(1)) # R: builtins.list[builtins.int] + reveal_type(collect.keys()) # R: builtins.tuple[builtins.str, ...] + + length: int = len(selection) + slot_length: int = len(collect) + has_plugin_name: bool = 'name' in collect + + for plugin in collect: + name: str = plugin.name + + selection_view(1) + callable_view(1) + slot_length + has_plugin_name + + +@pytest.mark.mypy_testing +def test_non_existent_slot_selection_keeps_slot_call_contract(): + @slot + def collect(value: int) -> Dict[str, int]: + return {} + + selection = collect['non_existent_key'] + selection_view: SlotSelectionProtocol[[int], Dict[str, int], int] = selection + callable_view: Callable[[int], Dict[str, int]] = selection + + reveal_type(selection(1)) # R: builtins.dict[builtins.str, builtins.int] + + def consume(payload: Dict[str, int]): + pass + + consume(selection(1)) + selection_view(1) + callable_view(1) + + +@pytest.mark.mypy_testing +def test_iterated_plugins_preserve_slot_parameter_types(): + @slot + def collect(value: int) -> List[int]: + return [] + + @collect.plugin + def plugin(value: int) -> int: + return value + + for loaded_plugin in collect: + callable_view: Callable[[int], int] = loaded_plugin + reveal_type(loaded_plugin(1)) # R: builtins.int + loaded_plugin('wrong') # E: [arg-type] + + callable_view(1) + + +@pytest.mark.mypy_testing +def test_plugin_argument_validation_is_typed(): + @slot + def collect(value: int) -> List[int]: + return [] + + collect.plugin(engine=[1]) # E: [list-item] + collect.plugin('named', engine=[1]) # E: [list-item] + collect.plugin(engine=['>1.0.0', 1]) # E: [list-item] + + +@pytest.mark.mypy_testing +def test_slot_bad_factory_arguments_stay_type_errors(): + """Pin invalid slot(...) calls via code-specific ignores. + + These scenarios produce call-overload plus several overload notes on the + same physical line. pytest-mypy-testing cannot express that message bundle + precisely in a .py test file, so this test relies on warn-unused-ignores: + if any bad call ever becomes valid, mypy will report unused-ignore and the + test will fail. + """ + slot(1) # type: ignore[call-overload] + slot(name=1) # type: ignore[call-overload] + slot(signature=1) # type: ignore[call-overload] + slot(max='1') # type: ignore[call-overload] + slot(type_check='yes') # type: ignore[call-overload] + slot(entrypoint_group=None) # type: ignore[call-overload] + + +@pytest.mark.mypy_testing +def test_plugin_bad_factory_arguments_stay_type_errors(): + """Pin invalid slot.plugin(...) calls via code-specific ignores. + + The fragility is the same as for slot(...): mypy emits an overload error + together with several note lines on the same source line, and the plugin + cannot model all of them inline. Specific ignores plus warn-unused-ignores + let us assert that these calls must remain invalid. + """ + + @slot + def collect(value: int) -> List[int]: + return [] + + collect.plugin(1) # type: ignore[call-overload] + collect.plugin(unique='yes') # type: ignore[call-overload] + collect.plugin(engine=1) # type: ignore[call-overload] + collect.plugin(run_once='yes') # type: ignore[call-overload] + + +@pytest.mark.mypy_testing +def test_slot_factory_results_are_typed_as_slot_decorators(): + bare_factory: SlotDecoratorProtocol = slot() + named_factory: SlotDecoratorProtocol = slot('named_slot') + keyword_named_factory: SlotDecoratorProtocol = slot(name='other_named_slot') + + @bare_factory + def collect_with_bare_factory(value: int) -> List[int]: + return [] + + @named_factory + def collect_with_named_factory(value: int) -> Dict[str, int]: + return {} + + @keyword_named_factory + def collect_with_keyword_named_factory(value: int) -> List[int]: + return [] + + reveal_type(collect_with_bare_factory(1)) # R: builtins.list[builtins.int] + reveal_type(collect_with_named_factory(1)) # R: builtins.dict[builtins.str, builtins.int] + reveal_type(collect_with_keyword_named_factory(1)) # R: builtins.list[builtins.int] + + +@pytest.mark.mypy_testing +def test_slot_selection_is_not_assignable_to_full_slot_protocol(): + @slot + def collect(value: int) -> List[int]: + return [] + + full_slot: SlotProtocol[[int], List[int], int] = collect['name'] # E: [assignment] + + +@pytest.mark.mypy_testing +def test_exact_result_type_is_not_widened_for_typing_collections(): + @slot + def collect_list(value: int) -> List[int]: + return [] + + @slot + def collect_dict(value: int) -> Dict[str, int]: + return {} + + def consume_list(payload: List[int]): + pass + + def consume_dict(payload: Dict[str, int]): + pass + + consume_dict(collect_list(1)) # E: [arg-type] + consume_list(collect_dict(1)) # E: [arg-type] + consume_dict(collect_list['name'](1)) # E: [arg-type] + + +@pytest.mark.mypy_testing +def test_plugin_return_type_mismatch_is_reported_for_typing_collections(): + @slot + def collect_list(value: int) -> List[int]: + return [] + + @collect_list.plugin # E: [arg-type] + def bad_list_plugin(value: int) -> str: + return str(value) + + @slot + def collect_dict(value: int) -> Dict[str, int]: + return {} + + @collect_dict.plugin() # E: [arg-type] + def bad_dict_plugin(value: int) -> str: + return str(value) + + +@pytest.mark.mypy_testing +def test_selection_does_not_expose_full_slot_api(): + @slot + def collect(value: int) -> List[int]: + return [] + + selection = collect['name'] + + selection.plugin('name') # E: [attr-defined] + selection.keys() # E: [attr-defined] + selection['nested'] # E: [index] + 'name' in selection # E: [operator] + + +@pytest.mark.mypy_testing +def test_collection_api_reports_wrong_argument_types(): + @slot + def collect(value: int) -> List[int]: + return [] + + collect.keys(1) # E: [call-arg] + collect[1] # E: [index] + + keys_as_list: List[str] = collect.keys() # E: [assignment] + wrong_selection: int = collect['name'] # E: [assignment] + + +@pytest.mark.mypy_testing +def test_slot_with_loose_list_and_dict_annotations_keeps_any_payload_type(): + @slot + def collect_list() -> list: + return [] + + @slot + def collect_dict() -> dict: + return {} + + @collect_list.plugin + def list_plugin() -> str: + return 'value' + + @collect_dict.plugin + def dict_plugin() -> str: + return 'value' + + reveal_type(collect_list()) # R: builtins.list[Any] + reveal_type(collect_dict()) # R: builtins.dict[builtins.str, Any] + + +@pytest.mark.mypy_testing +def test_decorated_plugin_type_is_not_widened(): + @slot + def collect(value: int) -> List[int]: + return [] + + @collect.plugin + def plugin_without_parentheses(value: int) -> int: + return value + + @collect.plugin() + def plugin_with_parentheses(value: int) -> int: + return value + 1 + + wrong_1: Callable[[int], str] = plugin_without_parentheses # E: [assignment] + wrong_2: Callable[[int], str] = plugin_with_parentheses # E: [assignment] + + +@pytest.mark.skipif(sys.version_info < (3, 9), reason='built-in generics require Python 3.9+') +@pytest.mark.mypy_testing +def test_slot_returns_exact_dict_and_list_types_with_built_in_generics(): + @slot + def collect_dict(value: int) -> dict[str, int]: + return {} + + @slot() + def collect_list(value: int) -> list[int]: + return [] + + callable_dict: Callable[[int], dict[str, int]] = collect_dict + callable_list: Callable[[int], list[int]] = collect_list + slot_dict: SlotProtocol[[int], dict[str, int], int] = collect_dict + slot_list: SlotProtocol[[int], list[int], int] = collect_list + selection: SlotSelectionProtocol[[int], list[int], int] = collect_list['name'] + + reveal_type(collect_dict(1)) # R: builtins.dict[builtins.str, builtins.int] + reveal_type(collect_list(1)) # R: builtins.list[builtins.int] + reveal_type(collect_list['name'](1)) # R: builtins.list[builtins.int] + + def consume_dict(payload: dict[str, int]): + pass + + def consume_list(payload: list[int]): + pass + + consume_dict(collect_dict(1)) + consume_list(collect_list(1)) + callable_dict(1) + callable_list(1) + slot_dict(1) + slot_list(1) + selection(1) + + +@pytest.mark.skipif(sys.version_info < (3, 9), reason='built-in generics require Python 3.9+') +@pytest.mark.mypy_testing +def test_slot_returns_exact_list_and_dict_types_with_built_in_generics_in_both_decorator_forms(): + @slot + def collect_list(value: int) -> list[int]: + return [] + + @slot() + def collect_dict(value: int) -> dict[str, int]: + return {} + + list_view: SlotProtocol[[int], list[int], int] = collect_list + dict_view: SlotProtocol[[int], dict[str, int], int] = collect_dict + + reveal_type(collect_list(1)) # R: builtins.list[builtins.int] + reveal_type(collect_dict(1)) # R: builtins.dict[builtins.str, builtins.int] + + list_view(1) + dict_view(1) + + +@pytest.mark.skipif(sys.version_info < (3, 9), reason='built-in generics require Python 3.9+') +@pytest.mark.mypy_testing +def test_built_in_generic_results_are_not_widened(): + @slot + def collect_list(value: int) -> list[int]: + return [] + + @slot + def collect_dict(value: int) -> dict[str, int]: + return {} + + def consume_list(payload: list[int]): + pass + + def consume_dict(payload: dict[str, int]): + pass + + consume_dict(collect_list(1)) # E: [arg-type] + consume_list(collect_dict(1)) # E: [arg-type] + consume_dict(collect_list['name'](1)) # E: [arg-type] + + +@pytest.mark.skipif(sys.version_info < (3, 9), reason='built-in generics require Python 3.9+') +@pytest.mark.mypy_testing +def test_plugin_return_type_mismatch_is_reported_for_built_in_generics(): + @slot + def collect_list(value: int) -> list[int]: + return [] + + @collect_list.plugin # E: [arg-type] + def bad_list_plugin(value: int) -> str: + return str(value) + + @slot + def collect_dict(value: int) -> dict[str, int]: + return {} + + @collect_dict.plugin(run_once=True) # E: [arg-type] + def bad_dict_plugin(value: int) -> str: + return str(value)