-
Notifications
You must be signed in to change notification settings - Fork 14
/
test_system_wscomms.py
95 lines (80 loc) · 2.71 KB
/
test_system_wscomms.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
import os
import sys
sys.path.append(os.path.join(os.path.dirname(__file__), ".."))
import setup_malcolm_paths
import time
# logging
# import logging
# logging.basicConfig(level=logging.DEBUG)
import unittest
# tornado
from tornado.websocket import websocket_connect
from tornado import gen
import json
# module imports
from malcolm.controllers.hellocontroller import HelloController
from malcolm.controllers.clientcontroller import ClientController
from malcolm.core.block import Block
from malcolm.core.process import Process
from malcolm.core.syncfactory import SyncFactory
from malcolm.wscomms.wsservercomms import WSServerComms
from malcolm.wscomms.wsclientcomms import WSClientComms
class TestSystemWSCommsServerOnly(unittest.TestCase):
def setUp(self):
self.sf = SyncFactory("sync")
self.process = Process("proc", self.sf)
block = Block("hello")
HelloController(self.process, block)
self.sc = WSServerComms("sc", self.process, 8888)
self.process.start()
self.sc.start()
def tearDown(self):
self.sc.stop()
self.sc.wait()
self.process.stop()
@gen.coroutine
def send_message(self):
conn = yield websocket_connect("ws://localhost:8888/ws")
req = dict(
type="Post",
id=0,
endpoint=["hello", "say_hello"],
parameters=dict(
name="me"
)
)
conn.write_message(json.dumps(req))
resp = yield conn.read_message()
resp = json.loads(resp)
self.assertEqual(resp, dict(
id=0,
type="Return",
value=dict(
greeting="Hello me"
)
))
conn.close()
def test_server_and_simple_client(self):
self.send_message()
class TestSystemWSCommsServerAndClient(TestSystemWSCommsServerOnly):
def setUp(self):
super(TestSystemWSCommsServerAndClient, self).setUp()
self.process2 = Process("proc2", self.sf)
self.block2 = Block("hello")
ClientController(self.process2, self.block2)
self.cc = WSClientComms("cc", self.process2, "ws://localhost:8888/ws")
self.process2.start()
self.cc.start()
def tearDown(self):
super(TestSystemWSCommsServerAndClient, self).tearDown()
self.cc.stop()
self.cc.wait()
self.process2.stop()
def test_server_with_malcolm_client(self):
# Normally we would wait for it to be connected here, but it isn't
# attached to a process so just sleep for a bit
time.sleep(0.1)
ret = self.block2.say_hello("me2")
self.assertEqual(ret, dict(greeting="Hello me2"))
if __name__ == "__main__":
unittest.main(verbosity=2)