# Coverage for client/parts/observe.py : 100%

## 35 statements 35 run 0 missing 2 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

6 import logging

7

8# -- third party --

9# -- own --

10 from utils.events import EventHub

11 import wire

12

13# -- typing --

14 if TYPE_CHECKING:

15 from client.core import Core # noqa: F401

16

17

18# -- code --

19 log = logging.getLogger('client.parts.Observe')

20 STOP = EventHub.STOP_PROPAGATION

21

22

23 class Observe(object):

24 def __init__(self, core: Core):

25 self.core = core

26

27 D = core.events.server_command

28 D[wire.ObserverEnter] += self._observer_enter

29 D[wire.ObserverLeave] += self._observer_leave

30 D[wire.ObserveRequest] += self._observe_request

31

32 # ---- Reactions -----

33 def _observer_enter(self, ev: wire.ObserverEnter) -> wire.ObserverEnter:

34 core = self.core

35 core.events.observer_enter.emit((ev.observer, ev.observee))

36 return ev

37

38 def _observer_leave(self, ev: wire.ObserverLeave) -> wire.ObserverLeave:

39 core = self.core

40 core.events.observer_leave.emit((ev.observer, ev.observee))

41 return ev

42

43 def _observe_request(self, ev: wire.ObserveRequest) -> wire.ObserveRequest:

44 core = self.core

45 core.events.observe_request.emit(ev.uid)

46 return ev

47

48 # ----- Public Method -----

49 def observe(self, uid: int):

50 core = self.core

51 core.server.write(wire.Observe(uid=uid))

52

53 def grant(self, uid: int, grant: bool):

54 core = self.core

55 core.server.write(wire.GrantObserve(uid=uid, grant=grant))

56

57 def kick(self, uid: int):

58 core = self.core

59 core.server.write(wire.KickObserver(uid=uid))