1# -*- coding: utf-8 -*- 

2from __future__ import annotations 

3 

4# -- stdlib -- 

5from typing import Any, Callable, Dict, List, TYPE_CHECKING, Mapping 

6import logging 

7import random 

8import re 

9 

10# -- third party -- 

11import requests 

12 

13# -- own -- 

14# -- typing -- 

15if TYPE_CHECKING: 

16 from server.core import Core # noqa: F401 

17 

18 

19# -- code -- 

20log = logging.getLogger('Backend') 

21 

22 

23class BackendError(Exception): 

24 __slots__ = ('message', 'errors') 

25 

26 def __init__(self, errors: List[Dict[str, Any]]): 

27 self.errors = errors 

28 self.message = ', '.join( 

29 '%s: %s' % ('.'.join(e['path']), e['message']) 

30 for e in errors 

31 ) 

32 

33 def __repr__(self) -> str: 

34 return f'BackendError({repr(self.message)})' 

35 

36 

37class Backend(object): 

38 def __init__(self, core: Core): 

39 self.core = core 

40 self._client = requests.Session() 

41 

42 # ----- Public Method ----- 

43 def query_raw(self, ql: str, **vars: Dict[str, Any]) -> dict: 

44 cli = self._client 

45 core = self.core 

46 resp = cli.post(core.options.backend, json={'query': ql, 'variables': vars}) 

47 rst = resp.json() 

48 return rst 

49 

50 def query(self, ql: str, **vars: Any) -> Dict[str, Any]: 

51 rst = self.query_raw(ql, **vars) 

52 if 'errors' in rst: 

53 raise BackendError(rst['errors']) 

54 

55 return rst['data'] 

56 

57 

58class MockBackend(object): 

59 MOCKED: Dict[str, Callable] = {} 

60 NAMED = { 

61 'Proton': 2, 

62 'Alice': 3, 

63 'Bob': 4, 

64 'Cirno': 9, 

65 'Daiyousei': 7, 

66 'Reimu': 8, 

67 'Marisa': 11, 

68 'Youmu': 12, 

69 'Sakuya': 16, 

70 'Rumia': 10, 

71 'Utsuho': 6, 

72 } 

73 

74 def __init__(self, core: Core): 

75 self.core = core 

76 self.items: Dict[int, Dict[str, int]] = {} 

77 

78 def __repr__(self) -> str: 

79 return self.__class__.__name__ 

80 

81 def query(self, q: str, **vars: Mapping[str, Any]) -> Dict[str, Any]: 

82 q = self._strip(q) 

83 if q not in self.MOCKED: 83 ↛ 84line 83 didn't jump to line 84, because the condition on line 83 was never true

84 raise Exception("Can't mock query %s" % q) 

85 

86 return self.MOCKED[q](self, vars) 

87 

88 def _strip(self, q: str) -> str: 

89 q = q.strip() 

90 q = re.sub(r'[\r\n]', '', q) 

91 q = re.sub(r' +', ' ', q) 

92 return q 

93 

94 def _reg(f: Callable, strip: Any = _strip, MOCKED: Any = MOCKED) -> Callable: # type: ignore 

95 q = strip(None, f.__doc__) 

96 MOCKED[q] = f 

97 return f 

98 

99 @_reg 

100 def gameId(self, v) -> Any: 

101 ''' 

102 query { gameId } 

103 ''' 

104 return {'gameId': random.randint(0, 1000000)} 

105 

106 def uid_of(self, token): 

107 return self.NAMED.get(token) 

108 

109 @_reg 

110 def login(self, v, NAMED=NAMED) -> Any: 

111 ''' 

112 query($token: String) { 

113 player(token: $token) { 

114 id 

115 user { 

116 isActive 

117 userPermissions { 

118 codename 

119 } 

120 groups { 

121 permissions { 

122 codename 

123 } 

124 } 

125 } 

126 name 

127 } 

128 } 

129 ''' 

130 

131 return { 

132 'player': { 

133 'id': NAMED.get(v['token'], abs(hash(v['token'])) % 120943), 

134 'user': { 

135 'isActive': True, 

136 'userPermissions': [], 

137 'groups': { 

138 'permissions': [] 

139 } 

140 }, 

141 'name': v['token'], 

142 } 

143 } 

144 

145 @_reg 

146 def add_reward(self, v) -> Any: 

147 ''' 

148 mutation AddReward($gid: Int!, $rewards: [GameRewardInput!]!) { 

149 game { 

150 addReward(gameId: $gid, rewards: $rewards) { 

151 id 

152 } 

153 } 

154 } 

155 ''' 

156 return None 

157 

158 @_reg 

159 def archive(self, v) -> Any: 

160 ''' 

161 mutation ArchiveGame($game: GameInput!, $archive: String!) { 

162 game { 

163 archive(game: $meta, archive: $archive) { 

164 gid 

165 } 

166 } 

167 } 

168 ''' 

169 return {'game': {'archive': {'gid': 0}}} 

170 

171 @_reg 

172 def if_have_item(self, v) -> Any: 

173 ''' 

174 query($uid: Int!, $sku: String!) { 

175 player(id: $id) { 

176 haveItem(sku: $sku) 

177 } 

178 } 

179 ''' 

180 return {'player': {'haveItem': self.items[v['uid']][v['sku']] > 0}} 

181 

182 @_reg 

183 def remove_item(self, v) -> Any: 

184 ''' 

185 mutation($id: Int!, $sku: String, $r: String) { 

186 item { 

187 remove(player: $id, sku: $sku, reason: $r) 

188 } 

189 } 

190 ''' 

191 self.items[v['id']][v['sku']] -= 1 

192 return {'item': {'remove': True}}