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

## 46 statements 33 run 13 missing 4 excluded 0 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 TYPE_CHECKING, Any, List

6 import json

7 import logging

8

9# -- third party --

10 import websocket

11

12# -- own --

13 from utils.misc import throttle

14

15# -- typing --

16 if TYPE_CHECKING:

17 from server.core import Core # noqa: F401

18

19

20# -- code --

21 log = logging.getLogger('Connect')

22

23

24 class Connect(object):

25 def __init__(self, core: Core):

26 self.core = core

27

28 self._wsconn = None

29

30 core.events.game_created += self.refresh_status

31 core.events.game_started += self.refresh_status

32 core.events.game_ended += self.refresh_status

33 core.events.game_aborted += self.refresh_status

34 core.events.user_state_transition += self.refresh_status

35

36 def __repr__(self) -> str:

37 return self.__class__.__name__

38

39 def refresh_status(self, ev: Any) -> Any:

40 self._refresh_status()

41 return ev

42

43 # ----- Public Methods -----

44 def speaker(self, name: str, text: str) -> None:

45 core = self.core

46 self._wssend({

47 'op': 'Message',

48 'arg': {

49 'entity': 'Speaker',

50 'channel': core.options.node,

51 'text': text,

52 }

53 })

54

55 # ----- Methods -----

56 @throttle(1.5)

57 def _refresh_status(self) -> None:

58 core = self.core

59 self._wssend({

60 'op': 'Message',

61 'arg': {

62 'entity': 'Interconnect',

63 'channel': 'users',

64 'text': json.dumps([

65 core.view.User(u)

66 for u in core.lobby.all_users()

67 ]),

68 }

69 })

70

71 def _wssend(self, v: Any) -> None:

72 core = self.core

73 for i in range(3):

74 try:

75 conn = self._wsconn

76 if not conn or not conn.connected:

77 conn = websocket.create_connection(core.options.interconnect)

78 self._wsconn = conn

79 conn.send(json.dumps(v))

80 return

81 except Exception:

82 log.exception('Error sending interconnect message')

83 core.runner.sleep(3)

84

85 log.error('WebSocket send with multiple failed attempts, giving up, message: %s', v)

86

87

88 class MockConnect(Connect):

89 def __init__(self, core: Core):

90 super().__init__(core)

91

92 self.mock_sent_messages: List[Any] = []

93

94 def _wssend(self, v: Any) -> None:

95 self.mock_sent_messages.append(v)

96 log.info("MockClient: Send %s", v)