Partially add tests for telegram event observer

This commit is contained in:
Alex Root Junior 2019-11-24 22:31:53 +02:00
parent 55f6c29ba6
commit 8204b6af52
3 changed files with 72 additions and 9 deletions

View file

@ -70,18 +70,18 @@ class TelegramEventObserver(EventObserver):
self.filters: List[Type[BaseFilter]] = []
def bind_filter(self, bound_filter: Type[BaseFilter]) -> None:
if not isinstance(bound_filter, BaseFilter):
pass
if not issubclass(bound_filter, BaseFilter):
raise TypeError(
"bound_filter() argument 'bound_filter' must be subclass of BaseFilter"
)
self.filters.append(bound_filter)
def _resolve_filters_chain(self):
registry: List[FilterType] = []
routers: List[Router] = []
router = self.router
while router and router not in routers:
while router:
observer = router.observers[self.event_name]
routers.append(router)
router = router.parent_router
for filter_ in observer.filters:

View file

@ -16,7 +16,7 @@ class BaseFilter(ABC, BaseModel):
@abstractmethod
async def __call__(
self, *args: Any, **kwargs: Any
) -> Callable[[Any], Awaitable[Union[bool, Dict[str, Any]]]]:
) -> Union[bool, Dict[str, Any]]:
pass
def __await__(self):