StealthOS/subiquity
StealthOS
/
subiquity
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
subiquity/subiquitycore/tests/test_pubsub.py

51 lines
1.8 KiB
Python
Raw Normal View History

Add EventCallback, and move mirror task things to GeoIP (#983) Move mock_app to common location. Move run_coro to subiquitycore so that subiquitycore doesn't have to reference things in subiquity, even for test. Move task tracking things from mirror to geoip. Server app owns the geoip instance. Create EventCallback as an alternative to MessageHub that should hopefully express clearer intermodule dependencies.
2021-06-14 22:05:27 +00:00
# Copyright 2021 Canonical, Ltd.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# 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/>.
pubsub: remove subscribe args, more broadcast args Remove the subscribe time args, as they weren't used. With that flexibility we can add many args at broadcast time.
2021-09-24 19:06:41 +00:00
from unittest.mock import MagicMock
Add EventCallback, and move mirror task things to GeoIP (#983) Move mock_app to common location. Move run_coro to subiquitycore so that subiquitycore doesn't have to reference things in subiquity, even for test. Move task tracking things from mirror to geoip. Server app owns the geoip instance. Create EventCallback as an alternative to MessageHub that should hopefully express clearer intermodule dependencies.
2021-06-14 22:05:27 +00:00
from subiquitycore.tests import SubiTestCase
EventCallback: remove
2021-09-01 23:58:09 +00:00
from subiquitycore.pubsub import MessageHub
Add EventCallback, and move mirror task things to GeoIP (#983) Move mock_app to common location. Move run_coro to subiquitycore so that subiquitycore doesn't have to reference things in subiquity, even for test. Move task tracking things from mirror to geoip. Server app owns the geoip instance. Create EventCallback as an alternative to MessageHub that should hopefully express clearer intermodule dependencies.
2021-06-14 22:05:27 +00:00
from subiquitycore.tests.util import run_coro
pubsub: Add unit test for MessageHub Also minor cleanup on the soon-to-be-removed EventCallback test to help demonostrate that some of this code isn't needed.
2021-09-01 23:46:38 +00:00
class TestMessageHub(SubiTestCase):
pubsub: remove subscribe args, more broadcast args Remove the subscribe time args, as they weren't used. With that flexibility we can add many args at broadcast time.
2021-09-24 19:06:41 +00:00
def setUp(self):
self.hub = MessageHub()
pubsub: Add unit test for MessageHub Also minor cleanup on the soon-to-be-removed EventCallback test to help demonostrate that some of this code isn't needed.
2021-09-01 23:46:38 +00:00
pubsub: remove subscribe args, more broadcast args Remove the subscribe time args, as they weren't used. With that flexibility we can add many args at broadcast time.
2021-09-24 19:06:41 +00:00
def test_multicall(self):
cb = MagicMock()
pubsub: optional argument at broadcast time
2021-09-22 19:31:53 +00:00
expected_calls = 3
pubsub: Add unit test for MessageHub Also minor cleanup on the soon-to-be-removed EventCallback test to help demonostrate that some of this code isn't needed.
2021-09-01 23:46:38 +00:00
channel_id = 1234
pubsub: optional argument at broadcast time
2021-09-22 19:31:53 +00:00
for _ in range(expected_calls):
pubsub: remove subscribe args, more broadcast args Remove the subscribe time args, as they weren't used. With that flexibility we can add many args at broadcast time.
2021-09-24 19:06:41 +00:00
self.hub.subscribe(channel_id, cb)
run_coro(self.hub.abroadcast(channel_id))
self.assertEqual(expected_calls, cb.call_count)
pubsub: optional argument at broadcast time
2021-09-22 19:31:53 +00:00
pubsub: remove subscribe args, more broadcast args Remove the subscribe time args, as they weren't used. With that flexibility we can add many args at broadcast time.
2021-09-24 19:06:41 +00:00
def test_multisubscriber(self):
cbs = [MagicMock() for _ in range(4)]
channel_id = 2345
for cb in cbs:
self.hub.subscribe(channel_id, cb)
run_coro(self.hub.abroadcast(channel_id))
for cb in cbs:
cb.assert_called_once_with()
pubsub: optional argument at broadcast time
2021-09-22 19:31:53 +00:00
pubsub: remove subscribe args, more broadcast args Remove the subscribe time args, as they weren't used. With that flexibility we can add many args at broadcast time.
2021-09-24 19:06:41 +00:00
def test_message_arg(self):
cb = MagicMock()
pubsub: optional argument at broadcast time
2021-09-22 19:31:53 +00:00
channel_id = 'test-message-arg'
pubsub: remove subscribe args, more broadcast args Remove the subscribe time args, as they weren't used. With that flexibility we can add many args at broadcast time.
2021-09-24 19:06:41 +00:00
self.hub.subscribe(channel_id, cb)
run_coro(self.hub.abroadcast(channel_id, '0', 1, 'two', [3], four=4))
cb.assert_called_once_with('0', 1, 'two', [3], four=4)