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 updating state in the mock queue and running the UWS database worker functions that normally would be run automatically by the queue.

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