adding two tests to ensure ttl behaviour

main
glaslos 10 years ago
parent 5b5ab4860f
commit 3af0e95ce5

@ -85,5 +85,6 @@ with Connection():
return x + y
def long_running_job():
time.sleep(10)
def long_running_job(timeout=10):
time.sleep(timeout)
return 'Done sleeping...'

@ -12,7 +12,7 @@ from rq.utils import utcformat
from tests import RQTestCase
from tests.fixtures import (access_self, CallableObject, Number, say_hello,
some_calculation)
some_calculation, long_running_job)
from tests.helpers import strip_microseconds
try:
@ -313,6 +313,22 @@ class TestJob(RQTestCase):
job.save()
self.assertEqual(job.get_ttl(), None)
def test_ttl_via_enqueue(self):
ttl = 1
queue = Queue(connection=self.testconn)
job = queue.enqueue(say_hello, ttl=ttl)
self.assertEqual(job.get_ttl(), ttl)
def test_expire_during_execution(self):
"""Test what happens when job expires during execution"""
ttl = 2
queue = Queue(connection=self.testconn)
job = queue.enqueue(long_running_job, args=(4,), ttl=ttl)
self.assertEqual(job.get_ttl(), ttl)
job.perform()
self.assertFalse(job.exists(job.id))
self.assertEqual(job.result, 'Done sleeping...')
def test_cleanup(self):
"""Test that jobs and results are expired properly."""
job = Job.create(func=say_hello)

Loading…
Cancel
Save