Fixed coverage

This commit is contained in:
Alex Root Junior 2023-07-02 14:23:10 +03:00
parent 9da899116d
commit 4c46596848
No known key found for this signature in database
GPG key ID: 074C1D455EBEA4AC
3 changed files with 13 additions and 20 deletions

View file

@ -10,7 +10,7 @@ if TYPE_CHECKING:
from ...bot import Bot
class NextRequestMiddlewareType(Protocol[TelegramType]):
class NextRequestMiddlewareType(Protocol[TelegramType]): # pragma: no cover
async def __call__(
self,
bot: "Bot",
@ -19,7 +19,7 @@ class NextRequestMiddlewareType(Protocol[TelegramType]):
pass
class RequestMiddlewareProtocol(Protocol):
class RequestMiddlewareType(Protocol): # pragma: no cover
async def __call__(
self,
make_request: NextRequestMiddlewareType[TelegramType],

View file

@ -5,47 +5,44 @@ from typing import Any, Callable, List, Optional, Sequence, Union, cast, overloa
from aiogram.client.session.middlewares.base import (
NextRequestMiddlewareType,
RequestMiddlewareProtocol,
RequestMiddlewareType,
)
from aiogram.methods.base import TelegramType
class RequestMiddlewareManager(Sequence[RequestMiddlewareProtocol]):
class RequestMiddlewareManager(Sequence[RequestMiddlewareType]):
def __init__(self) -> None:
self._middlewares: List[RequestMiddlewareProtocol] = []
self._middlewares: List[RequestMiddlewareType] = []
def register(
self,
middleware: RequestMiddlewareProtocol,
) -> RequestMiddlewareProtocol:
middleware: RequestMiddlewareType,
) -> RequestMiddlewareType:
self._middlewares.append(middleware)
return middleware
def unregister(self, middleware: RequestMiddlewareProtocol) -> None:
def unregister(self, middleware: RequestMiddlewareType) -> None:
self._middlewares.remove(middleware)
def __call__(
self,
middleware: Optional[RequestMiddlewareProtocol] = None,
) -> Union[
Callable[[RequestMiddlewareProtocol], RequestMiddlewareProtocol],
RequestMiddlewareProtocol,
]:
middleware: Optional[RequestMiddlewareType] = None,
) -> Union[Callable[[RequestMiddlewareType], RequestMiddlewareType], RequestMiddlewareType,]:
if middleware is None:
return self.register
return self.register(middleware)
@overload
def __getitem__(self, item: int) -> RequestMiddlewareProtocol:
def __getitem__(self, item: int) -> RequestMiddlewareType:
pass
@overload
def __getitem__(self, item: slice) -> Sequence[RequestMiddlewareProtocol]:
def __getitem__(self, item: slice) -> Sequence[RequestMiddlewareType]:
pass
def __getitem__(
self, item: Union[int, slice]
) -> Union[RequestMiddlewareProtocol, Sequence[RequestMiddlewareProtocol]]:
) -> Union[RequestMiddlewareType, Sequence[RequestMiddlewareType]]:
return self._middlewares[item]
def __len__(self) -> int:

View file

@ -29,10 +29,6 @@ class InputFile(ABC):
self.filename = filename
self.chunk_size = chunk_size
@classmethod
def __get_validators__(cls) -> Iterator[None]:
yield None
@abstractmethod
async def read(self, chunk_size: int) -> AsyncGenerator[bytes, None]: # pragma: no cover
yield b""