supervisor_test_case.py 文件源码

def tearDown(self):
        # You can't import this gobally because DJANGO_SETTINGS_MODULE is not initialized yet for some
        # reason, but maybe by the time the code meanders its way to here it will work.
        from chroma_core.services.rpc import RpcClientFactory

        test_failed = (sys.exc_info() != (None, None, None))

        if test_failed:
            log.info(self._xmlrpc.system.listMethods())
            log_chunk = self._xmlrpc.supervisor.readLog(0, 4096)
            log.error("Supervisor log: %s" % log_chunk)

        # Shutdown any RPC Threads if they were started. Bit of horrible insider knowledge here.
        if RpcClientFactory._lightweight is False:
            RpcClientFactory.shutdown_threads()
            RpcClientFactory._lightweight = True
            RpcClientFactory._available = True
            RpcClientFactory._instances = {}

        if self._supervisor is not None:
            try:
                self._xmlrpc.supervisor.shutdown()
                stdout, stderr = self._supervisor.communicate()
                # Echo these at the end: by outputting using sys.std* rather than
                # letting the subprocess write directly, this verbose output can be
                # captured by nose and only output on failure.
                sys.stdout.write(stdout)
                sys.stdout.write(stderr)
            except:
                self._supervisor.kill()
            finally:
                self._supervisor = None

        if self._tmp_conf and os.path.exists(self._tmp_conf.name):
            os.unlink(self._tmp_conf.name)