fix(major): upgrade redis-py and rq and try and fix scheduler tests

frappe/core/doctype/scheduled_job_type/scheduled_job_type.py

@@ -34,10 +34,10 @@ class ScheduledJobType(Document):
 					enqueue('frappe.core.doctype.scheduled_job_type.scheduled_job_type.run_scheduled_job',
 						queue = self.get_queue_name(), job_type=self.method)
 					return True
-
+				else:
+					pass
 		else:
 			pass
-			#print('not yet due')
 
 		return False
 
@@ -111,12 +111,11 @@ class ScheduledJobType(Document):
 def execute_event(doc):
 	frappe.only_for('System Manager')
 	doc = json.loads(doc)
-	frappe.get_doc('Scheduled Job Type', doc.get('name')).execute()
+	frappe.get_doc('Scheduled Job Type', doc.get('name')).enqueue()
 
 def run_scheduled_job(job_type):
 	'''This is a wrapper function that runs a hooks.scheduler_events method'''
 	try:
-		print('executing job {}'.format(job_type))
 		frappe.get_doc('Scheduled Job Type', dict(method=job_type)).execute()
 	except Exception:
 		print(frappe.get_traceback())

frappe/tests/test_scheduler.py

@@ -37,10 +37,17 @@ class TestScheduler(TestCase):
 
 	def test_queue_peeking(self):
 		job = get_test_job()
+
 		self.assertTrue(job.enqueue())
 		job.db_set('last_execution', '2010-01-01 00:00:00')
 		frappe.db.commit()
-		time.sleep(3) # wait if job is not yet queued
+
+		# 1 job in queue
+		self.assertTrue(job.enqueue())
+		job.db_set('last_execution', '2010-01-01 00:00:00')
+		frappe.db.commit()
+
+		# 2nd job not loaded
 		self.assertFalse(job.enqueue())
 		job.delete()
 
@@ -88,11 +95,11 @@ def get_test_job(method='frappe.tests.test_scheduler.test_timeout_10', queue='Al
 			last_execution = '2010-01-01 00:00:00',
 			queue = queue
 		)).insert()
-		frappe.db.commit()
 	else:
 		job = frappe.get_doc('Scheduled Job Type', dict(method=method))
 		job.db_set('last_execution', '2010-01-01 00:00:00')
 		job.db_set('queue', queue)
+	frappe.db.commit()
 
 	return job
 

frappe/utils/background_jobs.py

@@ -119,6 +119,8 @@ def execute_job(site, method, event, job_name, kwargs, user=None, is_async=True,
 	except:
 		frappe.db.rollback()
 		frappe.log_error(method_name)
+		frappe.db.commit()
+		print(frappe.get_traceback())
 		raise
 
 	else:
@@ -140,8 +142,6 @@ def start_worker(queue=None, quiet = False):
 	with Connection(redis_connection):
 		queues = get_queue_list(queue)
 		logging_level = "INFO"
-		if quiet:
-			logging_level = "WARNING"
 		Worker(queues, name=get_worker_name(queue)).work(logging_level = logging_level)
 
 def get_worker_name(queue):
@@ -203,13 +203,7 @@ def get_queue_list(queue_list=None):
 def get_queue(queue, is_async=True):
 	'''Returns a Queue object tied to a redis connection'''
 	validate_queue(queue)
-
-	kwargs = {
-		'connection': get_redis_conn(),
-		'async': is_async
-	}
-
-	return Queue(queue, **kwargs)
+	return Queue(queue, connection=get_redis_conn(), is_async=is_async)
 
 def validate_queue(queue, default_queue_list=None):
 	if not default_queue_list:

frappe/utils/redis_wrapper.py

@@ -141,6 +141,9 @@ class RedisWrapper(redis.Redis):
 		return super(RedisWrapper, self).llen(self.make_key(key))
 
 	def hset(self, name, key, value, shared=False):
+		if key is None:
+			return
+
 		_name = self.make_key(name, shared=shared)
 
 		# set in local
@@ -164,6 +167,8 @@ class RedisWrapper(redis.Redis):
 		if not _name in frappe.local.cache:
 			frappe.local.cache[_name] = {}
 
+		if not key: return None
+
 		if key in frappe.local.cache[_name]:
 			return frappe.local.cache[_name][key]
 

requirements.txt

@@ -13,7 +13,7 @@ werkzeug
 semantic_version
 rauth>=0.6.2
 requests
-redis==2.10.6
+redis>=3.0
 selenium
 babel==2.6.0
 ipython
@@ -26,7 +26,7 @@ bleach==2.1.4
 bleach-whitelist
 Pillow
 beautifulsoup4
-rq==0.12.0
+rq>=1.1.0
 schedule
 cryptography
 pyopenssl