srenblad
/
artiq
forked from M-Labs/artiq
1
0
Fork 0
Code Pull Requests Activity

sync_struct test: don't poll, use Event instead

This commit is contained in:
Yann Sionneau 2015-06-03 15:40:58 +02:00
parent 21d88d8345
commit b8bdce5bd1
1 changed files with 9 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
artiq/test/sync_struct.py
View File

@ -29,11 +29,11 @@ class SyncStructCase(unittest.TestCase):
self.test_dict = init self.test_dict = init
return init return init
@asyncio.coroutine def notify(self, mod):
def do_recv(self): print("mod: {}".format(mod))
while not hasattr(self, "test_dict")\ if (mod["action"] == "init" and "Finished" in mod["struct"])\
or "Finished" not in self.test_dict.keys(): or (mod["action"] == "setitem" and mod["key"] == "Finished"):
yield from asyncio.sleep(0.5) self.receiving_done.set()
def setUp(self): def setUp(self):
self.loop = asyncio.new_event_loop() self.loop = asyncio.new_event_loop()
@ -41,6 +41,7 @@ class SyncStructCase(unittest.TestCase):
def test_recv(self): def test_recv(self):
loop = self.loop loop = self.loop
self.receiving_done = asyncio.Event()
publisher = asyncio.Future() publisher = asyncio.Future()
test_dict = asyncio.Future() test_dict = asyncio.Future()
asyncio.async(start_server(publisher, test_dict)) asyncio.async(start_server(publisher, test_dict))
@ -53,10 +54,11 @@ class SyncStructCase(unittest.TestCase):
loop.run_until_complete(write_test_data(test_vector)) loop.run_until_complete(write_test_data(test_vector))
asyncio.async(write_test_data(test_dict)) asyncio.async(write_test_data(test_dict))
self.subscriber = sync_struct.Subscriber("test", self.init_test_dict) self.subscriber = sync_struct.Subscriber("test", self.init_test_dict,
self.notify)
loop.run_until_complete(self.subscriber.connect(test_address, loop.run_until_complete(self.subscriber.connect(test_address,
test_port)) test_port))
loop.run_until_complete(self.do_recv()) loop.run_until_complete(self.receiving_done.wait())
self.assertEqual(self.test_dict, test_vector) self.assertEqual(self.test_dict, test_vector)
self.loop.run_until_complete(self.subscriber.close()) self.loop.run_until_complete(self.subscriber.close())
self.loop.run_until_complete(self.publisher.stop()) self.loop.run_until_complete(self.publisher.stop())