python类SIGBREAK的实例源码

popen_spawn.py 文件源码 项目:watchmen 作者: lycclsltt 项目源码 文件源码 阅读 26 收藏 0 点赞 0 评论 0
def kill(self, sig):
        '''Sends a Unix signal to the subprocess.

        Use constants from the :mod:`signal` module to specify which signal.
        '''
        if sys.platform == 'win32':
            if sig in [signal.SIGINT, signal.CTRL_C_EVENT]:
                sig = signal.CTRL_C_EVENT
            elif sig in [signal.SIGBREAK, signal.CTRL_BREAK_EVENT]:
                sig = signal.CTRL_BREAK_EVENT
            else:
                sig = signal.SIGTERM

        os.kill(self.proc.pid, sig)
_utils.py 文件源码 项目:cotyledon 作者: sileht 项目源码 文件源码 阅读 26 收藏 0 点赞 0 评论 0
def __init__(self, master=False):
        # Setup signal fd, this allows signal to behave correctly
        if os.name == 'posix':
            self.signal_pipe_r, self.signal_pipe_w = os.pipe()
            self._set_nonblock(self.signal_pipe_r)
            self._set_nonblock(self.signal_pipe_w)
            signal.set_wakeup_fd(self.signal_pipe_w)

        self._signals_received = collections.deque()

        signal.signal(signal.SIGINT, signal.SIG_DFL)
        if os.name == 'posix':
            signal.signal(signal.SIGCHLD, signal.SIG_DFL)
            signal.signal(signal.SIGTERM, self._signal_catcher)
            signal.signal(signal.SIGALRM, self._signal_catcher)
            signal.signal(signal.SIGHUP, self._signal_catcher)
        else:
            # currently a noop on window...
            signal.signal(signal.SIGTERM, self._signal_catcher)
            # FIXME(sileht): should allow to catch signal CTRL_BREAK_EVENT,
            # but we to create the child process with CREATE_NEW_PROCESS_GROUP
            # to make this work, so current this is a noop for later fix
            signal.signal(signal.SIGBREAK, self._signal_catcher)
test_signal.py 文件源码 项目:web_ctp 作者: molebot 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def test_issue9324(self):
        # Updated for issue #10003, adding SIGBREAK
        handler = lambda x, y: None
        checked = set()
        for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
                    signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
                    signal.SIGTERM):
            # Set and then reset a handler for signals that work on windows.
            # Issue #18396, only for signals without a C-level handler.
            if signal.getsignal(sig) is not None:
                signal.signal(sig, signal.signal(sig, handler))
                checked.add(sig)
        # Issue #18396: Ensure the above loop at least tested *something*
        self.assertTrue(checked)

        with self.assertRaises(ValueError):
            signal.signal(-1, handler)

        with self.assertRaises(ValueError):
            signal.signal(7, handler)
test_signal.py 文件源码 项目:ouroboros 作者: pybee 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def test_issue9324(self):
        # Updated for issue #10003, adding SIGBREAK
        handler = lambda x, y: None
        checked = set()
        for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
                    signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
                    signal.SIGTERM):
            # Set and then reset a handler for signals that work on windows.
            # Issue #18396, only for signals without a C-level handler.
            if signal.getsignal(sig) is not None:
                signal.signal(sig, signal.signal(sig, handler))
                checked.add(sig)
        # Issue #18396: Ensure the above loop at least tested *something*
        self.assertTrue(checked)

        with self.assertRaises(ValueError):
            signal.signal(-1, handler)

        with self.assertRaises(ValueError):
            signal.signal(7, handler)
test_signal.py 文件源码 项目:kbe_server 作者: xiaohaoppy 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def test_issue9324(self):
        # Updated for issue #10003, adding SIGBREAK
        handler = lambda x, y: None
        checked = set()
        for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
                    signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
                    signal.SIGTERM):
            # Set and then reset a handler for signals that work on windows.
            # Issue #18396, only for signals without a C-level handler.
            if signal.getsignal(sig) is not None:
                signal.signal(sig, signal.signal(sig, handler))
                checked.add(sig)
        # Issue #18396: Ensure the above loop at least tested *something*
        self.assertTrue(checked)

        with self.assertRaises(ValueError):
            signal.signal(-1, handler)

        with self.assertRaises(ValueError):
            signal.signal(7, handler)
popen_spawn.py 文件源码 项目:leetcode 作者: thomasyimgit 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0
def kill(self, sig):
        '''Sends a Unix signal to the subprocess.

        Use constants from the :mod:`signal` module to specify which signal.
        '''
        if sys.platform == 'win32':
            if sig in [signal.SIGINT, signal.CTRL_C_EVENT]:
                sig = signal.CTRL_C_EVENT
            elif sig in [signal.SIGBREAK, signal.CTRL_BREAK_EVENT]:
                sig = signal.CTRL_BREAK_EVENT
            else:
                sig = signal.SIGTERM

        os.kill(self.proc.pid, sig)
test_signal.py 文件源码 项目:zippy 作者: securesystemslab 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def test_issue9324(self):
        # Updated for issue #10003, adding SIGBREAK
        handler = lambda x, y: None
        for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
                    signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
                    signal.SIGTERM):
            # Set and then reset a handler for signals that work on windows
            signal.signal(sig, signal.signal(sig, handler))

        with self.assertRaises(ValueError):
            signal.signal(-1, handler)

        with self.assertRaises(ValueError):
            signal.signal(7, handler)
test_signal.py 文件源码 项目:oil 作者: oilshell 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0
def test_issue9324(self):
        # Updated for issue #10003, adding SIGBREAK
        handler = lambda x, y: None
        for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
                    signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
                    signal.SIGTERM):
            # Set and then reset a handler for signals that work on windows
            signal.signal(sig, signal.signal(sig, handler))

        with self.assertRaises(ValueError):
            signal.signal(-1, handler)

        with self.assertRaises(ValueError):
            signal.signal(7, handler)
test_signal.py 文件源码 项目:python2-tracer 作者: extremecoders-re 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0
def test_issue9324(self):
        # Updated for issue #10003, adding SIGBREAK
        handler = lambda x, y: None
        for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
                    signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
                    signal.SIGTERM):
            # Set and then reset a handler for signals that work on windows
            signal.signal(sig, signal.signal(sig, handler))

        with self.assertRaises(ValueError):
            signal.signal(-1, handler)

        with self.assertRaises(ValueError):
            signal.signal(7, handler)
utils.py 文件源码 项目:treadmill 作者: Morgan-Stanley 项目源码 文件源码 阅读 15 收藏 0 点赞 0 评论 0
def term_signal():
    """Gets the term signal for the os."""
    if os.name == 'nt':
        return signal.SIGBREAK
    else:
        return signal.SIGTERM
popen_spawn.py 文件源码 项目:Repobot 作者: Desgard 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def kill(self, sig):
        '''Sends a Unix signal to the subprocess.

        Use constants from the :mod:`signal` module to specify which signal.
        '''
        if sys.platform == 'win32':
            if sig in [signal.SIGINT, signal.CTRL_C_EVENT]:
                sig = signal.CTRL_C_EVENT
            elif sig in [signal.SIGBREAK, signal.CTRL_BREAK_EVENT]:
                sig = signal.CTRL_BREAK_EVENT
            else:
                sig = signal.SIGTERM

        os.kill(self.proc.pid, sig)
test_signal.py 文件源码 项目:pefile.pypy 作者: cloudtracer 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def test_issue9324(self):
        # Updated for issue #10003, adding SIGBREAK
        handler = lambda x, y: None
        for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
                    signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
                    signal.SIGTERM):
            # Set and then reset a handler for signals that work on windows
            signal.signal(sig, signal.signal(sig, handler))

        with self.assertRaises(ValueError):
            signal.signal(-1, handler)

        with self.assertRaises(ValueError):
            signal.signal(7, handler)
popen_spawn.py 文件源码 项目:pipenv 作者: pypa 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
def kill(self, sig):
        '''Sends a Unix signal to the subprocess.

        Use constants from the :mod:`signal` module to specify which signal.
        '''
        if sys.platform == 'win32':
            if sig in [signal.SIGINT, signal.CTRL_C_EVENT]:
                sig = signal.CTRL_C_EVENT
            elif sig in [signal.SIGBREAK, signal.CTRL_BREAK_EVENT]:
                sig = signal.CTRL_BREAK_EVENT
            else:
                sig = signal.SIGTERM

        os.kill(self.proc.pid, sig)
test_signal.py 文件源码 项目:ndk-python 作者: gittor 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def test_issue9324(self):
        # Updated for issue #10003, adding SIGBREAK
        handler = lambda x, y: None
        for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
                    signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
                    signal.SIGTERM):
            # Set and then reset a handler for signals that work on windows
            signal.signal(sig, signal.signal(sig, handler))

        with self.assertRaises(ValueError):
            signal.signal(-1, handler)

        with self.assertRaises(ValueError):
            signal.signal(7, handler)
popen_spawn.py 文件源码 项目:yatta_reader 作者: sound88 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
def kill(self, sig):
        '''Sends a Unix signal to the subprocess.

        Use constants from the :mod:`signal` module to specify which signal.
        '''
        if sys.platform == 'win32':
            if sig in [signal.SIGINT, signal.CTRL_C_EVENT]:
                sig = signal.CTRL_C_EVENT
            elif sig in [signal.SIGBREAK, signal.CTRL_BREAK_EVENT]:
                sig = signal.CTRL_BREAK_EVENT
            else:
                sig = signal.SIGTERM

        os.kill(self.proc.pid, sig)
misc_test.py 文件源码 项目:cxflow 作者: Cognexa 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def kill_windows(pid, signum):
        """
        Adapt os.kill for Windows platform.

        Reference: <https://stackoverflow.com/questions/35772001/how-to-handle-the-signal-in-python-on-windows-machine>
        """
        sigmap = {signal.SIGINT: signal.CTRL_C_EVENT, signal.SIGBREAK: signal.CTRL_BREAK_EVENT}
        if signum in sigmap and pid == os.getpid():
            # we don't know if the current process is a
            # process group leader, so just broadcast
            # to all processes attached to this console.
            pid = 0
        thread = threading.current_thread()
        handler = signal.getsignal(signum)
        # work around the synchronization problem when calling
        # kill from the main thread.
        if (signum in sigmap and
            thread.name == 'MainThread' and
            callable(handler) and
            pid == 0):
            event = threading.Event()

            def handler_set_event(signum, frame):
                event.set()
                return handler(signum, frame)
            signal.signal(signum, handler_set_event)
            try:
                os.kill(pid, sigmap[signum])
                # busy wait because we can't block in the main
                # thread, else the signal handler can't execute.
                while not event.is_set():
                    pass
            finally:
                signal.signal(signum, handler)
        else:
            os.kill(pid, sigmap.get(signum, signum))
external_search_command.py 文件源码 项目:mongodb-monitoring 作者: jruaux 项目源码 文件源码 阅读 27 收藏 0 点赞 0 评论 0
def _execute(path, argv=None, environ=None):
            """ Executes an external search command.

            :param path: Path to the external search command.
            :type path: unicode

            :param argv: Argument list.
            :type argv: list or tuple
            The arguments to the child process should start with the name of the command being run, but this is not
            enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.

            :param environ: A mapping which is used to define the environment variables for the new process.
            :type environ: dict or None.
            This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
            the :data:`os.environ` mapping should be used.

            :return: None

            """
            search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
            found = ExternalSearchCommand._search_path(path, search_path)

            if found is None:
                raise ValueError('Cannot find command on path: {}'.format(path))

            path = found
            logger.debug('starting command="%s", arguments=%s', path, argv)

            def terminate(signal_number, frame):
                sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))

            def terminate_child():
                if p.pid is not None and p.returncode is None:
                    logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
                    os.kill(p.pid, CTRL_BREAK_EVENT)

            p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
            atexit.register(terminate_child)
            signal(SIGBREAK, terminate)
            signal(SIGINT, terminate)
            signal(SIGTERM, terminate)

            logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
            p.wait()

            logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
            sys.exit(p.returncode)
external_search_command.py 文件源码 项目:Splunk_CBER_App 作者: MHaggis 项目源码 文件源码 阅读 26 收藏 0 点赞 0 评论 0
def _execute(path, argv=None, environ=None):
            """ Executes an external search command.

            :param path: Path to the external search command.
            :type path: unicode

            :param argv: Argument list.
            :type argv: list or tuple
            The arguments to the child process should start with the name of the command being run, but this is not
            enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.

            :param environ: A mapping which is used to define the environment variables for the new process.
            :type environ: dict or None.
            This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
            the :data:`os.environ` mapping should be used.

            :return: None

            """
            search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
            found = ExternalSearchCommand._search_path(path, search_path)

            if found is None:
                raise ValueError('Cannot find command on path: {}'.format(path))

            path = found
            logger.debug('starting command="%s", arguments=%s', path, argv)

            def terminate(signal_number, frame):
                sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))

            def terminate_child():
                if p.pid is not None and p.returncode is None:
                    logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
                    os.kill(p.pid, CTRL_BREAK_EVENT)

            p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
            atexit.register(terminate_child)
            signal(SIGBREAK, terminate)
            signal(SIGINT, terminate)
            signal(SIGTERM, terminate)

            logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
            p.wait()

            logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
            sys.exit(p.returncode)
external_search_command.py 文件源码 项目:super_simple_siem 作者: elucidant 项目源码 文件源码 阅读 36 收藏 0 点赞 0 评论 0
def _execute(path, argv=None, environ=None):
            """ Executes an external search command.

            :param path: Path to the external search command.
            :type path: unicode

            :param argv: Argument list.
            :type argv: list or tuple
            The arguments to the child process should start with the name of the command being run, but this is not
            enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.

            :param environ: A mapping which is used to define the environment variables for the new process.
            :type environ: dict or None.
            This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
            the :data:`os.environ` mapping should be used.

            :return: None

            """
            search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
            found = ExternalSearchCommand._search_path(path, search_path)

            if found is None:
                raise ValueError('Cannot find command on path: {}'.format(path))

            path = found
            logger.debug('starting command="%s", arguments=%s', path, argv)

            def terminate(signal_number, frame):
                sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))

            def terminate_child():
                if p.pid is not None and p.returncode is None:
                    logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
                    os.kill(p.pid, CTRL_BREAK_EVENT)

            p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
            atexit.register(terminate_child)
            signal(SIGBREAK, terminate)
            signal(SIGINT, terminate)
            signal(SIGTERM, terminate)

            logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
            p.wait()

            logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)

            if p.returncode != 0:
                sys.exit(p.returncode)
external_search_command.py 文件源码 项目:TA-connectivity 作者: seunomosowon 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def _execute(path, argv=None, environ=None):
            """ Executes an external search command.

            :param path: Path to the external search command.
            :type path: unicode

            :param argv: Argument list.
            :type argv: list or tuple
            The arguments to the child process should start with the name of the command being run, but this is not
            enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.

            :param environ: A mapping which is used to define the environment variables for the new process.
            :type environ: dict or None.
            This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
            the :data:`os.environ` mapping should be used.

            :return: None

            """
            search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
            found = ExternalSearchCommand._search_path(path, search_path)

            if found is None:
                raise ValueError('Cannot find command on path: {}'.format(path))

            path = found
            logger.debug('starting command="%s", arguments=%s', path, argv)

            def terminate(signal_number, frame):
                sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))

            def terminate_child():
                if p.pid is not None and p.returncode is None:
                    logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
                    os.kill(p.pid, CTRL_BREAK_EVENT)

            p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
            atexit.register(terminate_child)
            signal(SIGBREAK, terminate)
            signal(SIGINT, terminate)
            signal(SIGTERM, terminate)

            logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
            p.wait()

            logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
            sys.exit(p.returncode)
external_search_command.py 文件源码 项目:splunk_ta_ps4_f1_2016 作者: jonathanvarley 项目源码 文件源码 阅读 26 收藏 0 点赞 0 评论 0
def _execute(path, argv=None, environ=None):
            """ Executes an external search command.

            :param path: Path to the external search command.
            :type path: unicode

            :param argv: Argument list.
            :type argv: list or tuple
            The arguments to the child process should start with the name of the command being run, but this is not
            enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.

            :param environ: A mapping which is used to define the environment variables for the new process.
            :type environ: dict or None.
            This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
            the :data:`os.environ` mapping should be used.

            :return: None

            """
            search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
            found = ExternalSearchCommand._search_path(path, search_path)

            if found is None:
                raise ValueError('Cannot find command on path: {}'.format(path))

            path = found
            logger.debug('starting command="%s", arguments=%s', path, argv)

            def terminate(signal_number, frame):
                sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))

            def terminate_child():
                if p.pid is not None and p.returncode is None:
                    logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
                    os.kill(p.pid, CTRL_BREAK_EVENT)

            p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
            atexit.register(terminate_child)
            signal(SIGBREAK, terminate)
            signal(SIGINT, terminate)
            signal(SIGTERM, terminate)

            logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
            p.wait()

            logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
            sys.exit(p.returncode)
external_search_command.py 文件源码 项目:splunk_ta_ps4_f1_2016 作者: jonathanvarley 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def _execute(path, argv=None, environ=None):
            """ Executes an external search command.

            :param path: Path to the external search command.
            :type path: unicode

            :param argv: Argument list.
            :type argv: list or tuple
            The arguments to the child process should start with the name of the command being run, but this is not
            enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.

            :param environ: A mapping which is used to define the environment variables for the new process.
            :type environ: dict or None.
            This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
            the :data:`os.environ` mapping should be used.

            :return: None

            """
            search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
            found = ExternalSearchCommand._search_path(path, search_path)

            if found is None:
                raise ValueError('Cannot find command on path: {}'.format(path))

            path = found
            logger.debug('starting command="%s", arguments=%s', path, argv)

            def terminate(signal_number, frame):
                sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))

            def terminate_child():
                if p.pid is not None and p.returncode is None:
                    logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
                    os.kill(p.pid, CTRL_BREAK_EVENT)

            p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
            atexit.register(terminate_child)
            signal(SIGBREAK, terminate)
            signal(SIGINT, terminate)
            signal(SIGTERM, terminate)

            logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
            p.wait()

            logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
            sys.exit(p.returncode)
external_search_command.py 文件源码 项目:TA-SyncKVStore 作者: georgestarcher 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def _execute(path, argv=None, environ=None):
            """ Executes an external search command.

            :param path: Path to the external search command.
            :type path: unicode

            :param argv: Argument list.
            :type argv: list or tuple
            The arguments to the child process should start with the name of the command being run, but this is not
            enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.

            :param environ: A mapping which is used to define the environment variables for the new process.
            :type environ: dict or None.
            This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
            the :data:`os.environ` mapping should be used.

            :return: None

            """
            search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
            found = ExternalSearchCommand._search_path(path, search_path)

            if found is None:
                raise ValueError('Cannot find command on path: {}'.format(path))

            path = found
            logger.debug('starting command="%s", arguments=%s', path, argv)

            def terminate(signal_number, frame):
                sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))

            def terminate_child():
                if p.pid is not None and p.returncode is None:
                    logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
                    os.kill(p.pid, CTRL_BREAK_EVENT)

            p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
            atexit.register(terminate_child)
            signal(SIGBREAK, terminate)
            signal(SIGINT, terminate)
            signal(SIGTERM, terminate)

            logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
            p.wait()

            logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
            sys.exit(p.returncode)
external_search_command.py 文件源码 项目:TA-SyncKVStore 作者: georgestarcher 项目源码 文件源码 阅读 27 收藏 0 点赞 0 评论 0
def _execute(path, argv=None, environ=None):
            """ Executes an external search command.

            :param path: Path to the external search command.
            :type path: unicode

            :param argv: Argument list.
            :type argv: list or tuple
            The arguments to the child process should start with the name of the command being run, but this is not
            enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.

            :param environ: A mapping which is used to define the environment variables for the new process.
            :type environ: dict or None.
            This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
            the :data:`os.environ` mapping should be used.

            :return: None

            """
            search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
            found = ExternalSearchCommand._search_path(path, search_path)

            if found is None:
                raise ValueError('Cannot find command on path: {}'.format(path))

            path = found
            logger.debug('starting command="%s", arguments=%s', path, argv)

            def terminate(signal_number, frame):
                sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))

            def terminate_child():
                if p.pid is not None and p.returncode is None:
                    logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
                    os.kill(p.pid, CTRL_BREAK_EVENT)

            p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
            atexit.register(terminate_child)
            signal(SIGBREAK, terminate)
            signal(SIGINT, terminate)
            signal(SIGTERM, terminate)

            logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
            p.wait()

            logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
            sys.exit(p.returncode)
external_search_command.py 文件源码 项目:cb-defense-splunk-app 作者: carbonblack 项目源码 文件源码 阅读 31 收藏 0 点赞 0 评论 0
def _execute(path, argv=None, environ=None):
            """ Executes an external search command.

            :param path: Path to the external search command.
            :type path: unicode

            :param argv: Argument list.
            :type argv: list or tuple
            The arguments to the child process should start with the name of the command being run, but this is not
            enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.

            :param environ: A mapping which is used to define the environment variables for the new process.
            :type environ: dict or None.
            This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
            the :data:`os.environ` mapping should be used.

            :return: None

            """
            search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
            found = ExternalSearchCommand._search_path(path, search_path)

            if found is None:
                raise ValueError('Cannot find command on path: {}'.format(path))

            path = found
            logger.debug('starting command="%s", arguments=%s', path, argv)

            def terminate(signal_number, frame):
                sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))

            def terminate_child():
                if p.pid is not None and p.returncode is None:
                    logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
                    os.kill(p.pid, CTRL_BREAK_EVENT)

            p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
            atexit.register(terminate_child)
            signal(SIGBREAK, terminate)
            signal(SIGINT, terminate)
            signal(SIGTERM, terminate)

            logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
            p.wait()

            logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
            sys.exit(p.returncode)
external_search_command.py 文件源码 项目:cb-defense-splunk-app 作者: carbonblack 项目源码 文件源码 阅读 33 收藏 0 点赞 0 评论 0
def _execute(path, argv=None, environ=None):
            """ Executes an external search command.

            :param path: Path to the external search command.
            :type path: unicode

            :param argv: Argument list.
            :type argv: list or tuple
            The arguments to the child process should start with the name of the command being run, but this is not
            enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.

            :param environ: A mapping which is used to define the environment variables for the new process.
            :type environ: dict or None.
            This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
            the :data:`os.environ` mapping should be used.

            :return: None

            """
            search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
            found = ExternalSearchCommand._search_path(path, search_path)

            if found is None:
                raise ValueError('Cannot find command on path: {}'.format(path))

            path = found
            logger.debug('starting command="%s", arguments=%s', path, argv)

            def terminate(signal_number, frame):
                sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))

            def terminate_child():
                if p.pid is not None and p.returncode is None:
                    logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
                    os.kill(p.pid, CTRL_BREAK_EVENT)

            p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
            atexit.register(terminate_child)
            signal(SIGBREAK, terminate)
            signal(SIGINT, terminate)
            signal(SIGTERM, terminate)

            logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
            p.wait()

            logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
            sys.exit(p.returncode)
external_search_command.py 文件源码 项目:TA-statemachine 作者: doksu 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def _execute(path, argv=None, environ=None):
            """ Executes an external search command.

            :param path: Path to the external search command.
            :type path: unicode

            :param argv: Argument list.
            :type argv: list or tuple
            The arguments to the child process should start with the name of the command being run, but this is not
            enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.

            :param environ: A mapping which is used to define the environment variables for the new process.
            :type environ: dict or None.
            This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
            the :data:`os.environ` mapping should be used.

            :return: None

            """
            search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
            found = ExternalSearchCommand._search_path(path, search_path)

            if found is None:
                raise ValueError('Cannot find command on path: {}'.format(path))

            path = found
            logger.debug('starting command="%s", arguments=%s', path, argv)

            def terminate(signal_number, frame):
                sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))

            def terminate_child():
                if p.pid is not None and p.returncode is None:
                    logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
                    os.kill(p.pid, CTRL_BREAK_EVENT)

            p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
            atexit.register(terminate_child)
            signal(SIGBREAK, terminate)
            signal(SIGINT, terminate)
            signal(SIGTERM, terminate)

            logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
            p.wait()

            logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)

            if p.returncode != 0:
                sys.exit(p.returncode)
external_search_command.py 文件源码 项目:mysplunk_csc 作者: patel-bhavin 项目源码 文件源码 阅读 43 收藏 0 点赞 0 评论 0
def _execute(path, argv=None, environ=None):
            """ Executes an external search command.

            :param path: Path to the external search command.
            :type path: unicode

            :param argv: Argument list.
            :type argv: list or tuple
            The arguments to the child process should start with the name of the command being run, but this is not
            enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.

            :param environ: A mapping which is used to define the environment variables for the new process.
            :type environ: dict or None.
            This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
            the :data:`os.environ` mapping should be used.

            :return: None

            """
            search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
            found = ExternalSearchCommand._search_path(path, search_path)

            if found is None:
                raise ValueError('Cannot find command on path: {}'.format(path))

            path = found
            logger.debug('starting command="%s", arguments=%s', path, argv)

            def terminate(signal_number, frame):
                sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))

            def terminate_child():
                if p.pid is not None and p.returncode is None:
                    logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
                    os.kill(p.pid, CTRL_BREAK_EVENT)

            p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
            atexit.register(terminate_child)
            signal(SIGBREAK, terminate)
            signal(SIGINT, terminate)
            signal(SIGTERM, terminate)

            logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
            p.wait()

            logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
            sys.exit(p.returncode)
external_search_command.py 文件源码 项目:SplunkForPCAP 作者: DanielSchwartz1 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def _execute(path, argv=None, environ=None):
            """ Executes an external search command.

            :param path: Path to the external search command.
            :type path: unicode

            :param argv: Argument list.
            :type argv: list or tuple
            The arguments to the child process should start with the name of the command being run, but this is not
            enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.

            :param environ: A mapping which is used to define the environment variables for the new process.
            :type environ: dict or None.
            This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
            the :data:`os.environ` mapping should be used.

            :return: None

            """
            search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
            found = ExternalSearchCommand._search_path(path, search_path)

            if found is None:
                raise ValueError('Cannot find command on path: {}'.format(path))

            path = found
            logger.debug('starting command="%s", arguments=%s', path, argv)

            def terminate(signal_number, frame):
                sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))

            def terminate_child():
                if p.pid is not None and p.returncode is None:
                    logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
                    os.kill(p.pid, CTRL_BREAK_EVENT)

            p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
            atexit.register(terminate_child)
            signal(SIGBREAK, terminate)
            signal(SIGINT, terminate)
            signal(SIGTERM, terminate)

            logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
            p.wait()

            logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
            sys.exit(p.returncode)
external_search_command.py 文件源码 项目:elasticsplunk 作者: brunotm 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def _execute(path, argv=None, environ=None):
            """ Executes an external search command.

            :param path: Path to the external search command.
            :type path: unicode

            :param argv: Argument list.
            :type argv: list or tuple
            The arguments to the child process should start with the name of the command being run, but this is not
            enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.

            :param environ: A mapping which is used to define the environment variables for the new process.
            :type environ: dict or None.
            This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
            the :data:`os.environ` mapping should be used.

            :return: None

            """
            search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
            found = ExternalSearchCommand._search_path(path, search_path)

            if found is None:
                raise ValueError('Cannot find command on path: {}'.format(path))

            path = found
            logger.debug('starting command="%s", arguments=%s', path, argv)

            def terminate(signal_number, frame):
                sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))

            def terminate_child():
                if p.pid is not None and p.returncode is None:
                    logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
                    os.kill(p.pid, CTRL_BREAK_EVENT)

            p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
            atexit.register(terminate_child)
            signal(SIGBREAK, terminate)
            signal(SIGINT, terminate)
            signal(SIGTERM, terminate)

            logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
            p.wait()

            logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)

            if p.returncode != 0:
                sys.exit(p.returncode)


问题


面经


文章

微信
公众号

扫码关注公众号