use warnings.warn() instead of raise for DeprecationWarning

main
Andrew Gorcester 11 years ago
parent 052d0df4bf
commit 17a4ed056d

@ -4,6 +4,7 @@ from __future__ import (absolute_import, division, print_function,
import inspect
from uuid import uuid4
import warnings
from rq.compat import as_text, decode_redis_hash, string_types, text_type
@ -131,8 +132,9 @@ class Job(object):
return self._status
def _get_status(self):
raise DeprecationWarning(
"job.status is deprecated. Use job.get_status() instead"
warnings.warn(
"job.status is deprecated. Use job.get_status() instead",
DeprecationWarning
)
return self.get_status()
@ -141,8 +143,9 @@ class Job(object):
self.connection.hset(self.key, 'status', self._status)
def _set_status(self, status):
raise DeprecationWarning(
"job.status is deprecated. Use job.set_status() instead"
warnings.warn(
"job.status is deprecated. Use job.set_status() instead",
DeprecationWarning
)
self.set_status(status)

@ -11,6 +11,7 @@ import socket
import sys
import time
import traceback
import warnings
from rq.compat import as_text, text_type
@ -235,8 +236,9 @@ class Worker(object):
def _set_state(self, state):
"""Raise a DeprecationWarning if ``worker.state = X`` is used"""
raise DeprecationWarning(
"worker.state is deprecated, use worker.set_state() instead."
warnings.warn(
"worker.state is deprecated, use worker.set_state() instead.",
DeprecationWarning
)
self.set_state(state)
@ -245,8 +247,9 @@ class Worker(object):
def _get_state(self):
"""Raise a DeprecationWarning if ``worker.state == X`` is used"""
raise DeprecationWarning(
"worker.state is deprecated, use worker.get_state() instead."
warnings.warn(
"worker.state is deprecated, use worker.get_state() instead.",
DeprecationWarning
)
return self.get_state()

@ -3,10 +3,11 @@ from __future__ import (absolute_import, division, print_function,
unicode_literals)
from datetime import datetime
import warnings
from rq.compat import as_text, PY2
from rq.exceptions import NoSuchJobError, UnpickleError
from rq.job import get_current_job, Job
from rq.job import get_current_job, Job, Status
from rq.queue import Queue
from rq.utils import utcformat
@ -327,3 +328,21 @@ class TestJob(RQTestCase):
job.cancel()
self.assertFalse(self.testconn.exists(job.key))
self.assertFalse(self.testconn.exists(job.dependents_key))
def test_get_and_set_job_status(self):
"""job.get_status and set_status have same functionality as property"""
job = Job.create(func=say_hello)
job2 = Job.create(func=say_hello)
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("always")
job.set_status(Status.FINISHED)
job2.status = Status.FINISHED
self.assertEqual(len(w), 1)
self.assertTrue(issubclass(w[-1].category, DeprecationWarning))
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("always")
self.assertEqual(job.get_status(), job2.status)
self.assertEqual(len(w), 1)
self.assertTrue(issubclass(w[-1].category, DeprecationWarning))

Loading…
Cancel
Save