# Coverage for tests/e2e/mock.py : 94%

## 71 statements 66 run 5 missing 0 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 import logging

6 import random

7

8# -- third party --

9 from gevent.pool import Pool

10 import gevent

11

12# -- own --

13 from core import CoreRunner

14 from server.parts.backend import MockBackend

15 from server.parts.connect import MockConnect

16 import client.core

17 import server.core

18 from utils.events import EventHub

19

20

21# -- code --

22 log = logging.getLogger('mock')

23

24

25 def enable_endpoint_logging():

26 import endpoint

27 endpoint.log.setLevel(logging.DEBUG)

28

29

30 class EventTap(object):

31

32 def __init__(self):

33 self._taps = {}

34

35 def tap(self, *cores):

36 for core in cores:

37 for k in dir(core.events):

38 if k.startswith('__'):

39 continue

40

41 hub = getattr(core.events, k)

42 if not isinstance(hub, EventHub):

43 continue

44

45 def tapper(ev, hub=hub):

46 self._taps[hub] = ev

47 return ev

48

49 hub += tapper

50

51 def take(self, hub):

52 v = self._taps[hub]

53 del self._taps[hub]

54 return v

55

56 def clear(self):

57 self._taps.clear()

58

59 def __getitem__(self, k):

60 return self._taps[k]

61

62 def __contains__(self, k):

63 return k in self._taps

64

65

66 class Environ(object):

67 def __init__(self):

68 self.pool = Pool(32)

69 p = random.randint(10000, 30000)

70 ip = random.randint(1, 16777215)

71 self.rendezvous = f'tcp://127.{ip}:{p}'

72 self.parent = gevent.getcurrent()

73

74 def _run(self, runner):

75 try:

76 g = gevent.getcurrent()

77 g.gr_name = f'Environ.init({runner})'

78 runner.run()

79 except Exception as e:

80 gevent.kill(self.parent, e)

81

82 def client_core(self) -> client.core.Core:

83 core = client.core.Core(disables=['warpgate'], testing=True)

84 runner = CoreRunner(core, paranoid=True)

85 self.pool.spawn(self._run, runner)

86 runner.ready.wait()

87 core.server.connect(self.rendezvous)

88 return core

89

90 def server_core(self) -> server.core.Core:

91 core = server.core.Core(disables=[

92 'connect', 'stats', 'backend'

93 ], listen=self.rendezvous, testing=True)

94 core.backend = MockBackend(core)

95 core.connect = MockConnect(core)

96 runner = CoreRunner(core, paranoid=True)

97 self.pool.spawn(self._run, runner)

98 gevent.sleep(0.05)

99 return core

100

101 def shutdown(self):

102 self.pool.kill()