| #!/usr/bin/env python3 |
| # |
| # Copyright 2018 - The Android Open Source Project |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| import unittest |
| from unittest import TestCase |
| |
| from mock import Mock |
| from mock import patch |
| |
| from acts.event import event_bus |
| from acts.event.event import Event |
| from acts.event.event_subscription import EventSubscription |
| |
| |
| class EventBusTest(TestCase): |
| """Tests the event_bus functions.""" |
| |
| def setUp(self): |
| """Clears all state from the event_bus between test cases.""" |
| event_bus._event_bus = event_bus._EventBus() |
| |
| def get_subscription_argument(self, register_subscription_call): |
| """Gets the subscription argument from a register_subscription call.""" |
| return register_subscription_call[0][0] |
| |
| @patch('acts.event.event_bus._event_bus.register_subscription') |
| def test_register_registers_a_subscription(self, register_subscription): |
| """Tests that register creates and registers a subscription.""" |
| mock_event = Mock() |
| mock_func = Mock() |
| order = 43 |
| event_bus.register(mock_event, mock_func, order=order) |
| |
| args, _ = register_subscription.call_args |
| subscription = args[0] |
| |
| # Instead of writing an equality operator for only testing, |
| # check the internals to make sure they are expected values. |
| self.assertEqual(subscription._event_type, mock_event) |
| self.assertEqual(subscription._func, mock_func) |
| self.assertEqual(subscription.order, order) |
| |
| @patch('acts.event.event_bus._event_bus.register_subscription') |
| def test_register_subscriptions_for_list(self, register_subscription): |
| """Tests that register_subscription is called for each subscription.""" |
| mocks = [Mock(), Mock(), Mock()] |
| subscriptions = [ |
| EventSubscription(mocks[0], lambda _: None), |
| EventSubscription(mocks[1], lambda _: None), |
| EventSubscription(mocks[2], lambda _: None), |
| ] |
| |
| event_bus.register_subscriptions(subscriptions) |
| received_subscriptions = set() |
| for index, call in enumerate(register_subscription.call_args_list): |
| received_subscriptions.add(self.get_subscription_argument(call)) |
| |
| self.assertEqual(register_subscription.call_count, len(subscriptions)) |
| self.assertSetEqual(received_subscriptions, set(subscriptions)) |
| |
| def test_register_subscription_new_event_type(self): |
| """Tests that the event_bus can register a new event type.""" |
| mock_type = Mock() |
| bus = event_bus._event_bus |
| subscription = EventSubscription(mock_type, lambda _: None) |
| |
| reg_id = event_bus.register_subscription(subscription) |
| |
| self.assertTrue(mock_type in bus._subscriptions.keys()) |
| self.assertTrue(subscription in bus._subscriptions[mock_type]) |
| self.assertTrue(reg_id in bus._registration_id_map.keys()) |
| |
| def test_register_subscription_existing_type(self): |
| """Tests that the event_bus can register an existing event type.""" |
| mock_type = Mock() |
| bus = event_bus._event_bus |
| bus._subscriptions[mock_type] = [ |
| EventSubscription(mock_type, lambda _: None) |
| ] |
| new_subscription = EventSubscription(mock_type, lambda _: True) |
| |
| reg_id = event_bus.register_subscription(new_subscription) |
| |
| self.assertTrue(new_subscription in bus._subscriptions[mock_type]) |
| self.assertTrue(reg_id in bus._registration_id_map.keys()) |
| |
| def test_post_to_unregistered_event_does_not_call_other_funcs(self): |
| """Tests posting an unregistered event will not call other funcs.""" |
| mock_subscription = Mock() |
| bus = event_bus._event_bus |
| mock_type = Mock() |
| mock_subscription.event_type = mock_type |
| bus._subscriptions[mock_type] = [mock_subscription] |
| |
| event_bus.post(Mock()) |
| |
| self.assertEqual(mock_subscription.deliver.call_count, 0) |
| |
| def test_post_to_registered_event_calls_all_registered_funcs(self): |
| """Tests posting to a registered event calls all registered funcs.""" |
| mock_subscriptions = [Mock(), Mock(), Mock()] |
| bus = event_bus._event_bus |
| for subscription in mock_subscriptions: |
| subscription.order = 0 |
| mock_event = Mock() |
| bus._subscriptions[type(mock_event)] = mock_subscriptions |
| |
| event_bus.post(mock_event) |
| |
| for subscription in mock_subscriptions: |
| subscription.deliver.assert_called_once_with(mock_event) |
| |
| def test_post_with_ignore_errors_calls_all_registered_funcs(self): |
| """Tests posting with ignore_errors=True calls all registered funcs, |
| even if they raise errors. |
| """ |
| |
| def _raise(_): |
| raise Exception |
| |
| mock_event = Mock() |
| mock_subscriptions = [Mock(), Mock(), Mock()] |
| mock_subscriptions[0].deliver.side_effect = _raise |
| bus = event_bus._event_bus |
| for i, subscription in enumerate(mock_subscriptions): |
| subscription.order = i |
| bus._subscriptions[type(mock_event)] = mock_subscriptions |
| |
| event_bus.post(mock_event, ignore_errors=True) |
| |
| for subscription in mock_subscriptions: |
| subscription.deliver.assert_called_once_with(mock_event) |
| |
| @patch('acts.event.event_bus._event_bus.unregister') |
| def test_unregister_all_from_list(self, unregister): |
| """Tests unregistering from a list unregisters the specified list.""" |
| unregister_list = [Mock(), Mock()] |
| |
| event_bus.unregister_all(from_list=unregister_list) |
| |
| self.assertEqual(unregister.call_count, len(unregister_list)) |
| for args, _ in unregister.call_args_list: |
| subscription = args[0] |
| self.assertTrue(subscription in unregister_list) |
| |
| @patch('acts.event.event_bus._event_bus.unregister') |
| def test_unregister_all_from_event(self, unregister): |
| """Tests that all subscriptions under the event are unregistered.""" |
| mock_event = Mock() |
| mock_event_2 = Mock() |
| bus = event_bus._event_bus |
| unregister_list = [Mock(), Mock()] |
| bus._subscriptions[type(mock_event_2)] = [Mock(), Mock(), Mock()] |
| bus._subscriptions[type(mock_event)] = unregister_list |
| for sub_type in bus._subscriptions.keys(): |
| for subscription in bus._subscriptions[sub_type]: |
| subscription.event_type = sub_type |
| bus._registration_id_map[id(subscription)] = subscription |
| |
| event_bus.unregister_all(from_event=type(mock_event)) |
| |
| self.assertEqual(unregister.call_count, len(unregister_list)) |
| for args, _ in unregister.call_args_list: |
| subscription = args[0] |
| self.assertTrue(subscription in unregister_list) |
| |
| @patch('acts.event.event_bus._event_bus.unregister') |
| def test_unregister_all_no_args_unregisters_everything(self, unregister): |
| """Tests unregister_all without arguments will unregister everything.""" |
| mock_event_1 = Mock() |
| mock_event_2 = Mock() |
| bus = event_bus._event_bus |
| unregister_list_1 = [Mock(), Mock()] |
| unregister_list_2 = [Mock(), Mock(), Mock()] |
| bus._subscriptions[type(mock_event_1)] = unregister_list_1 |
| bus._subscriptions[type(mock_event_2)] = unregister_list_2 |
| for sub_type in bus._subscriptions.keys(): |
| for subscription in bus._subscriptions[sub_type]: |
| subscription.event_type = sub_type |
| bus._registration_id_map[id(subscription)] = subscription |
| |
| event_bus.unregister_all() |
| |
| self.assertEqual(unregister.call_count, |
| len(unregister_list_1) + len(unregister_list_2)) |
| for args, _ in unregister.call_args_list: |
| subscription = args[0] |
| self.assertTrue(subscription in unregister_list_1 |
| or subscription in unregister_list_2) |
| |
| def test_unregister_given_an_event_subscription(self): |
| """Tests that unregister can unregister a given EventSubscription.""" |
| mock_event = Mock() |
| bus = event_bus._event_bus |
| subscription = EventSubscription(type(mock_event), lambda _: None) |
| bus._registration_id_map[id(subscription)] = subscription |
| bus._subscriptions[type(mock_event)] = [subscription] |
| |
| val = event_bus.unregister(subscription) |
| |
| self.assertTrue(val) |
| self.assertTrue(subscription not in bus._registration_id_map) |
| self.assertTrue( |
| subscription not in bus._subscriptions[type(mock_event)]) |
| |
| def test_unregister_given_a_registration_id(self): |
| """Tests that unregister can unregister a given EventSubscription.""" |
| mock_event = Mock() |
| bus = event_bus._event_bus |
| subscription = EventSubscription(type(mock_event), lambda _: None) |
| registration_id = id(subscription) |
| bus._registration_id_map[id(subscription)] = subscription |
| bus._subscriptions[type(mock_event)] = [subscription] |
| |
| val = event_bus.unregister(registration_id) |
| |
| self.assertTrue(val) |
| self.assertTrue(subscription not in bus._registration_id_map) |
| self.assertTrue( |
| subscription not in bus._subscriptions[type(mock_event)]) |
| |
| def test_unregister_given_object_that_is_not_a_subscription(self): |
| """Asserts that a ValueError is raised upon invalid arguments.""" |
| with self.assertRaises(ValueError): |
| event_bus.unregister(Mock()) |
| |
| def test_unregister_given_invalid_registration_id(self): |
| """Asserts that a false is returned upon invalid registration_id.""" |
| val = event_bus.unregister(9) |
| self.assertFalse(val) |
| |
| def test_listen_for_registers_listener(self): |
| """Tests listen_for registers the listener within the with statement.""" |
| bus = event_bus._event_bus |
| |
| def event_listener(_): |
| pass |
| |
| with event_bus.listen_for(Event, event_listener): |
| self.assertEqual(len(bus._registration_id_map), 1) |
| |
| def test_listen_for_unregisters_listener(self): |
| """Tests listen_for unregisters the listener after the with statement. |
| """ |
| bus = event_bus._event_bus |
| |
| def event_listener(_): |
| pass |
| |
| with event_bus.listen_for(Event, event_listener): |
| pass |
| |
| self.assertEqual(len(bus._registration_id_map), 0) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
