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

## 90 statements 85 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, Optional, Set, TYPE_CHECKING, Tuple

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) -> Optional[EventHub.StopPropagation]:

79 core = self.core

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

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

82 return None

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 None

101

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

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

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 ↛ 109line 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 assert g

113

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

115

116 other.write(wire.InviteRequest(

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

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

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

120 type=g.__class__.__name__,

121 ))

122

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

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

125 core = self.core

126 ouid = ev.uid

127 other = core.lobby.get(ouid)

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

129 return

130

131 g = core.game.current(u)

132 g2 = core.game.current(other)

133 assert g

134 assert g2

135

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

137 return

138

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

140 return

141

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

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

144

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

146 u.write(wire.KickRequest(

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

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

149 votes=len(bl),

150 ))

151

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

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

154 core.room.exit_game(other)

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))