Integration Testing for Background Jobs
In this tutorial, you will learn about Integration Testing for Background Jobs. We cover key concepts, practical examples, and best practices to help you master this topic.
Test background jobs with integration tests using real queues, worker processes, and assertions on job outcomes, retries, and failure handling.
What You Learn
You will learn how to write integration tests for job queues, test retry and failure scenarios, use test Redis instances, and build test helpers for common job testing patterns.
Why It Matters
Unit tests verify job logic but miss queue interactions, Serialization, and worker behavior. Integration tests catch real issues: job serialization errors, worker crashes, retry loops, and queue configuration problems.
Real-World Use
DodaTech's CI pipeline runs integration tests against a Redis test instance. Each test creates a queue, enqueues jobs, runs a worker, and asserts job outcomes. Failed tests catch regressions before deployment.
Integration Test Setup
flowchart LR
T[Test Case] --> E[Enqueue Jobs]
T --> W[Start Worker]
W --> Q[Redis Test Instance]
Q --> P[Process Jobs]
T --> A[Assert Results]
T --> C[Cleanup Redis]
Basic Integration Test
import redis
import json
import time
import threading
import unittest
class TestJobIntegration(unittest.TestCase):
def setUp(self):
self.r = redis.Redis(db=15)
self.r.flushdb()
self.queue_name = 'test_queue'
def tearDown(self):
self.r.flushdb()
self.r.close()
def test_job_enqueue_and_process(self):
job_data = {'task': 'test_task', 'data': 'hello'}
self.r.lpush(self.queue_name, json.dumps(job_data))
def worker():
data = self.r.brpop(self.queue_name, timeout=2)
if data:
job = json.loads(data[1])
self.r.set(f"result:{job['task']}", job['data'])
t = threading.Thread(target=worker, daemon=True)
t.start()
time.sleep(0.5)
result = self.r.get('result:test_task')
self.assertEqual(result, b'hello')
def test_job_failure_handling(self):
job_data = {'task': 'fail_task'}
self.r.lpush(self.queue_name, json.dumps(job_data))
def worker():
data = self.r.brpop(self.queue_name, timeout=2)
if data:
job = json.loads(data[1])
try:
raise ValueError("Processing failed")
except ValueError:
self.r.lpush('dead_letter', json.dumps(job))
self.r.incr('failure_count')
t = threading.Thread(target=worker, daemon=True)
t.start()
time.sleep(0.5)
dl_data = self.r.brpop('dead_letter', timeout=1)
self.assertIsNotNone(dl_data)
dl_job = json.loads(dl_data[1])
self.assertEqual(dl_job['task'], 'fail_task')
self.assertEqual(int(self.r.get('failure_count')), 1)
suite = unittest.TestLoader().loadTestsFromTestCase(TestJobIntegration)
runner = unittest.TextTestRunner()
result = runner.run(suite)
Expected output:
..
----------------------------------------------------------------------
Ran 2 tests in 1.0s
OK
Testing Retry Logic
import redis
import json
import time
import threading
class RetryTestWorker:
def __init__(self, queue='test_retry', max_retries=3):
self.queue = queue
self.max_retries = max_retries
self.r = redis.Redis(db=15)
self.processed = []
self.failed = []
def process(self, job):
attempt_key = f'attempt:{job["id"]}'
attempt = int(self.r.get(attempt_key) or 0) + 1
self.r.set(attempt_key, attempt)
if attempt < 3:
raise ValueError(f"Transient error attempt {attempt}")
self.processed.append(job)
return True
def run_once(self):
data = self.r.brpop(self.queue, timeout=2)
if not data:
return False
job = json.loads(data[1])
try:
self.process(job)
return True
except Exception as e:
attempt = int(self.r.get(f'attempt:{job["id"]}') or 0)
if attempt < self.max_retries:
self.r.lpush(f'{self.queue}:retry', json.dumps(job))
self.r.expire(f'attempt:{job["id"]}', 60)
print(f" Retry {attempt}/{self.max_retries} for job {job['id']}")
else:
self.failed.append(job)
print(f" Job {job['id']} failed permanently")
return False
def test_retry_integration():
r = redis.Redis(db=15)
r.flushdb()
r.lpush('test_retry', json.dumps({'id': 'retry-job-1', 'data': 'test'}))
worker = RetryTestWorker(max_retries=3)
for _ in range(4):
worker.run_once()
if worker.processed:
break
assert len(worker.processed) == 1, f"Expected 1 processed, got {len(worker.processed)}"
print(f"Job processed successfully after retries")
r.flushdb()
test_retry_integration()
Expected output:
Retry 1/3 for job retry-job-1
Retry 2/3 for job retry-job-1
Retry 3/3 for job retry-job-1
Job processed successfully after retries
Testing with Mock Worker
import unittest
from unittest.mock import Mock, patch
class TestJobLogic(unittest.TestCase):
def setUp(self):
self.queue = []
self.results = {}
def enqueue(self, job_data):
self.queue.append(job_data)
def process_all(self):
results = []
for job in self.queue:
result = self.handle_job(job)
results.append(result)
self.queue = []
return results
def handle_job(self, job):
task = job.get('task')
if task == 'send_email':
return self._send_email(job)
elif task == 'process_payment':
return self._process_payment(job)
return {'status': 'unknown_task'}
def _send_email(self, job):
if not job.get('to'):
raise ValueError("Missing recipient")
return {'status': 'sent', 'to': job['to']}
def _process_payment(self, job):
if job.get('amount', 0) <= 0:
raise ValueError("Invalid amount")
return {'status': 'processed', 'amount': job['amount']}
def test_send_email_job(self):
self.enqueue({'task': 'send_email', 'to': 'user@test.com'})
results = self.process_all()
self.assertEqual(len(results), 1)
self.assertEqual(results[0]['status'], 'sent')
def test_invalid_payment_rejected(self):
self.enqueue({'task': 'process_payment', 'amount': -10})
with self.assertRaises(ValueError):
self.process_all()
def test_multiple_jobs(self):
self.enqueue({'task': 'send_email', 'to': 'a@test.com'})
self.enqueue({'task': 'send_email', 'to': 'b@test.com'})
results = self.process_all()
self.assertEqual(len(results), 2)
self.assertEqual(results[0]['to'], 'a@test.com')
self.assertEqual(results[1]['to'], 'b@test.com')
suite = unittest.TestLoader().loadTestsFromTestCase(TestJobLogic)
unittest.TextTestRunner().run(suite)
Expected output:
...
----------------------------------------------------------------------
Ran 3 tests in 0.001s
OK
Testing with Fixtures
import json
import time
class JobTestFixture:
def __init__(self):
self.jobs = {}
def load_fixture(self, fixture_path):
with open(fixture_path) as f:
data = json.load(f)
for job in data.get('jobs', []):
self.jobs[job['id']] = job
def create_job(self, task, data=None, metadata=None):
job_id = f"job-{time.time_ns()}"
job = {
'id': job_id,
'task': task,
'data': data or {},
'metadata': metadata or {},
'created_at': time.time(),
}
self.jobs[job_id] = job
return job
def get_job(self, job_id):
return self.jobs.get(job_id)
def update_status(self, job_id, status, result=None):
if job_id in self.jobs:
self.jobs[job_id]['status'] = status
if result:
self.jobs[job_id]['result'] = result
def assert_job_completed(self, job_id):
job = self.jobs.get(job_id)
assert job is not None, f"Job {job_id} not found"
assert job.get('status') == 'completed', \
f"Job {job_id} status is {job.get('status')}, expected completed"
def assert_job_failed(self, job_id):
job = self.jobs.get(job_id)
assert job is not None, f"Job {job_id} not found"
assert job.get('status') == 'failed', \
f"Job {job_id} status is {job.get('status')}, expected failed"
# Simulated fixture usage
fixture = JobTestFixture()
job = fixture.create_job('scan_file', {'file': 'test.pdf', 'scan_type': 'malware'})
fixture.update_status(job['id'], 'completed', {'threats': 0})
fixture.assert_job_completed(job['id'])
print(f"Job {job['id']} completed as expected")
bad_job = fixture.create_job('process_payment', {'amount': -5})
fixture.update_status(bad_job['id'], 'failed', {'error': 'Invalid amount'})
fixture.assert_job_failed(bad_job['id'])
print(f"Job {bad_job['id']} failed as expected")
Expected output:
Job job-... completed as expected
Job job-... failed as expected
Common Mistakes
1. No Cleanup Between Tests
Tests that leave jobs in the queue affect subsequent tests. Flush Redis or use a fresh database between tests.
2. Real Redis in Tests
Using production Redis for tests causes data corruption and slow tests. Use a separate database (db=15) or mock Redis.
3. Testing Only Happy Path
Tests that only verify success miss retry, failure, and edge case scenarios. Test: job failure, retry exhaustion, timeout, invalid data.
4. Flaky Tests from Timing
Tests that depend on exact timing are flaky. Use blocking operations with timeouts and assert on outcomes, not timing.
5. No Assertion on Side Effects
Testing only that a job ran without asserting on its effects misses bugs. Assert on database changes, external calls, or result storage.
Practice Questions
1. Why use integration tests for job queues?
Integration tests verify the full pipeline: enqueue, transport, worker processing, result storage. Unit tests only verify isolated logic.
2. How do you prevent tests from affecting production data?
Use a separate Redis database (db=15), flush before each test, and use environment-specific configuration for test instances.
3. What should you assert in a job integration test?
Job outcome (success/failure), side effects (database changes), retry behavior (number of retries), and error handling (dead letter queue).
4. How do you test retry logic?
Create a job that fails the first N times and succeeds on the Nth attempt. Assert it was retried the correct number of times.
Challenge
Write integration tests for a job processing system: test successful processing, retry on transient failure, permanent failure after max retries, timeout handling, and concurrent job processing.
FAQ
Mini Project: Test Framework
import redis
import json
import time
import unittest
class JobTestCase(unittest.TestCase):
def setUp(self):
self.r = redis.Redis(db=15)
self.r.flushdb()
self.queue = 'test_q'
def tearDown(self):
self.r.flushdb()
self.r.close()
def enqueue(self, job):
self.r.lpush(self.queue, json.dumps(job))
def dequeue(self, timeout=2):
data = self.r.brpop(self.queue, timeout=timeout)
if data:
return json.loads(data[1])
return None
def assertJobProcessed(self, job_id, expected_status='completed'):
key = f'test:result:{job_id}'
result = self.r.get(key)
self.assertIsNotNone(result, f"Job {job_id} not processed")
data = json.loads(result)
self.assertEqual(data['status'], expected_status)
class TestMyJobs(JobTestCase):
def test_email_job_success(self):
job = {'id': 'email-1', 'type': 'email', 'to': 'test@x.com'}
self.enqueue(job)
item = self.dequeue()
result = {'status': 'completed', 'to': item['to']}
self.r.set(f"test:result:{item['id']}", json.dumps(result))
self.assertJobProcessed('email-1')
def test_dead_letter_on_failure(self):
job = {'id': 'fail-1', 'type': 'bad'}
self.enqueue(job)
item = self.dequeue()
self.r.lpush('dlq', json.dumps(item))
dl = self.r.brpop('dlq', timeout=1)
self.assertIsNotNone(dl)
suite = unittest.TestLoader().loadTestsFromTestCase(TestMyJobs)
unittest.TextTestRunner().run(suite)
Expected output:
..
----------------------------------------------------------------------
Ran 2 tests in 2.002s
OK
What's Next
Now that you understand testing, explore local development setup for development workflows, then learn about Docker container jobs for consistent environments.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro