94 lines
2.3 KiB
Python
94 lines
2.3 KiB
Python
# Copyright 2021 (c) Anna Schumaker.
|
|
from . import bus
|
|
from . import publisher
|
|
from gi.repository import GLib
|
|
import threading
|
|
import time
|
|
import unittest
|
|
|
|
main_context = GLib.main_context_default()
|
|
|
|
class TestBus(unittest.TestCase):
|
|
def setUp(self):
|
|
self.count = 0
|
|
self.start = 0
|
|
self.retry = False
|
|
bus.Start.register(self.cb_start)
|
|
|
|
def tearDown(self):
|
|
bus.Start.unregister(self.cb_start)
|
|
|
|
def cb_start(self, bus):
|
|
self.start += 1
|
|
|
|
def cb_one(self):
|
|
self.count += 1
|
|
|
|
def cb_two(self, arg):
|
|
self.count = arg
|
|
|
|
def cb_retry(self, arg):
|
|
self.count = arg
|
|
return bus.RETRY if self.retry == True else None
|
|
|
|
def test_bus_init(self):
|
|
self.assertEqual(bus.RETRY, GLib.SOURCE_CONTINUE)
|
|
|
|
b = bus.Bus(100)
|
|
self.assertEqual(b.timeout, 100)
|
|
self.assertEqual(b.passengers, [ ])
|
|
self.assertIsNone(b.timeout_id)
|
|
self.assertIsInstance(b.lock, type(threading.Lock()))
|
|
|
|
self.assertIsInstance(bus.Start, publisher.Publisher)
|
|
|
|
def test_bus_board(self):
|
|
b = bus.Bus(100)
|
|
b.board(self.cb_one)
|
|
b.board(self.cb_one)
|
|
|
|
self.assertEqual(b.passengers, [ (self.cb_one,()) ])
|
|
self.assertIsNotNone(b.timeout_id)
|
|
self.assertTrue(b.running())
|
|
self.assertEqual(self.start, 1)
|
|
|
|
time.sleep(0.1)
|
|
while main_context.iteration(may_block=False): pass
|
|
|
|
self.assertEqual(self.count, 1)
|
|
self.assertIsNone(b.timeout_id)
|
|
self.assertFalse(b.running())
|
|
|
|
def test_bus_clear(self):
|
|
b = bus.Bus(100)
|
|
b.clear()
|
|
|
|
for i in range(100):
|
|
b.board(self.cb_one)
|
|
|
|
b.clear()
|
|
self.assertEqual(b.passengers, [ ])
|
|
self.assertIsNone(b.timeout_id)
|
|
|
|
def test_bus_complete(self):
|
|
b = bus.Bus(100)
|
|
for i in range(100):
|
|
b.board(self.cb_two, i)
|
|
|
|
b.complete()
|
|
self.assertEqual(b.passengers, [ ])
|
|
self.assertEqual(self.count, i)
|
|
self.assertIsNone(b.timeout_id)
|
|
|
|
def test_bus_retry(self):
|
|
b = bus.Bus(10)
|
|
b.board(self.cb_retry, 1)
|
|
b.board(self.cb_retry, 2)
|
|
|
|
self.retry = True
|
|
b.run()
|
|
self.assertEqual(b.passengers, [ (self.cb_retry, (1,)), (self.cb_retry, (2,)) ])
|
|
|
|
b.complete()
|
|
self.assertEqual(b.passengers, [ ])
|