Function run_worker

t/integration/test_quorum_queue_qos_cluster_simulation.py:56–81

(simulate_qos_issue: bool, result_queue: multiprocessing.Queue)

def run_worker(simulate_qos_issue: bool, result_queue: multiprocessing.Queue):
    queue_name = f"race_quorum_queue_{uuid.uuid4().hex}"
    app = create_app(queue_name)
    logger.info("[Celery config snapshot]:\n%s", pprint.pformat(dict(app.conf)))
    task = dummy_task_factory(app, simulate_qos_issue)

    try:
        with start_worker(
            app,
            queues=[queue_name],
            loglevel="INFO",
            perform_ping_check=False,
            shutdown_timeout=15,
        ):
            res = task.delay()
            try:
                result = res.get(timeout=10)
                result_queue.put({"status": "ok", "result": result})
            except Exception as e:
                result_queue.put({"status": "error", "reason": str(e)})
    except Exception as e:
        logger.exception("[worker %s] external failure", simulate_qos_issue)
        result_queue.put({"status": "external_failure", "reason": str(e)})
    finally:
        if result_queue.empty():
            result_queue.put({"status": "crash", "reason": "Worker crashed without reporting"})


@pytest.mark.amqp