# Coverage for server/parts/admin.py : 67%

## 94 statements 67 run 27 missing 4 excluded 3 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 Any, Callable, List, TYPE_CHECKING, Tuple, TypeVar

6 import logging

7

8# -- third party --

9# -- own --

10 from server.endpoint import Client

11 from utils.events import EventHub

12 from wire import msg

13

14# -- typing --

15 if TYPE_CHECKING:

16 from server.core import Core # noqa: F401

17

18

19# -- code --

20 log = logging.getLogger('Admin')

21 T = TypeVar('T', bound=msg.Message)

22 STOP = EventHub.STOP_PROPAGATION

23

24

25 def _need_admin(f: Callable[[Admin, Client, T], Any]) -> Callable[[Admin, Tuple[Client, T]], EventHub.StopPropagation]:

26 def wrapper(self: Admin, ev: Tuple[Client, T]) -> EventHub.StopPropagation:

27 core = self.core

28 u, m = ev

29 if core.auth.uid_of(u) not in self.admins:

30 return STOP

31

32 f(self, u, m)

33 u.write(msg.SystemMsg(msg='成功的执行了管理命令'))

34 return STOP

35

36 return wrapper

37

38

39 class Admin(object):

40 def __init__(self, core: Core):

41 self.core = core

42

43 D = core.events.client_command

44 D[msg.AdminStacktrace] += self._stacktrace

45 D[msg.AdminClearZombies] += self._clearzombies

46 D[msg.AdminMigrate] += self._migrate

47 D[msg.AdminKick] += self._kick

48 D[msg.AdminKillGame] += self._kill_game

49 D[msg.AdminAdd] += self._add

50 D[msg.AdminRemove] += self._remove

51 D[msg.AdminAddBigbrother] += self._add_bigbrother

52 D[msg.AdminRemoveBigbrother] += self._remove_bigbrother

53

54 self.admins: List[int] = [2, 109, 351, 3044, 6573, 6584, 9783]

55

56 def __repr__(self) -> str:

57 return self.__class__.__name__

58

59 @_need_admin

60 def _kick(self, c: Client, m: msg.AdminKick) -> None:

61 core = self.core

62 u = core.lobby.get(m.uid)

63 if u: u.terminate()

64

65 @_need_admin

66 def _clearzombies(self, c: Client, m: msg.AdminClearZombies) -> None:

67 core = self.core

68 users = core.lobby.all_users()

69 for u in users:

70 if u.is_dead(): 70 ↛ 71 line 70 didn't jump to line 71, because the condition on line 70 was never true

71 log.info('Clear zombie: %r', u)

72 core.events.client_dropped.emit(u)

73

74 @_need_admin

75 def _migrate(self, c: Client, m: msg.AdminMigrate) -> None:

76 core = self.core

77

78 @core.runner.spawn

79 def sysmsg() -> None:

80 while True:

81 users = core.lobby.all_users()

82 users.write(msg.SystemMsg(msg='游戏已经更新，当前的游戏结束后将会被自动踢出，请更新后重新游戏'))

83 core.runner.sleep(300)

84

85 @core.runner.spawn

86 def kick() -> None:

87 core.runner.sleep(30)

88 while True:

89 users = core.lobby.all_users()

90 for u in users:

91 if core.lobby.state_of(u) in ('lobby', 'room', 'ready', 'connected'):

92 u.terminate()

93

94 core.runner.sleep(1)

95

96 @_need_admin

97 def _stacktrace(self, c: Client, m: msg.AdminStacktrace) -> None:

98 core = self.core

99 g = core.game.current(c)

100 if not g: 100 ↛ 103 line 100 didn't jump to line 103, because the condition on line 100 was never false

101 return

102

103 log.info('>>>>> GAME STACKTRACE <<<<<')

104

105 def logtraceback(gr: Any) -> None:

106 import traceback

107 log.info('----- %r -----\n%s', gr, ''.join(traceback.format_stack(gr)))

108

109 logtraceback(g)

110

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

112 gr = u.get_greenlet()

113 if gr and gr.gr_frame:

114 logtraceback(gr.gr_frame)

115

116 log.info('===========================')

117

118 @_need_admin

119 def _kill_game(self, c: Client, m: msg.AdminKillGame) -> None:

120 core = self.core

121 g = core.room.get(m.gid)

122 if not g: return 122 ↛ exit line 122 didn't return from function '_kill_game', because the return on line 122 wasn't executed

123

124 users = core.room.online_users_of(g)

125 for u in users:

126 core.room.exit_game(u)

127

128 @_need_admin

129 def _add(self, c: Client, m: msg.AdminAdd) -> None:

130 self.admins.append(m.uid)

131

132 @_need_admin

133 def _remove(self, c: Client, m: msg.AdminRemove) -> None:

134 try:

135 self.admins.remove(m.uid)

136 except Exception:

137 pass

138

139 @_need_admin

140 def _add_bigbrother(self, c: Client, m: msg.AdminAddBigbrother) -> None:

141 core = self.core

142 core.observe.add_bigbrother(m.uid)

143

144 @_need_admin

145 def _remove_bigbrother(self, c: Client, m: msg.AdminRemoveBigbrother) -> None:

146 core = self.core

147 core.observe.remove_bigbrother(m.uid)