test_system_listener.py 文件源码

def test_doesnt_sends_message_for_update_on_node_alloc_type_no_ip(self):
        yield deferToDatabase(register_system_triggers)
        node = yield deferToDatabase(self.create_node_with_interface)
        sip = yield deferToDatabase(self.get_node_ip_address, node)
        yield self.capturePublication()
        dv = DeferredValue()
        listener = self.make_listener_without_delay()
        listener.register(
            "sys_dns", lambda *args: dv.set(args))
        yield listener.startService()
        try:
            yield deferToDatabase(self.update_staticipaddress, sip.id, {
                "alloc_type": IPADDRESS_TYPE.STICKY,
            })
            with ExpectedException(CancelledError):
                yield dv.get(timeout=1)
        finally:
            yield listener.stopService()