# Coverage for server/parts/kedama.py : 96%

## 39 statements 38 run 1 missing 4 excluded 1 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 from __future__ import annotations

3

4# -- stdlib --

5 from typing import Optional, TYPE_CHECKING

6 import logging

7

8# -- third party --

9# -- own --

10 from server.endpoint import Client

11 from server.utils import command

12 from utils.events import EventHub

13 import wire

14

15# -- typing --

16 if TYPE_CHECKING:

17 from server.core import Core # noqa: F401

18

19

20# -- code --

21 log = logging.getLogger('server.parts.kedama')

22

23

24 class Kedama(object):

25 def __init__(self, core: Core):

26 self.core = core

27 D = core.events.client_command

28 D[wire.CreateRoom].subscribe(self._room_create_limit, -5)

29 D[wire.JoinRoom].subscribe(self._room_join_limit, -5)

30 D[wire.Invite].subscribe(self._invite_limit, -5)

31

32 def __repr__(self) -> str:

33 return self.__class__.__name__

34

35 # ----- Commands -----

36 @command('*')

37 def _room_create_limit(self, u: Client, ev: wire.CreateRoom) -> Optional[EventHub.StopPropagation]:

38 core = self.core

39 from thb import modes_kedama

40 if core.auth.is_kedama(u) and ev.mode not in modes_kedama:

41 u.write(wire.Error('kedama_limitation'))

42 return EventHub.STOP_PROPAGATION

43

44 return None

45

46 @command('*')

47 def _room_join_limit(self, u: Client, ev: wire.JoinRoom) -> Optional[EventHub.StopPropagation]:

48 core = self.core

49 g = core.room.get(ev.gid)

50 if not g: 50 ↛ 51line 50 didn't jump to line 51, because the condition on line 50 was never true

51 return None

52

53 from thb import modes_kedama

54 if core.auth.is_kedama(u) and g.__class__.__name__ not in modes_kedama:

55 u.write(wire.Error('kedama_limitation'))

56 return EventHub.STOP_PROPAGATION

57

58 return None

59

60 @command('*')

61 def _invite_limit(self, c: Client, ev: wire.Invite) -> Optional[EventHub.StopPropagation]:

62 core = self.core

63 uid = core.auth.uid_of(c)

64 if uid <= 0:

65 c.write(wire.Error('kedama_limitation'))

66 return EventHub.STOP_PROPAGATION

67

68 return None