Fixes Job.fetch when return value is unpickleable (#1184)

* Fixes Job.fetch when return value is unpickleable

* Fixed connection test in newer versions of Redis

rq/job.py

@@ -477,7 +477,12 @@ class Job(object):
         self.enqueued_at = str_to_date(obj.get('enqueued_at'))
         self.started_at = str_to_date(obj.get('started_at'))
         self.ended_at = str_to_date(obj.get('ended_at'))
-        self._result = unpickle(obj.get('result')) if obj.get('result') else None  # noqa
+        result = obj.get('result')
+        if result:
+            try:
+                self._result = unpickle(obj.get('result'))
+            except UnpickleError:
+                self._result = 'Unpickleable return value'
         self.timeout = parse_timeout(obj.get('timeout')) if obj.get('timeout') else None
         self.result_ttl = int(obj.get('result_ttl')) if obj.get('result_ttl') else None  # noqa
         self.failure_ttl = int(obj.get('failure_ttl')) if obj.get('failure_ttl') else None  # noqa

tests/test_connection.py

@@ -2,6 +2,8 @@
 from __future__ import (absolute_import, division, print_function,
                         unicode_literals)
 
+from redis import Redis
+
 from rq import Connection, Queue, use_connection, get_current_connection, pop_connection
 from rq.connections import NoRedisConnectionException
 
@@ -21,8 +23,8 @@ class TestConnectionInheritance(RQTestCase):
 
     def test_connection_stacking(self):
         """Connection stacking."""
-        conn1 = new_connection()
+        conn1 = Redis(db=4)
-        conn2 = new_connection()
+        conn2 = Redis(db=5)
 
         with Connection(conn1):
             q1 = Queue()

tests/test_job.py

@@ -367,11 +367,17 @@ class TestJob(RQTestCase):
         self.assertDictEqual(serialized, serialized2)
 
     def test_unpickleable_result(self):
-        """Unpickleable job result doesn't crash job.to_dict()"""
+        """Unpickleable job result doesn't crash job.save() and job.refresh()"""
         job = Job.create(func=fixtures.say_hello, args=('Lionel',))
         job._result = queue.Queue()
-        data = job.to_dict()
+        job.save()
-        self.assertEqual(data['result'], 'Unpickleable return value')
+        self.assertEqual(
+            self.testconn.hget(job.key, 'result').decode('utf-8'),
+            'Unpickleable return value'
+        )
+
+        job = Job.fetch(job.id)
+        self.assertEqual(job.result, 'Unpickleable return value')
 
     def test_result_ttl_is_persisted(self):
         """Ensure that job's result_ttl is set properly"""