-
Notifications
You must be signed in to change notification settings - Fork 1
/
test_zmq_system.py
94 lines (77 loc) · 2.57 KB
/
test_zmq_system.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
#!/bin/env dls-python
from pkg_resources import require
import subprocess
require("mock")
require("pyzmq")
require("cothread")
import unittest
import os
import sys
import logging
#logging.basicConfig()
#logging.basicConfig(level=logging.DEBUG)
# format="%(asctime)s;%(levelname)s;%(message)s")
# Module import
sys.path.append(os.path.join(os.path.dirname(__file__), "..", ".."))
from malcolm.core.process import Process
from malcolm.core.directoryservice import DirectoryService
from malcolm.core.loop import TimerLoop
from malcolm.core import Device, wrap_method
class Counter(Device):
def __init__(self, name):
super(Counter, self).__init__(name)
self.counter = 0
self.add_loop(TimerLoop("timer", self.do_count, 0.01))
def do_count(self):
self.counter += 1
@wrap_method()
def getCount(self):
return self.counter
@wrap_method()
def hello(self):
self.cothread.Sleep(0.1)
return "world"
@wrap_method()
def longHello(self):
self.cothread.Sleep(0.5)
return "long world"
def exit(self):
pass
class ZmqSystemTest(unittest.TestCase):
# class ZmqSystemTest(object):
def setUp(self):
# Spawn ds under cothread
self.ds = DirectoryService(["zmq://ipc:///tmp/sock.ipc"])
self.ds.run(block=False)
self.lp = Process(
[], "Local Process", ds_string="zmq://ipc:///tmp/sock.ipc")
self.lp.run(block=False)
self.lp.ds.createCounter(name="The Counter")
self.c = self.lp.get_device("The Counter")
def test_simple_function(self):
import cothread
cothread.Sleep(0.2)
start = self.c.getCount()
self.assertEqual(self.c.hello(), "world")
# Hello world takes about 10 ticks
self.assertAlmostEqual(self.c.getCount(), start + 10, delta=3)
# Do a long running call
s = cothread.Spawn(self.c.longHello)
# Check it returns immediately
self.assertAlmostEqual(self.c.getCount(), start + 10, delta=3)
self.assertEqual(self.c.hello(), "world")
# Hello world takes 10 ticks
self.assertAlmostEqual(self.c.getCount(), start + 20, delta=3)
self.assertEqual(s.Wait(), "long world")
# Long hello takes about 50 ticks from send
self.assertAlmostEqual(self.c.getCount(), start + 60, delta=8)
s.Wait()
def tearDown(self):
self.c = None
self.lp.ds.exit()
self.lp.exit()
self.lp = None
self.ds.loop_wait()
self.ds = None
if __name__ == '__main__':
unittest.main(verbosity=2)