pytest_plugin.py 文件源码

def kill_crossbar(try_again=True):
    pids = get_process_ids()
    if pids and try_again is True:
        logger.error(
            "Crossbar.io did not stop when sig term issued!"
        )

    for pid_as_str in pids:
        try:
            pid = os.getpgid(int(pid_as_str))
        except OSError:
            continue

        logger.warning("OS sending SIGTERM to crossbar pid: %s", pid)

        try:
            os.kill(pid, signal.SIGTERM)
        except Exception:  # anything Twisted raises
            logger.exception(
                "Failed to terminate router process: %s", pid
            )

            try:
                os.waitpid(pid, options=os.WNOHANG)
            except OSError:
                pass

            try:
                os.kill(pid, signal.SIGKILL)
            except Exception as exc:
                if "No such process" in str(exc):
                    continue
                logger.exception(
                    "Failed again to terminate router process: %s", pid)

    pids = get_process_ids()
    if pids and try_again is True:
        logger.warning('try one more time to shutdown Crossbar')
        sleep(5)
        kill_crossbar(try_again=False)
    elif pids and try_again is False:
        logger.error("Failed to shutdown all router processes")