# Coverage for server/utils.py : 100%

## 23 statements 23 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

21

22 def command(*states: Sequence[str]) -> Callable[[Callable[[Any, Client, T], Optional[EventHub.StopPropagation]]], Callable[[Any, Tuple[Client, T]], Union[Tuple[Client, T], EventHub.StopPropagation]]]:

23 def decorate(f: Any) -> Any:

24 @functools.wraps(f)

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

26 core = self.core

27 u, msg = ev

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

29 return ev

30 else:

31 ret = f(self, u, msg)

32 if ret is STOP:

33 return STOP

34 return ev

35

36 return wrapper

37

38 return decorate