archive/ipc-unix
Archived
1
Fork 0
Code Releases Packages Activity

test getting latest message

Browse source
This commit is contained in:
Jake Howard 2018-12-08 12:37:57 +00:00
parent e605438d33
commit 65fd392a70
Signed by: jake
GPG key ID: 57AFB45680EDD477
1 changed files with 7 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
tests/test_pubsub.py
View file

@ -30,6 +30,13 @@ class PubSubTestCase(TestCase):
message_ids = [message["data"] for message in all_messages] message_ids = [message["data"] for message in all_messages]
self.assertEqual(message_ids, [0, 1, 2, 3, 4]) self.assertEqual(message_ids, [0, 1, 2, 3, 4])
def test_get_latest_message(self):
for i in range(5):
self.publisher.write({"data": i})
latest_message = self.subscriber.get_latest_message()
self.assertEqual(latest_message, {"data": 4})
self.assertIsNone(self.subscriber.get_latest_message())
def test_multiple_subscribers(self): def test_multiple_subscribers(self):
subscriber_2 = pubsub.Subscriber(self.socket_path) subscriber_2 = pubsub.Subscriber(self.socket_path)
self.publisher.write({"foo": "bar"}) self.publisher.write({"foo": "bar"})