Function assert_container_exited

t/smoke/tests/test_worker.py:15–25

It might take a few moments for the container to exit after the worker is killed.

(worker: CeleryTestWorker, attempts: int = RESULT_TIMEOUT)

def assert_container_exited(worker: CeleryTestWorker, attempts: int = RESULT_TIMEOUT):
    """It might take a few moments for the container to exit after the worker is killed."""
    while attempts:
        worker.container.reload()
        if worker.container.status == "exited":
            break
        attempts -= 1
        sleep(1)

    worker.container.reload()
    assert worker.container.status == "exited"


@pytest.mark.parametrize("method", list(WorkerRestart.Method))