test_power.py 文件源码

def test__handles_timeout(self):
        node, power_info = yield deferToDatabase(
            self.make_node_with_power_info)

        def defer_way_later(*args, **kwargs):
            # Create a defer that will finish in 1 minute.
            return deferLater(reactor, 60 * 60, lambda: None)

        rack_id = factory.make_name("system_id")
        client = Mock()
        client.ident = rack_id
        client.side_effect = defer_way_later

        self.patch(power_module, "getAllClients").return_value = [client]
        power_state, success_racks, failed_racks = yield power_query_all(
            node.system_id, node.hostname, power_info, timeout=0.5)

        self.assertEqual(POWER_STATE.UNKNOWN, power_state)
        self.assertItemsEqual([], success_racks)
        self.assertItemsEqual([rack_id], failed_racks)