test_watchdog.py 文件源码

def test_simulate_broker_not_starting_up(self):
        """
        When a daemon repeatedly dies, the watchdog gives up entirely and shuts
        down.
        """
        stop = []
        stopped = []
        self.log_helper.ignore_errors("Can't keep landscape-broker running. "
                                      "Exiting.")

        output_filename = self.makeFile("NOT RUN")

        self._write_script(
            "#!/bin/sh\necho RUN >> %s" % output_filename)

        def got_result(result):
            self.assertEqual(len(list(open(output_filename))),
                             MAXIMUM_CONSECUTIVE_RESTARTS)

            self.assertTrue("Can't keep landscape-broker running." in
                            self.logfile.getvalue())
            self.assertCountEqual([True], stopped)
            reactor.stop = stop[0]

        result = Deferred()
        result.addCallback(got_result)

        def mock_reactor_stop():
            stop.append(reactor.stop)
            reactor.stop = lambda: stopped.append(True)

        reactor.callLater(0, mock_reactor_stop)
        reactor.callLater(1, result.callback, None)

        daemon = self.get_daemon(reactor=reactor)
        daemon.BIN_DIR = self.config.bindir
        daemon.start()

        return result