Function check_task_processes_exited

tests/test_task_cleanup.py:161–197

Creates a task that spawns a child, and then calls `stop_fn`, which should kill either the task or the worker. The function then checks that both the task process and its child have been killed.

(hq_env: HqEnv, stop_fn: Callable[[subprocess.Popen], None], terminates_worker=True)

def check_task_processes_exited(hq_env: HqEnv, stop_fn: Callable[[subprocess.Popen], None], terminates_worker=True):
    """
    Creates a task that spawns a child, and then calls `stop_fn`, which should kill either the task
    or the worker. The function then checks that both the task process and its child have been killed.
    """
    hq_env.start_server()
    worker_process = hq_env.start_worker()

    hq_env.command(
        [
            "submit",
            "--",
            *python(
                """
import os
import sys
import time
print(os.getpid(), flush=True)
pid = os.fork()
if pid > 0:
    print(pid, flush=True)
time.sleep(3600)
"""
            ),
        ]
    )
    wait_for_job_state(hq_env, 1, "RUNNING")
    wait_until(lambda: len(read_file(default_task_output()).splitlines()) == 2)
    pids = [int(pid) for pid in read_file(default_task_output()).splitlines()]

    stop_fn(worker_process)

    parent, child = pids
    wait_for_pid_exit(parent)
    wait_for_pid_exit(child)
    if terminates_worker:
        hq_env.check_process_exited(worker_process, None)