Rooba
/
rq
mirror of https://github.com/rq/rq.git
1
1
Fork 0
Code Issues Packages Projects Releases Wiki Activity
rq/tests/test_group.py

145 lines
5.8 KiB
Python
Raw Normal View History

Group jobs into batches and retrieve by batch name (#1945) * Batch.py initial commit * Add method to refresh batch status * Clean up Redis key properties * Remove persist_job method * Execute refresh method when fetching batch * Add batch_id to job * Add option to batch jobs with enqueue_many * Add set_batch_id method * Handle batch_ttl when job is started * Renew ttl on all jobs in batch when a job finishes * Use fetch_jobs method in refresh() * Remove batch TTL During worker maintenance task, worker will loop through batches and delete expired jobs. When all jobs are expired, the batch is deleted. * Fix buggy connection * Raise batch error in fetch method * Fix fetch connection arg * Add testing for batch * Remove unused import * Update batch documentation * Add batch ID to job persistence test * Test that empty batch is removed from batch list * Make default batch ID full uuid4 to match Job * Add method to return batch key * Add all method to return iterable of batches * Revert changes to queue.py * Separate batch creating from job enqueueing. Batches can be created by passing an array of jobs. * Fix bug with stopped callback in enqueue_many * Rename delete_expired_jobs batch method * Use exists to identify expired jobs * Remove jobs attribute from batch Jobs have to be explicitly fetched using the get_jobs method * Don't add batch_id to job in _add_jobs method This should be done when the job is created before it is saved to redis. * Use cleanup method to determine if batch should be deleted * Add get_jobs method This method will return an array of jobs belonging to the batch. * Update create method to not allow jobs arg Jobs should be added to batches in the Queue.enqueue_many function * Add batch_id to job create method * Delete set_batch_id method The batch ID should be set atomically when the job is created * Add batch argument to queue.enqueue_many This will be how jobs can be batched. The batch ID is added to each job created by enqueue_many, and then those jobs are all added to the batch in Redis using the batch's _add_jobs method * Update batch tests to use new enqueue_many arg * Fix batch_id for non_batched jobs * Use different pipeline name * Don't use fetch method when enqueuing jobs If the batch is empty, fetch will delete it from Redis * Test enqueuing batched jobs with string ID * Update batch documentation * Remove unused variables * Fix expired job tracking bug * Add delete_job method * Use delete_job method on batch * Pass multiple jobs into srem command * Add multiple keys to exists call * Pipeline _add_jobs call * Fix missing job_id arg * Move clean_batch_registry to Batch class * Remove unused variables * Fix missing dependency deletion I accidentally deleted this earlier * Move batch enqueueing to batch class * Update batch docs * Add test for batched jobs with str queue arg * Don't delete batch in cleanup * Change Batch to Group * Import EnqueueData for type hint * Sort imports * Remove local config file * Rename clean_group_registries method * Update feature version * Fix group constant names * Only allow Queue in enqueue many * Move cleanup logic to classmethod * Remove str group test * Remove unused import * Make pipeline mandatory in _add_jobs method * Rename expired job ids array * Add slow decorator * Assert job 1 data in group * Rename id arg to name * Only run cleanup when jobs are retrieved * Remove job_ids variable * Fix group name arg * Fix group name arg * Clean groups before fetching * Use uuid4().hex for shorter id * Move cleanup logic to instance method * Use all method when cleaning group registry * Check if group exists instead of catching error * Cleanup expired jobs * Apply black formatting * Fix black formatting * Pipeline group cleanup * Fix pipeline call to exists * Add pyenv version file * Remove group cleanup after job completes * Use existing pipeline * Remove unneeded pipeline check * Fix empty call * Test group __repr__ method * Remove unnecessary pipeline assignment * Remove unused delete method * Fix pipeline name * Remove unnecessary conditional block
2024-03-11 08:08:53 +00:00
from time import sleep
import pytest
from rq import Queue, SimpleWorker
from rq.exceptions import NoSuchGroupError
from rq.group import Group
from rq.job import Job
from rq.utils import as_text
from tests import RQTestCase
from tests.fixtures import say_hello
class TestGroup(RQTestCase):
job_1_data = Queue.prepare_data(say_hello, job_id='job1')
job_2_data = Queue.prepare_data(say_hello, job_id='job2')
def test_create_group(self):
q = Queue(connection=self.testconn)
group = Group.create(connection=self.testconn)
group.enqueue_many(q, [self.job_1_data, self.job_2_data])
assert isinstance(group, Group)
assert len(group.get_jobs()) == 2
q.empty()
def test_group_repr(self):
group = Group.create(name="foo", connection=self.testconn)
assert group.__repr__() == "Group(id=foo)"
def test_group_jobs(self):
q = Queue(connection=self.testconn)
group = Group.create(connection=self.testconn)
jobs = group.enqueue_many(q, [self.job_1_data, self.job_2_data])
self.assertCountEqual(group.get_jobs(), jobs)
q.empty()
def test_fetch_group(self):
q = Queue(connection=self.testconn)
enqueued_group = Group.create(connection=self.testconn)
enqueued_group.enqueue_many(q, [self.job_1_data, self.job_2_data])
fetched_group = Group.fetch(enqueued_group.name, self.testconn)
self.assertCountEqual(enqueued_group.get_jobs(), fetched_group.get_jobs())
assert len(fetched_group.get_jobs()) == 2
q.empty()
def test_add_jobs(self):
q = Queue(connection=self.testconn)
group = Group.create(connection=self.testconn)
group.enqueue_many(q, [self.job_1_data, self.job_2_data])
job2 = group.enqueue_many(q, [self.job_1_data, self.job_2_data])[0]
assert job2 in group.get_jobs()
self.assertEqual(job2.group_id, group.name)
q.empty()
def test_jobs_added_to_group_key(self):
q = Queue(connection=self.testconn)
group = Group.create(connection=self.testconn)
jobs = group.enqueue_many(q, [self.job_1_data, self.job_2_data])
job_ids = [job.id for job in group.get_jobs()]
jobs = list({as_text(job) for job in self.testconn.smembers(group.key)})
self.assertCountEqual(jobs, job_ids)
q.empty()
def test_group_id_added_to_jobs(self):
q = Queue(connection=self.testconn)
group = Group.create(connection=self.testconn)
jobs = group.enqueue_many(q, [self.job_1_data])
assert jobs[0].group_id == group.name
fetched_job = Job.fetch(jobs[0].id, connection=self.testconn)
assert fetched_job.group_id == group.name
def test_deleted_jobs_removed_from_group(self):
q = Queue(connection=self.testconn)
group = Group.create(connection=self.testconn)
group.enqueue_many(q, [self.job_1_data, self.job_2_data])
job = group.get_jobs()[0]
job.delete()
group.cleanup()
redis_jobs = list({as_text(job) for job in self.testconn.smembers(group.key)})
assert job.id not in redis_jobs
assert job not in group.get_jobs()
def test_group_added_to_registry(self):
q = Queue(connection=self.testconn)
group = Group.create(connection=self.testconn)
group.enqueue_many(q, [self.job_1_data])
redis_groups = {as_text(group) for group in self.testconn.smembers("rq:groups")}
assert group.name in redis_groups
q.empty()
@pytest.mark.slow
def test_expired_jobs_removed_from_group(self):
q = Queue(connection=self.testconn)
w = SimpleWorker([q], connection=q.connection)
short_lived_job = Queue.prepare_data(say_hello, result_ttl=1)
group = Group.create(connection=self.testconn)
group.enqueue_many(q, [short_lived_job, self.job_1_data])
w.work(burst=True, max_jobs=1)
sleep(2)
w.run_maintenance_tasks()
group.cleanup()
assert len(group.get_jobs()) == 1
assert self.job_1_data.job_id in [job.id for job in group.get_jobs()]
q.empty()
@pytest.mark.slow
def test_empty_group_removed_from_group_list(self):
q = Queue(connection=self.testconn)
w = SimpleWorker([q], connection=q.connection)
short_lived_job = Queue.prepare_data(say_hello, result_ttl=1)
group = Group.create(connection=self.testconn)
group.enqueue_many(q, [short_lived_job])
w.work(burst=True, max_jobs=1)
sleep(2)
w.run_maintenance_tasks()
redis_groups = {as_text(group) for group in self.testconn.smembers("rq:groups")}
assert group.name not in redis_groups
@pytest.mark.slow
def test_fetch_expired_group_raises_error(self):
q = Queue(connection=self.testconn)
w = SimpleWorker([q], connection=q.connection)
short_lived_job = Queue.prepare_data(say_hello, result_ttl=1)
group = Group.create(connection=self.testconn)
group.enqueue_many(q, [short_lived_job])
w.work(burst=True, max_jobs=1)
sleep(2)
w.run_maintenance_tasks()
self.assertRaises(NoSuchGroupError, Group.fetch, group.name, group.connection)
q.empty()
def test_get_group_key(self):
group = Group(name="foo", connection=self.testconn)
self.assertEqual(Group.get_key(group.name), "rq:group:foo")
def test_all_returns_all_groups(self):
q = Queue(connection=self.testconn)
group1 = Group.create(name="group1", connection=self.testconn)
Group.create(name="group2", connection=self.testconn)
group1.enqueue_many(q, [self.job_1_data, self.job_2_data])
all_groups = Group.all(self.testconn)
assert len(all_groups) == 1
assert "group1" in [group.name for group in all_groups]
assert "group2" not in [group.name for group in all_groups]