M-Labs
/
artiq
mirror of https://github.com/m-labs/artiq.git
2
0
Fork 2
Code Issues Projects Releases Wiki Activity

test/sync_struct: cleanup

This commit is contained in:
Sebastien Bourdeauducq 2015-10-03 19:40:11 +08:00
parent f552d62b69
commit 090a7e5871
1 changed files with 18 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
artiq/test/sync_struct.py
View File

@ -29,18 +29,9 @@ def write_test_data(test_dict):
test_dict["finished"] = True test_dict["finished"] = True
async def start_server(publisher_future, test_dict_future):
test_dict = sync_struct.Notifier(dict())
publisher = sync_struct.Publisher(
{"test": test_dict})
await publisher.start(test_address, test_port)
publisher_future.set_result(publisher)
test_dict_future.set_result(test_dict)
class SyncStructCase(unittest.TestCase): class SyncStructCase(unittest.TestCase):
def init_test_dict(self, init): def init_test_dict(self, init):
self.test_dict = init self.received_dict = init
return init return init
def notify(self, mod): def notify(self, mod):
@ -52,29 +43,27 @@ class SyncStructCase(unittest.TestCase):
self.loop = asyncio.new_event_loop() self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.loop) asyncio.set_event_loop(self.loop)
def test_recv(self): async def _do_test_recv(self):
loop = self.loop
self.receiving_done = asyncio.Event() self.receiving_done = asyncio.Event()
publisher = asyncio.Future()
test_dict = asyncio.Future()
asyncio.ensure_future(start_server(publisher, test_dict))
loop.run_until_complete(publisher)
loop.run_until_complete(test_dict)
self.publisher = publisher.result() test_dict = sync_struct.Notifier(dict())
test_dict = test_dict.result() publisher = sync_struct.Publisher({"test": test_dict})
test_vector = dict() await publisher.start(test_address, test_port)
write_test_data(test_vector)
subscriber = sync_struct.Subscriber("test", self.init_test_dict,
self.notify)
await subscriber.connect(test_address, test_port)
write_test_data(test_dict) write_test_data(test_dict)
self.subscriber = sync_struct.Subscriber("test", self.init_test_dict, await self.receiving_done.wait()
self.notify)
loop.run_until_complete(self.subscriber.connect(test_address, await subscriber.close()
test_port)) await publisher.stop()
loop.run_until_complete(self.receiving_done.wait())
self.assertEqual(self.test_dict, test_vector) self.assertEqual(self.received_dict, test_dict.read)
self.loop.run_until_complete(self.subscriber.close())
self.loop.run_until_complete(self.publisher.stop()) def test_recv(self):
self.loop.run_until_complete(self._do_test_recv())
def tearDown(self): def tearDown(self):
self.loop.close() self.loop.close()