MockUWSJobRunner

class safir.testing.uws.MockUWSJobRunner(config, arq_queue)

Bases: object

Simulate execution of jobs with a mock queue.

When running the test suite, the arq queue is replaced with a mock queue that doesn’t execute workers. That execution has to be simulated by manually invoking the backend worker.

Parameters:

• config (UWSConfig) – UWS configuration.

• arq_queue (MockArqQueue) – Mock arq queue for testing.

Methods Summary

get_job_metadata(token, job_id)	Get the arq job metadata for a job.
get_job_result(token, job_id)	Get the arq job result for a job.
mark_complete(token, job_id, results, *[, delay])	Mark an in progress job as complete.
mark_in_progress(token, job_id, *[, delay])	Mark a queued job in progress.

Methods Documentation

async get_job_metadata(token, job_id)

Get the arq job metadata for a job.

Parameters:

• token (str) – Token for the user.

• job_id (str) – UWS job ID.

Returns:

arq job metadata.

Return type:

JobMetadata

async get_job_result(token, job_id)

Get the arq job result for a job.

Parameters:

• token (str) – Token for the user.

• job_id (str) – UWS job ID.

Returns:

arq job metadata.

Return type:

JobMetadata

async mark_complete(token, job_id, results, *, delay=None)

Mark an in progress job as complete.

Parameters:

• token (str) – Token for the user.

• job_id (str) – Job ID.

• results (list[WorkerResult] | Exception) – Results to return. May be an exception to simulate a job failure.

• delay (float | None, default: None) – How long to delay in seconds before marking the job as complete.

Returns:

Record of the job.

Return type:

Job

async mark_in_progress(token, job_id, *, delay=None)

Mark a queued job in progress.

Parameters:

• token (str) – Token for the user.

• job_id (str) – Job ID.

• delay (float | None, default: None) – How long to delay in seconds before marking the job as complete.

Returns:

Record of the job.

Return type:

Job