integrationtest.py 文件源码

def teardown_integration_test(self):
        """
        Stops the broker running in the separate process.
        :return:
        """
        try:
            stop_broker()
            if self.broker_process is not None:
                self.broker_process.join(timeout=_BROKER_SHUTDOWN_TIMEOUT_TIME)
        except Exception as _:
            logging.log(msg="Broker didn't shut down.  Killing broker process.", level=logging.WARNING)
        self.broker_process.terminate()
        self.broker_process.join(timeout=_BROKER_SHUTDOWN_TIMEOUT_TIME)
        if self.broker_process.is_alive():
            logging.log(msg="Broker won't terminate.  Integration test exiting.", level=logging.ERROR)
            sys.exit(1)
        self.broker_process = None