test_listener.py 文件源码

def test_on_remote_event(self):
        """ Test the reception of a Supervisor remote comm event. """
        from supvisors.listener import SupervisorListener
        listener = SupervisorListener(self.supvisors)
        # add patches for what is tested just above
        with patch.multiple(listener, unstack_event=DEFAULT,
                unstack_info=DEFAULT, authorization=DEFAULT):
            # test unknown type
            event = Mock(type='unknown', data='')
            listener.on_remote_event(event)
            self.assertFalse(listener.unstack_event.called)
            self.assertFalse(listener.unstack_info.called)
            self.assertFalse(listener.authorization.called)
            # test event
            event = Mock(type='event', data={'state': 'RUNNING'})
            listener.on_remote_event(event)
            self.assertEqual([call({'state': 'RUNNING'})],
                listener.unstack_event.call_args_list)
            self.assertFalse(listener.unstack_info.called)
            self.assertFalse(listener.authorization.called)
            listener.unstack_event.reset_mock()
            # test info
            event = Mock(type='info', data={'name': 'dummy_process'})
            listener.on_remote_event(event)
            self.assertFalse(listener.unstack_event.called)
            self.assertEqual([call({'name': 'dummy_process'})],
                listener.unstack_info.call_args_list)
            self.assertFalse(listener.authorization.called)
            listener.unstack_info.reset_mock()
            # test authorization
            event = Mock(type='auth', data=('10.0.0.1', True))
            listener.on_remote_event(event)
            self.assertFalse(listener.unstack_event.called)
            self.assertFalse(listener.unstack_info.called)
            self.assertEqual([call(('10.0.0.1', True))],
                listener.authorization.call_args_list)