fix cron with py3, added initial tests, close #1618

This commit is contained in:
ilvalle
2017-04-29 20:41:12 +02:00
parent 1d77968a06
commit 90e606dcfd
3 changed files with 31 additions and 3 deletions

View File

@@ -19,13 +19,14 @@ import sched
import re
import datetime
import platform
import gluon.fileutils
from functools import reduce
try:
import cPickle as pickle
except:
import pickle
from gluon.settings import global_settings
from gluon import fileutils
from gluon._compat import to_bytes
from pydal.contrib import portalocker
logger = logging.getLogger("web2py.cron")
@@ -116,7 +117,7 @@ class Token(object):
def __init__(self, path):
self.path = os.path.join(path, 'cron.master')
if not os.path.exists(self.path):
fileutils.write_file(self.path, '', 'wb')
fileutils.write_file(self.path, to_bytes(''), 'wb')
self.master = None
self.now = time.time()
@@ -139,7 +140,7 @@ class Token(object):
if portalocker.LOCK_EX is None:
logger.warning('WEB2PY CRON: Disabled because no file locking')
return None
self.master = open(self.path, 'rb+')
self.master = fileutils.open_file(self.path, 'rb+')
try:
ret = None
portalocker.lock(self.master, portalocker.LOCK_EX)
@@ -167,6 +168,7 @@ class Token(object):
"""
Writes into cron.master the time when cron job was completed
"""
ret = self.master.closed
if not self.master.closed:
portalocker.lock(self.master, portalocker.LOCK_EX)
logger.debug('WEB2PY CRON: Releasing cron lock')
@@ -177,6 +179,7 @@ class Token(object):
pickle.dump((self.now, time.time()), self.master)
portalocker.unlock(self.master)
self.master.close()
return ret
def rangetolist(s, period='min'):

View File

@@ -22,6 +22,7 @@ from .test_appadmin import *
from .test_web import *
from .test_sqlhtml import *
from .test_scheduler import *
from .test_cron import *
if sys.version[:3] == '2.7':
from .test_is_url import *

24
gluon/tests/test_cron.py Normal file
View File

@@ -0,0 +1,24 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
""" Unit tests for cron """
import unittest, os
from gluon.newcron import Token, crondance
class TestCron(unittest.TestCase):
def test_Token(self):
appname_path = os.path.join(os.getcwd(), 'applications', 'welcome')
t = Token(path=appname_path)
self.assertNotEqual(t.acquire(), None)
self.assertFalse(t.release())
self.assertEqual(t.acquire(), None)
self.assertTrue(t.release())
return
def test_crondance(self):
#TODO update crondance to return something
crondance(os.getcwd())