Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def create_crontab(uts):
ctab = CronTab()
if uts.cron_type == 'h':
if uts.cron_interval > 1:
interval_list = [str(x) for x in range(0, 23, uts.cron_interval)]
interval_str = ','.join(interval_list)
else:
interval_str = '*'
cronline = "%s %s * * * %s runtests %s %s >/tmp/unitTestCronjob%s.out 2>&1" % (
uts.cron_start, interval_str, os.path.join(settings.BASE_DIR, 'runmanage.sh'), uts.branch.repo.name,
uts.branch.name, uts.id)
logger.debug('Creating cron tab with line ' + cronline)
item = CronItem(line=cronline + ' #' + ('%s %d' % (CRON_COMMENT, uts.id)))
ctab.add(item)
ctab.write()
branch.name)
logger.debug('Creating cron tab with line ' + cronline)
item = CronItem(line=cronline + ' #' + (CRON_COMMENT + ' %d' % branch.id))
ctab.add(item)
ctab.write()
if branch.code_cron_enabled and branch.code_cron_type == 'h':
if branch.code_cron_interval > 1:
interval_list = [str(x) for x in range(0, 23, branch.code_cron_interval)]
interval_str = ','.join(interval_list)
else:
interval_str = '*'
cronline = "%s %s * * * %s %s %s >/var/sftmp/code_cronjob.out 2>&1" % (
branch.code_cron_start, interval_str, os.path.join(settings.BASE_DIR, 'code_cronjob.sh'), branch.repo.name,
branch.name)
logger.debug('Creating cron tab with line ' + cronline)
item = CronItem(line=cronline + ' #' + (CRON_COMMENT + ' %d' % branch.id))
ctab.add(item)
ctab.write()
def find_trigger(self, trigger_list):
for job in self.crons:
if job.get_trigger() == trigger_list:
yield job
def activate_triggered_jobs(self, job, state):
for job in self.find_trigger(["toggle", job, state]):
job.enable(not job.is_enabled())
for job in self.find_trigger(["on", job, state]):
job.enable(True)
for job in self.find_trigger(["off", job, state]):
job.enable(False)
class TCronItem(CronItem):
"""class for extending CronItem with triggers"""
PREFIX = "SuperCron__"
SEPARATOR = "%"
def __init__(self, line=None, command='', comment='', user=None, cron=None):
super(TCronItem, self).__init__(line, command, comment, user, cron)
def is_superjob(self):
return self.comment.startswith(self.PREFIX)
def get_name(self):
if self.is_superjob():
sep = self.comment.find(self.SEPARATOR)
if sep == -1:
return self.comment[len(self.PREFIX):]
def __call__(self, form, field, message=None):
schedule = (field.data or '').strip()
if not schedule:
return
try:
CronItem().setall(schedule)
except (KeyError, ValueError):
message = message or self.message or field.gettext('Invalid cron')
raise ValidationError(message)