def test_cancel(make_job, exec_ctx):
minimal_job = make_job(sched_access=exec_ctx.access)
prepare_job(minimal_job, "sleep 30")
t_job = datetime.now()
minimal_job.submit()
minimal_job.cancel()
// We give some time to the local scheduler for the TERM signal to be
// delivered; if we poll immediately, the process may have not been killed
// yet, and the scheduler will assume that it"s ignoring its signal, then
// wait for a grace period and send a KILL signal, which is not what we
// want to test here.
time.sleep(0.01)
minimal_job.wait()
t_job = datetime.now() - t_job
assert minimal_job.finished()
assert t_job.total_seconds() < 30
After Change
def test_cancel(make_job, exec_ctx):
minimal_job = make_job(sched_access=exec_ctx.access)
prepare_job(minimal_job, "sleep 30")
t_job = time.time()
minimal_job.submit()
minimal_job.cancel()
// We give some time to the local scheduler for the TERM signal to be
// delivered; if we poll immediately, the process may have not been killed
// yet, and the scheduler will assume that it"s ignoring its signal, then
// wait for a grace period and send a KILL signal, which is not what we
// want to test here.
time.sleep(0.01)
minimal_job.wait()
t_job = time.time() - t_job
assert minimal_job.finished()
assert t_job < 30
