queue.test package

Submodules

queue.test.test_job_load module

class queue.test.test_job_load.TestJobLoadProcessor(methodName='runTest')

Bases: django.test.testcases.TestCase

setUp()
test_process_event()

This method tests the Job Load Processor

queue.test.test_models module

class queue.test.test_models.TestJobLoadManager(methodName='runTest')

Bases: django.test.testcases.TestCase

setUp()
test_calculate_empty()

Tests calculating job load when there are no jobs.

test_calculate_job_type()

Tests calculating job load grouping by job type.

test_calculate_status()

Tests calculating job load filtering by status.

class queue.test.test_models.TestQueueManager(methodName='runTest')

Bases: django.test.testcases.TransactionTestCase

fixtures = [u'basic_errors.json']
setUp()
test_get_queue_fifo()

Tests calling QueueManager.get_queue() in FIFO mode

test_get_queue_lifo()

Tests calling QueueManager.get_queue() in LIFO mode

class queue.test.test_models.TestQueueManagerQueueNewJob(methodName='runTest')

Bases: django.test.testcases.TransactionTestCase

setUp()
test_successful(*args, **keywargs)

Tests calling QueueManager.queue_new_job_v6() successfully with a Seed job type

class queue.test.test_models.TestQueueManagerQueueNewRecipe(methodName='runTest')

Bases: django.test.testcases.TransactionTestCase

fixtures = [u'basic_system_job_types.json', u'ingest_job_types.json']
setUp()
test_inactive(*args, **keywargs)
test_successful(*args, **keywargs)
test_successful_ingest(*args, **keywargs)
class queue.test.test_models.TestQueueManagerRequeueJobs(methodName='runTest')

Bases: django.test.testcases.TransactionTestCase

setUp()
test_successful(*args, **keywargs)

Tests calling QueueManager.requeue_jobs() successfully

queue.test.test_views module

class queue.test.test_views.TestJobLoadView(methodName='runTest')

Bases: rest_framework.test.APITransactionTestCase

setUp()
test_job_type_id()

Tests successfully calling the job laod view filtered by job type identifier.

test_job_type_name()

Tests successfully calling the job load view filtered by job type name.

test_max_duration()

Tests calling the job load view with time values that define a range greater than 31 days

test_successful()

Tests successfully calling the job load view.

class queue.test.test_views.TestQueueStatusView(methodName='runTest')

Bases: rest_framework.test.APITransactionTestCase

setUp()
test_successful()

Tests successfully calling the queue status view.

queue.test.utils module

Defines utility methods for testing queues

queue.test.utils.create_job_load(job_type=None, measured=None, pending_count=0, queued_count=0, running_count=0)

Creates a job load model for unit testing

Returns:The job load model
Return type:queue.models.JobLoad
queue.test.utils.create_queue(job_type=None, priority=1, timeout=3600, resources=None, cpus_required=1.0, mem_required=512.0, disk_in_required=200.0, disk_out_required=100.0, disk_total_required=300.0, gpus_required=0, queued=datetime.datetime(2020, 8, 18, 12, 11, 59, 466209, tzinfo=<UTC>))

Creates a queue model for unit testing

Parameters:
  • job_type (job.models.JobType) – The job type
  • priority (int) – The priority
  • timeout (int) – The timeout
  • resources_required (node.resources.node_resources.NodeResources) – The required resources
  • cpus_required (float) – The number of CPUs required
  • mem_required (float) – The memory required in MiB
  • disk_in_required (float) – The input disk space required in MiB
  • disk_out_required (float) – The output disk space required in MiB
  • disk_total_required (float) – The total disk space required in MiB
  • gpus_required (float) – The number of GPUs required
  • queued (datetime.datetime) – The time the execution was queued

Module contents