# Coverage for server/parts/invite.py : 91%

## 89 statements 84 run 5 missing 4 excluded 5 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 collections import defaultdict

6 from typing import Dict, Set, TYPE_CHECKING, Tuple, Union

7 import logging

8

9# -- third party --

10 from mypy_extensions import TypedDict

11

12# -- own --

13 from server.base import Game

14 from server.endpoint import Client

15 from server.utils import command

16 from utils.events import EventHub

17 import wire

18

19# -- typing --

20 if TYPE_CHECKING:

21 from server.core import Core # noqa: F401

22

23

24# -- code --

25 log = logging.getLogger('Invite')

26

27

28 class InviteAssocOnGame(TypedDict):

29 invited: Set[int]

30 banned: Dict[int, Set[int]]

31

32

33 def A(self: Invite, g: Game) -> InviteAssocOnGame:

34 return g._[self]

35

36

37 class Invite(object):

38 def __init__(self, core: Core):

39 self.core = core

40

41 core.events.game_created += self.handle_game_created

42 core.events.game_left += self.handle_game_left

43 core.events.game_successive_create += self.handle_game_successive_create

44

45 D = core.events.client_command

46 D[wire.Invite] += self._invite

47 D[wire.Kick] += self._kick

48

49 D[wire.JoinRoom].subscribe(self._room_join_invite_limit, -3)

50

51 def __repr__(self) -> str:

52 return self.__class__.__name__

53

54 def handle_game_created(self, g: Game) -> Game:

55 assoc: InviteAssocOnGame = {

56 'invited': set(),

57 'banned': defaultdict(set),

58 }

59 g._[self] = assoc

60 return g

61

62 def handle_game_successive_create(self, ev: Tuple[Game, Game]) -> Tuple[Game, Game]:

63 old, g = ev

64 g._[self] = old._[self]

65 return ev

66

67 def handle_game_left(self, ev: Tuple[Game, Client]) -> Tuple[Game, Client]:

68 g, c = ev

69 core = self.core

70

71 for bl in A(self, g)['banned'].values():

72 bl.discard(core.auth.uid_of(c))

73

74 return ev

75

76 # ----- Commands -----

77 @command('*')

78 def _room_join_invite_limit(self, u: Client, ev: wire.JoinRoom) -> Union[wire.JoinRoom, EventHub.StopPropagation]:

79 core = self.core

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

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

82 return ev

83

84 flags = core.room.flags_of(g)

85 uid = core.auth.uid_of(u)

86

87 banned = A(self, g)['banned']

88 invited = A(self, g)['invited']

89

90 # banned

91 if len(banned[uid]) >= max(g.n_persons // 2, 1):

92 u.write(wire.Error('banned'))

93 return EventHub.STOP_PROPAGATION

94

95 # invite

96 if flags.get('invite') and uid not in invited:

97 u.write(wire.Error('not_invited'))

98 return EventHub.STOP_PROPAGATION

99

100 return ev

101

102 @command('room', 'ready')

103 def _invite(self, u: Client, ev: wire.Invite) -> wire.Invite:

104 core = self.core

105 ouid = ev.uid

106

107 other = core.lobby.get(ouid)

108 if not (other and core.lobby.state_of(other) in ('lobby', 'ob')): 108 ↛ 109 line 108 didn't jump to line 109, because the condition on line 108 was never true

109 return

110

111 g = core.game.current(u)

112

113 A(self, g)['invited'].add(ouid)

114

115 other.write(wire.InviteRequest(

116 uid=core.auth.uid_of(u),

117 name=core.auth.name_of(u),

118 gid=core.room.gid_of(g),

119 type=g.__class__.__name__,

120 ))

121

122 return ev

123

124 @command('room', 'ready')

125 def _kick(self, u: Client, ev: wire.Kick) -> wire.Kick:

126 core = self.core

127 ouid = ev.uid

128 other = core.lobby.get(ouid)

129 if not other: 129 ↛ 130 line 129 didn't jump to line 130, because the condition on line 129 was never true

130 return

131

132 g = core.game.current(u)

133 g2 = core.game.current(other)

134 if g is not g2: 134 ↛ 135 line 134 didn't jump to line 135, because the condition on line 134 was never true

135 return

136

137 if core.lobby.state_of(other) not in ('room', 'ready'): 137 ↛ 138 line 137 didn't jump to line 138, because the condition on line 137 was never true

138 return

139

140 bl = A(self, g)['banned'][ouid]

141 bl.add(core.auth.uid_of(u))

142

143 for u in core.room.online_users_of(g):

144 u.write(wire.KickRequest(

145 uid=core.auth.uid_of(u),

146 victim=core.auth.uid_of(other),

147 votes=len(bl),

148 ))

149

150 if len(bl) >= len(core.room.users_of(g)) // 2:

151 A(self, g)['invited'].discard(ouid)

152 core.room.exit_game(other)

153

154 return ev

155

156 # ----- Methods -----

157 def add_invited(self, g: Game, u: Client) -> None:

158 core = self.core

159 A(self, g)['invited'].add(core.auth.uid_of(u))