# Coverage for server/parts/item.py : 88%

## 70 statements 63 run 7 missing 4 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 collections import defaultdict

6 from typing import Dict, List, TYPE_CHECKING, Tuple

7 import logging

8

9# -- third party --

10 from mypy_extensions import TypedDict

11

12# -- own --

13 from game.base import GameItem

14 from server.base import Game

15 from server.endpoint import Client

16 from server.utils import command

17 from utils.misc import BusinessException

18 import wire

19

20# -- typing --

21 if TYPE_CHECKING:

22 from server.core import Core # noqa: F401

23

24

25# -- code --

26 log = logging.getLogger('server.parts.item')

27

28

29 class ItemAssocOnGame(TypedDict):

30 items: Dict[int, List[GameItem]]

31

32

33 def A(self: Item, g: Game) -> ItemAssocOnGame:

34 return g._[self]

35

36

37 class Item(object):

38 def __init__(self, core: Core):

39 self.core = core

40 core.events.game_created += self.handle_game_created

41 core.events.game_started += self.handle_game_started

42 _ = core.events.client_command

43 _[wire.UseItem] += self._use_item

44

45 def __repr__(self) -> str:

46 return self.__class__.__name__

47

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

49 core = self.core

50 final: Dict[int, List[GameItem]] = {}

51

52 for uid, l in A(self, g)['items'].items():

53 consumed = []

54 for i in l:

55 try:

56 rst = core.backend.query('''

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

58 item {

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

60 }

61 }

62 ''', id=uid, sku=i.sku, reason="Use in game %s" % core.room.gid_of(g))

63

64 if rst['item']['remove']: 64 ↛ 54line 64 didn't jump to line 54, because the condition on line 64 was never false

65 consumed.append(i)

66 except Exception:

67 log.exception('Error consuming item')

68

69 final[uid] = consumed

70

71 A(self, g)['items'] = final

72

73 return g

74

75 def handle_game_created(self, g: Game) -> Game:

76 assoc: ItemAssocOnGame = {

77 'items': defaultdict(list),

78 }

79 g._[self] = assoc

80 return g

81

82 def handle_game_left(self, ev: Tuple[Game, Client]) -> Tuple[Game, Client]:

83 g, u = ev

84 core = self.core

85 if not core.room.is_started(g) and not g.ended:

86 A(self, g)['items'].pop(core.auth.uid_of(u), None)

87

88 return ev

89

90 # ----- Command -----

91 @command('room')

92 def _use_item(self, u: Client, ev: wire.UseItem) -> None:

93 core = self.core

94 g = core.game.current(u)

95 assert g

96

97 try:

98 uid = core.auth.uid_of(u)

99 i = GameItem.from_sku(ev.sku)

100 i.should_usable(core, g, u)

101

102 have_item = core.backend.query('''

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

104 player(id: $id) {

105 haveItem(sku: $sku)

106 }

107 }

108 ''', uid=uid, sku=ev.sku)['player']['haveItem']

109

110 if not have_item:

111 from utils.misc import exceptions

112 raise exceptions.ItemNotFound

113

114 A(self, g)['items'][uid].append(i)

115 u.write(wire.Info('use_item_success'))

116 except BusinessException as e:

117 uid = core.auth.uid_of(u)

118 log.error('User %s failed to use item %s: %s', uid, ev.sku, e.name)

119 u.write(wire.Error(e.snake_case))

120

121 # ----- Methods ------

122 # ----- Public Methods -----

123 def items_of(self, g: Game) -> Dict[int, List[GameItem]]:

124 return A(self, g)['items']

125

126 def item_skus_of(self, g: Game) -> Dict[int, List[str]]:

127 return {k: [i.sku for i in v] for k, v in self.items_of(g).items()}