lbry-sdk/lbry/tests/unit/wallet/test_stream_controller.py

30 lines
900 B
Python
Raw Normal View History

2019-08-07 07:48:40 +02:00
from torba.stream import StreamController
2019-12-11 23:31:45 +01:00
from torba.tasks import TaskGroup
2019-08-07 16:27:25 +02:00
from torba.testcase import AsyncioTestCase
2019-08-07 07:48:40 +02:00
2019-08-07 16:27:25 +02:00
class StreamControllerTestCase(AsyncioTestCase):
2019-08-07 07:48:40 +02:00
def test_non_unique_events(self):
events = []
controller = StreamController()
controller.stream.listen(on_data=events.append)
controller.add("yo")
controller.add("yo")
self.assertListEqual(events, ["yo", "yo"])
2019-08-07 07:48:40 +02:00
def test_unique_events(self):
events = []
controller = StreamController(merge_repeated_events=True)
controller.stream.listen(on_data=events.append)
controller.add("yo")
controller.add("yo")
self.assertListEqual(events, ["yo"])
2019-12-11 23:31:45 +01:00
class TaskGroupTestCase(AsyncioTestCase):
async def test_cancel_sets_it_done(self):
2019-12-11 23:31:45 +01:00
group = TaskGroup()
group.cancel()
2019-12-12 03:33:34 +01:00
self.assertTrue(group.done.is_set())