test_websocket_listener.py 文件源码

def test__skips_notify_on_controllerinfo_interface_update(self):
        yield deferToDatabase(register_websocket_triggers)
        listener = self.make_listener_without_delay()
        dv = DeferredValue()
        params = self.params.copy()
        controller = yield deferToDatabase(self.create_node, params)
        yield deferToDatabase(self.set_version, controller, '')
        listener.register(self.listener, lambda *args: dv.set(args))
        yield listener.startService()
        try:
            yield deferToDatabase(
                self.set_interface_update_info, controller, '{]', '{}')
            with ExpectedException(CancelledError):
                yield dv.get(timeout=0.2)
        finally:
            yield listener.stopService()