Coverage for utilities/testing/mixins.py: 0%

1from django.test.testcases import SerializeMixin 

2from django_rq import get_queue 

3from django_rq.workers import get_worker 

4from rq import SimpleWorker 

5 

6__all__ = ( 

7 'RQQueueTestMixin', 

8) 

9 

10 

11class RQQueueTestMixin(SerializeMixin): 

12 """ 

13 Clear RQ queues before and after each test. 

14 

15 Test classes using this mixin share a single RQ (Redis) instance. Under the parallel 

16 test runner that Redis is not isolated per worker (unlike the database), so concurrent 

17 classes that enqueue and assert exact queue counts race each other. SerializeMixin 

18 holds an exclusive lock on `lockfile`, so no two classes using this mixin run at the 

19 same time, which removes that cross-worker contention. 

20 """ 

21 lockfile = __file__ 

22 rq_queue_names = ('default', 'high', 'low') 

23 

24 @classmethod 

25 def clear_rq_queues(cls): 

26 for queue_name in cls.rq_queue_names: 

27 # Flush only the queue's own database. FLUSHALL would empty every database on the Redis 

28 # server, including the caching database, whose keys (e.g. the cached config revision) 

29 # are shared by the other parallel test workers. 

30 get_queue(queue_name).connection.flushdb() 

31 

32 def run_rq_jobs(self, *queue_names, burst=True): 

33 """ 

34 Process queued RQ jobs synchronously for the given queue(s) (defaulting to 'default'). 

35 

36 Uses a non-forking SimpleWorker: the default RQ worker forks a work horse which would 

37 inherit the test's open database connection. Two processes sharing one connection 

38 corrupts it — on an SSL-encrypted connection this surfaces as "bad record mac" and 

39 closes the connection for every subsequent test. SimpleWorker runs jobs in-process, 

40 so the connection is never shared. 

41 """ 

42 worker = get_worker(*(queue_names or ('default',)), worker_class=SimpleWorker) 

43 worker.work(burst=burst) 

44 

45 def setUp(self): 

46 super().setUp() 

47 

48 # Clear all queues before running each test 

49 self.clear_rq_queues() 

50 

51 def tearDown(self): 

52 try: 

53 # Clear all queues after each test so no leftover jobs leak into the next test suite 

54 self.clear_rq_queues() 

55 finally: 

56 super().tearDown()