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

## 120 statements 113 run 7 missing 4 excluded 8 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 List, Optional, Sequence, TYPE_CHECKING, Tuple

6 import logging

7

8# -- third party --

9 from mypy_extensions import TypedDict

10

11# -- own --

12 from server.base import Game, HumanPlayer

13 from server.endpoint import Client

14 from server.utils import command

15 from utils.events import EventHub

16 import wire

17

18# -- typing --

19 if TYPE_CHECKING:

20 from server.core import Core # noqa: F401

21

22

23# -- code --

24 log = logging.getLogger('Contest')

25

26

27 class ContestAssocOnGame(TypedDict):

28 uids: List[int]

29

30

31 def A(self: Contest, g: Game) -> ContestAssocOnGame:

32 return g._[self]

33

34

35 class Contest(object):

36 def __init__(self, core: Core):

37 self.core = core

38

39 core.events.user_state_transition += self.handle_user_state_transition

40 core.events.game_started += self.handle_game_started

41 core.events.game_aborted += self.handle_game_aborted

42 core.events.game_ended += self.handle_game_ended

43 core.events.game_successive_create += self.handle_game_successive_create

44

45 D = core.events.client_command

46 D[wire.SetupContest] += self._contest

47 D[wire.JoinRoom].subscribe(self._room_join_contest_limit, -3)

48

49 def __repr__(self) -> str:

50 return self.__class__.__name__

51

52 def handle_user_state_transition(self, ev: Tuple[Client, str, str]) -> Tuple[Client, str, str]:

53 c, f, t = ev

54 return ev

55

56 def handle_game_started(self, g: Game) -> Game:

57 core = self.core

58

59 name = core.room.name_of(g)

60 flags = core.room.flags_of(g)

61 users = core.room.users_of(g)

62

63 if flags.get('contest'):

64 core.connect.speaker(

65 '文文', '“%s”开始了！参与玩家：%s' % (

66 name, '，'.join([core.auth.name_of(u) for u in users])

67 )

68 )

69

70 return g

71

72 def handle_game_aborted(self, g: Game) -> Game:

73 core = self.core

74 if core.room.is_started(g) and core.room.flags_of(g).get('contest'): 74 ↛ 75line 74 didn't jump to line 75, because the condition on line 74 was never true

75 core.connect.speaker(

76 '文文', '“%s”意外终止了，比赛结果作废！' % core.room.name_of(g)

77 )

78

79 return g

80

81 def handle_game_ended(self, g: Game) -> Game:

82 core = self.core

83

84 if not core.room.flags_of(g).get('contest'):

85 return g

86

87 if core.game.is_aborted(g): 87 ↛ 88line 87 didn't jump to line 88, because the condition on line 87 was never true

88 return g

89

90 winners: List[Client] = [

91 u.client for u in core.game.winners_of(g)

92 if isinstance(u, HumanPlayer)

93 ]

94

95 core.connect.speaker(

96 '文文',

97 '“%s”结束了！获胜玩家：%s' % (

98 core.room.name_of(g),

99 '，'.join([core.auth.name_of(u) for u in winners])

100 )

101 )

102

103 return g

104

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

106 core = self.core

107 old, g = ev

108 flags = core.room.flags_of(g)

109 if flags.get('contest'):

110 fields = old._[self]

111 g._[self] = fields

112 self._start_poll(g, fields['uids'])

113 return ev

114

115 # ----- Client Commands -----

116 @command('*')

117 def _contest(self, c: Client, ev: wire.SetupContest) -> None:

118 core = self.core

119 from thb import modes

120 gamecls = modes[ev.mode]

121 if len(ev.uids) != gamecls.n_persons:

122 c.write(wire.Error(msg='wrong_players_count'))

123 return

124

125 g = core.room.create_game(gamecls, ev.name, {'contest': True})

126

127 assoc: ContestAssocOnGame = {

128 'uids': ev.uids,

129 }

130 g._[self] = assoc

131 self._start_poll(g, ev.uids)

132

133 @command('*')

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

135 core = self.core

136

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

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

139 return None

140

141 flags = core.room.flags_of(g)

142 uid = core.auth.uid_of(u)

143

144 if flags.get('contest'):

145 uid = core.auth.uid_of(u)

146 if uid not in A(self, g)['uids']: 146 ↛ 150line 146 didn't jump to line 150, because the condition on line 146 was never false

147 u.write(wire.Error('not_competitor'))

148 return EventHub.STOP_PROPAGATION

149

150 return None

151

152 # ----- Methods -----

153 def _start_poll(self, g: Game, uids: Sequence[int]) -> None:

154 core = self.core

155 gid = core.room.gid_of(g)

156 name = core.room.name_of(g)

157

158 core.runner.spawn(core.connect.speaker, '文文', f'“{name}”房间已经建立，请相关玩家就位！')

159

160 @core.runner.spawn

161 def pull() -> None:

162 while core.room.get(gid) is g: 162 ↛ 198line 162 didn't jump to line 198, because the condition on line 162 was never false

163 users = core.room.online_users_of(g)

164 uids = {core.auth.uid_of(u) for u in users}

165 contest_uids = set(A(self, g)['uids'])

166 pending = contest_uids - uids

167 if not pending:

168 break

169

170 log.debug("Contest: Ready to pull %s", pending)

171

172 for uid in pending:

173 u = core.lobby.get(uid)

174 if not u:

175 log.debug("Contest: %s not found", uid)

176 continue

177

178 if core.lobby.state_of(u) == 'lobby':

179 log.debug("Contest: Pulling %s", uid)

180 core.room.join_game(g, u)

181 elif core.lobby.state_of(u) in ('ready', 'room'): 181 ↛ 182line 181 didn't jump to line 182, because the condition on line 181 was never true

182 log.debug("Contest: Pulling %s from state [%s]", uid, core.lobby.state_of(u).state)

183 core.room.exit_game(u)

184 core.room.join_game(g, u)

185 elif core.lobby.state_of(u) == 'ob':

186 log.debug("Contest: Pulling %s from state [%s]", uid, core.lobby.state_of(u).state)

187 core.observe.observe_detach(u)

188 core.room.join_game(g, u)

189 elif core.lobby.state_of(u) == 'game': 189 ↛ 172line 189 didn't jump to line 172, because the condition on line 189 was never false

190 log.debug("Contest: %s in game, sending warning", uid)

191 core.runner.spawn(u.write, wire.SystemMsg('你有比赛房间，请尽快结束游戏参与比赛'))

192

193 if core.options.testing: 193 ↛ 196line 193 didn't jump to line 196, because the condition on line 193 was never false

194 core.runner.sleep(0.01)

195 else:

196 core.runner.sleep(30)

197

198 log.debug("Contest: Finished pulling for game %s", gid)