# Coverage for server/utils.py : 100%

## 26 statements 26 run 0 missing 0 excluded 0 partial

Hot-keys on this page

r m x p toggle line displays

j k next/prev highlighted chunk

0 (zero) top of page

1 (one) first highlighted chunk

1# -*- coding: utf-8 -*-

2

3# -- stdlib --

4 from typing import Any, Callable, Optional, Sequence, Tuple, TypeVar, Union

5 import functools

6 import logging

7

8# -- third party --

9# -- own --

10 from server.endpoint import Client

11 from utils.events import EventHub

12 from wire.msg import Message

13

14

15# -- code --

16 log = logging.getLogger("server.utils")

17 STOP = EventHub.STOP_PROPAGATION

18

19 T = TypeVar('T', bound=Message)

20 Self = TypeVar('Self')

21

22 CommandHandler = Callable[[Self, Client, T], Optional[EventHub.StopPropagation]]

23 WrappedCommandHandler = Callable[[Self, Tuple[Client, T]], Union[Tuple[Client, T], EventHub.StopPropagation]]

24

25

26 def command(*states: Sequence[str]) -> Callable[[CommandHandler], WrappedCommandHandler]:

27 def decorate(f: CommandHandler) -> WrappedCommandHandler:

28 @functools.wraps(f)

29 def wrapper(self: Self, ev: Tuple[Client, T]) -> Union[Tuple[Client, T], EventHub.StopPropagation]:

30 core = self.core # type: ignore

31 u, msg = ev

32 if '*' not in states and core.lobby.state_of(u) not in states:

33 return ev

34 else:

35 ret = f(self, u, msg)

36 if ret is STOP:

37 return STOP

38 return ev

39

40 return wrapper

41

42 return decorate