Patching a function/task used in a task

[pai-martinreidemeister]

Hello,

I want to write a unit test, that tests if my workflow handles exceptions in a task.

This is a simplified version of my setup:

from my_proj.celery import celery_app

LOGGER=get_task_logger(__name__)

@celery_app.task
def stage_1() -> None:
  LOGGER.info("run stage 1")

@celery_app.task
def stage_2() -> None:
  LOGGER.info("run stage 2")

@celery_app.task
def finish_task() -> None: ...

@celery_app.task
def failure_task(request: Request, exc: BaseException, _tb: TracebackType | None) -> None: ...

def get_stages() -> list[Signature]:
  return [stage_1.si(), stage_2.si()]


@celery_app.task
def trigger_workflow(self):
  stages = get_stages()

  chain(*stages, finish_task.si()).apply_async(link_error=failure_task.s())

I have no idea how to patch a task from get_stagess (e.g. stage_1) so that it raises an exception, so I can see the results of failure_task.