test_process_executor.py 文件源码

def test_process_termination(self, executor, model, fs_test_holder, tmpdir):
        freeze_script_path = str(tmpdir.join('freeze_script'))
        with open(freeze_script_path, 'w+b') as f:
            f.write(
                '''import time
while True:
    time.sleep(5)
                '''
            )
        holder_path_argument = models.Argument.wrap('holder_path', fs_test_holder._path)
        script_path_argument = models.Argument.wrap('freezing_script_path',
                                                    str(tmpdir.join('freeze_script')))

        model.argument.put(holder_path_argument)
        model.argument.put(script_path_argument)
        ctx = MockContext(
            task_kwargs=dict(
                function='{0}.{1}'.format(__name__, freezing_task.__name__),
                arguments=dict(holder_path=holder_path_argument,
                               freezing_script_path=script_path_argument)),
        )

        executor.execute(ctx)

        @retrying.retry(retry_on_result=lambda r: r is False, stop_max_delay=60000, wait_fixed=500)
        def wait_for_extra_process_id():
            return fs_test_holder.get('subproc', False)

        task_pid = executor._tasks[ctx.task.id].proc.pid
        extra_process_pid = wait_for_extra_process_id()

        assert set([task_pid, extra_process_pid]).issubset(set(psutil.pids()))
        executor.terminate(ctx.task.id)

        # Give a chance to the processes to terminate
        time.sleep(2)

        # all processes should be either zombies or non existent
        pids = [task_pid, extra_process_pid]
        for pid in pids:
            if pid in psutil.pids():
                assert psutil.Process(pid).status() == psutil.STATUS_ZOMBIE
            else:
                # making the test more readable
                assert pid not in psutil.pids()