StealthOS/subiquity
StealthOS
/
subiquity
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

pubsub: fix test fragility

This commit is contained in:
Dan Bungert 2022-05-24 13:31:04 -06:00
parent a97333bebb
commit 884f1c87ae
1 changed files with 8 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
subiquitycore/tests/test_pubsub.py
View File

@ -13,38 +13,37 @@
# You should have received a copy of the GNU Affero General Public License # You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>. # along with this program. If not, see <http://www.gnu.org/licenses/>.
import unittest
from unittest.mock import MagicMock from unittest.mock import MagicMock
from subiquitycore.tests import SubiTestCase
from subiquitycore.pubsub import MessageHub from subiquitycore.pubsub import MessageHub
from subiquitycore.tests.util import run_coro
class TestMessageHub(SubiTestCase): class TestMessageHub(unittest.IsolatedAsyncioTestCase):
def setUp(self): def setUp(self):
self.hub = MessageHub() self.hub = MessageHub()
def test_multicall(self): async def test_multicall(self):
cb = MagicMock() cb = MagicMock()
expected_calls = 3 expected_calls = 3
channel_id = 1234 channel_id = 1234
for _ in range(expected_calls): for _ in range(expected_calls):
self.hub.subscribe(channel_id, cb) self.hub.subscribe(channel_id, cb)
run_coro(self.hub.abroadcast(channel_id)) await self.hub.abroadcast(channel_id)
self.assertEqual(expected_calls, cb.call_count) self.assertEqual(expected_calls, cb.call_count)
def test_multisubscriber(self): async def test_multisubscriber(self):
cbs = [MagicMock() for _ in range(4)] cbs = [MagicMock() for _ in range(4)]
channel_id = 2345 channel_id = 2345
for cb in cbs: for cb in cbs:
self.hub.subscribe(channel_id, cb) self.hub.subscribe(channel_id, cb)
run_coro(self.hub.abroadcast(channel_id)) await self.hub.abroadcast(channel_id)
for cb in cbs: for cb in cbs:
cb.assert_called_once_with() cb.assert_called_once_with()
def test_message_arg(self): async def test_message_arg(self):
cb = MagicMock() cb = MagicMock()
channel_id = 'test-message-arg' channel_id = 'test-message-arg'
self.hub.subscribe(channel_id, cb) self.hub.subscribe(channel_id, cb)
run_coro(self.hub.abroadcast(channel_id, '0', 1, 'two', [3], four=4)) await self.hub.abroadcast(channel_id, '0', 1, 'two', [3], four=4)
cb.assert_called_once_with('0', 1, 'two', [3], four=4) cb.assert_called_once_with('0', 1, 'two', [3], four=4)