# Coverage for server/parts/backend.py : 72%

## 62 statements 48 run 14 missing 6 excluded 1 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, Dict, List, TYPE_CHECKING, Mapping

6 import logging

7 import random

8 import re

9

10# -- third party --

11 import requests

12

13# -- own --

14# -- typing --

15 if TYPE_CHECKING:

16 from server.core import Core # noqa: F401

17

18

19# -- code --

20 log = logging.getLogger('Backend')

21

22

23 class 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

37 class 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

58 class 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 ↛ 84 line 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) -> Any:

101 '''

102 query { gameId }

103 '''

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

105

106 def uid_of(self, token: str) -> int:

107 return self.NAMED.get(token, abs(hash(token)) % 120943)

108

109 @_reg

110 def login(self, v: Any) -> 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': self.uid_of(v['token']),

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