Script 'mail_helper' called by obssrc
Hello community,
here is the log from the commit of package python-python-crontab for
openSUSE:Factory checked in at 2024-01-03 12:28:00
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Comparing /work/SRC/openSUSE:Factory/python-python-crontab (Old)
and /work/SRC/openSUSE:Factory/.python-python-crontab.new.28375 (New)
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Package is "python-python-crontab"
Wed Jan 3 12:28:00 2024 rev:8 rq:1136017 version:3.0.0
Changes:
--------
---
/work/SRC/openSUSE:Factory/python-python-crontab/python-python-crontab.changes
2023-02-28 12:49:50.336904563 +0100
+++
/work/SRC/openSUSE:Factory/.python-python-crontab.new.28375/python-python-crontab.changes
2024-01-03 12:28:00.974113554 +0100
@@ -1,0 +2,10 @@
+Mon Jan 1 20:28:24 UTC 2024 - Dirk Müller <dmuel...@suse.com>
+
+- update to 3.0.0:
+ * Add frequency checks at specific timestamp
+ * Fix lots of pylint errors and improve test coverage
+ * Improve schedule running with more information about what was
+ returned
+ * Cause an error when setting an invalid frequency
+
+-------------------------------------------------------------------
Old:
----
python-crontab-2.7.1.tar.gz
New:
----
python-crontab-3.0.0.tar.gz
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Other differences:
------------------
++++++ python-python-crontab.spec ++++++
--- /var/tmp/diff_new_pack.q52vru/_old 2024-01-03 12:28:01.614136927 +0100
+++ /var/tmp/diff_new_pack.q52vru/_new 2024-01-03 12:28:01.614136927 +0100
@@ -1,7 +1,7 @@
#
# spec file for package python-python-crontab
#
-# Copyright (c) 2023 SUSE LLC
+# Copyright (c) 2024 SUSE LLC
#
# All modifications and additions to the file contributed by third parties
# remain the property of their copyright owners, unless otherwise agreed
@@ -16,15 +16,17 @@
#
+%{?sle15_python_module_pythons}
Name: python-python-crontab
-Version: 2.7.1
+Version: 3.0.0
Release: 0
Summary: Python Crontab API
License: LGPL-3.0-only
Group: Development/Languages/Python
URL: https://gitlab.com/doctormo/python-crontab/
Source:
https://files.pythonhosted.org/packages/source/p/python-crontab/python-crontab-%{version}.tar.gz
-BuildRequires: %{python_module setuptools}
+BuildRequires: %{python_module pip}
+BuildRequires: %{python_module wheel}
BuildRequires: fdupes
BuildRequires: python-rpm-macros
Requires: python-python-dateutil
@@ -50,10 +52,10 @@
%setup -q -n python-crontab-%{version}
%build
-%python_build
+%pyproject_wheel
%install
-%python_install
+%pyproject_install
%python_expand %fdupes %{buildroot}%{$python_sitelib}
%check
@@ -63,7 +65,8 @@
}
export PATH=$PWD/build/bin:$PATH
# test_07_non_posix_shell - only for Windows
-%pytest -k "not test_07_non_posix_shell"
+# test_20_frequency_at_year - broken test which fails in leap years
+%pytest -k "not test_07_non_posix_shell and not test_20_frequency_at_year"
%files %{python_files}
%doc README.rst
@@ -71,6 +74,6 @@
%{python_sitelib}/cronlog.py
%{python_sitelib}/crontab.py
%{python_sitelib}/crontabs.py
-%{python_sitelib}/python_crontab-%{version}*-info
+%{python_sitelib}/python_crontab-%{version}.dist-info
%pycache_only %{python_sitelib}/__pycache__
++++++ python-crontab-2.7.1.tar.gz -> python-crontab-3.0.0.tar.gz ++++++
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/python-crontab-2.7.1/PKG-INFO
new/python-crontab-3.0.0/PKG-INFO
--- old/python-crontab-2.7.1/PKG-INFO 2022-12-22 05:48:55.191304000 +0100
+++ new/python-crontab-3.0.0/PKG-INFO 2023-07-13 16:53:01.829805600 +0200
@@ -1,6 +1,6 @@
Metadata-Version: 2.1
Name: python-crontab
-Version: 2.7.1
+Version: 3.0.0
Summary: Python Crontab API
Home-page: https://gitlab.com/doctormo/python-crontab/
Author: Martin Owens
@@ -26,8 +26,9 @@
Provides: crontab
Provides: crontabs
Provides: cronlog
-Provides-Extra: cron-description
+Description-Content-Type: text/x-rst
Provides-Extra: cron-schedule
+Provides-Extra: cron-description
License-File: COPYING
License-File: AUTHORS
@@ -345,7 +346,9 @@
tab = CronTab(tabfile='MyScripts.tab')
for result in tab.run_scheduler():
- print("This was printed to stdout by the process.")
+ print("Return code: {result.returncode}")
+ print("Standard Out: {result.stdout}")
+ print("Standard Err: {result.stderr}")
Do not do this, it won't work because it returns generator function::
@@ -361,15 +364,20 @@
Frequency Calculation
=====================
-
Every job's schedule has a frequency. We can attempt to calculate the number
-of times a job would execute in a give amount of time. We have three simple
-methods::
+of times a job would execute in a give amount of time. We have two variants
+`frequency_per_*` and `frequency_at_*` calculations. The `freqency_at_*`
+always returnes *times* a job would execute and is aware of leap years.
+
+
+`frequency_per_*`
+-----------------
+For `frequency_per_*` We have three simple methods::
job.setall("1,2 1,2 * * *")
job.frequency_per_day() == 4
-The per year frequency method will tell you how many days a year the
+The per year frequency method will tell you how many **days** a year the
job would execute::
job.setall("* * 1,2 1,2 *")
@@ -386,6 +394,43 @@
job > job2
job.slices == "*/5"
+
+`frequency_at_*`
+----------------
+For `frequency_at_*` We have four simple methods.
+
+The at per hour frequency method will tell you how many times the job would
+execute at a given hour::
+
+ job.setall("*/2 0 * * *")
+ job.frequency_at_hour() == 30
+ job.frequency_at_hour(year=2010, month=1, day=1, hour=0) == 30 # even
hour
+ job.frequency_at_hour(year=2010, month=1, day=1, hour=1) == 0 # odd hour
+
+The at day frequency method parameterized tells you how many times the job
+would execute at a given day::
+
+ job.setall("0 0 * * 1,2")
+ job.frequency_at_day(year=2010, month=1, day=18) == 24 # Mon Jan 18th 2020
+ job.frequency_at_day(year=2010, month=1, day=21) == 0 # Thu Jan 21th 2020
+
+The at month frequency method will tell you how many times the job would
+execute at a given month::
+
+ job.setall("0 0 * * *")
+ job.frequency_at_month() == <output_of_current_month>
+ job.frequency_at_month(year=2010, month=1) == 31
+ job.frequency_at_month(year=2010, month=2) == 28
+ job.frequency_at_month(year=2012, month=2) == 29 # leap year
+
+The at year frequency method will tell you how many times a year the
+job would execute::
+
+ job.setall("* * 3,29 2 *")
+ job.frequency_at_year(year=2021) == 24
+ job.frequency_at_year(year=2024) == 48 # leap year
+
+
Log Functionality
=================
@@ -472,5 +517,3 @@
- Python 3 (3.7, 3.8, 3.10) tested, python 2.6, 2.7 removed from support.
- Windows support works for non-system crontabs only.
( see mem_cron and file_cron examples above for usage )
-
-
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/python-crontab-2.7.1/README.rst
new/python-crontab-3.0.0/README.rst
--- old/python-crontab-2.7.1/README.rst 2022-12-22 05:32:19.000000000 +0100
+++ new/python-crontab-3.0.0/README.rst 2023-07-13 16:51:49.000000000 +0200
@@ -312,7 +312,9 @@
tab = CronTab(tabfile='MyScripts.tab')
for result in tab.run_scheduler():
- print("This was printed to stdout by the process.")
+ print("Return code: {result.returncode}")
+ print("Standard Out: {result.stdout}")
+ print("Standard Err: {result.stderr}")
Do not do this, it won't work because it returns generator function::
@@ -328,15 +330,20 @@
Frequency Calculation
=====================
-
Every job's schedule has a frequency. We can attempt to calculate the number
-of times a job would execute in a give amount of time. We have three simple
-methods::
+of times a job would execute in a give amount of time. We have two variants
+`frequency_per_*` and `frequency_at_*` calculations. The `freqency_at_*`
+always returnes *times* a job would execute and is aware of leap years.
+
+
+`frequency_per_*`
+-----------------
+For `frequency_per_*` We have three simple methods::
job.setall("1,2 1,2 * * *")
job.frequency_per_day() == 4
-The per year frequency method will tell you how many days a year the
+The per year frequency method will tell you how many **days** a year the
job would execute::
job.setall("* * 1,2 1,2 *")
@@ -353,6 +360,43 @@
job > job2
job.slices == "*/5"
+
+`frequency_at_*`
+----------------
+For `frequency_at_*` We have four simple methods.
+
+The at per hour frequency method will tell you how many times the job would
+execute at a given hour::
+
+ job.setall("*/2 0 * * *")
+ job.frequency_at_hour() == 30
+ job.frequency_at_hour(year=2010, month=1, day=1, hour=0) == 30 # even
hour
+ job.frequency_at_hour(year=2010, month=1, day=1, hour=1) == 0 # odd hour
+
+The at day frequency method parameterized tells you how many times the job
+would execute at a given day::
+
+ job.setall("0 0 * * 1,2")
+ job.frequency_at_day(year=2010, month=1, day=18) == 24 # Mon Jan 18th 2020
+ job.frequency_at_day(year=2010, month=1, day=21) == 0 # Thu Jan 21th 2020
+
+The at month frequency method will tell you how many times the job would
+execute at a given month::
+
+ job.setall("0 0 * * *")
+ job.frequency_at_month() == <output_of_current_month>
+ job.frequency_at_month(year=2010, month=1) == 31
+ job.frequency_at_month(year=2010, month=2) == 28
+ job.frequency_at_month(year=2012, month=2) == 29 # leap year
+
+The at year frequency method will tell you how many times a year the
+job would execute::
+
+ job.setall("* * 3,29 2 *")
+ job.frequency_at_year(year=2021) == 24
+ job.frequency_at_year(year=2024) == 48 # leap year
+
+
Log Functionality
=================
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/python-crontab-2.7.1/crontab.py
new/python-crontab-3.0.0/crontab.py
--- old/python-crontab-2.7.1/crontab.py 2022-12-22 05:43:07.000000000 +0100
+++ new/python-crontab-3.0.0/crontab.py 2023-07-13 15:57:37.000000000 +0200
@@ -93,12 +93,13 @@
import platform
import subprocess as sp
+from calendar import monthrange
from time import sleep
from datetime import time, date, datetime, timedelta
from collections import OrderedDict
__pkgname__ = 'python-crontab'
-__version__ = '2.7.1'
+__version__ = '3.0.0'
ITEMREX = re.compile(r'^\s*([^@#\s]+)\s+([^@#\s]+)\s+([^@#\s]+)\s+([^@#\s]+)'
r'\s+([^@#\s]+)\s+([^\n]*?)(\s+#\s*([^\n]*)|$)')
@@ -158,22 +159,6 @@
"""Returns the username of the current user"""
return pwd.getpwuid(os.getuid())[0]
-def open_pipe(cmd, *args, **flags):
- """Runs a program and orders the arguments for compatability.
-
- a. keyword args are flags and always appear /before/ arguments for bsd
- """
- cmd_args = tuple(shlex.split(cmd, posix=flags.pop('posix', POSIX)))
- env = flags.pop('env', None)
- for (key, value) in flags.items():
- if len(key) == 1:
- cmd_args += (("-%s" % key),)
- if value is not None:
- cmd_args += (str(value),)
- else:
- cmd_args += (("--%s=%s" % (key, value)),)
- args = tuple(arg for arg in (cmd_args + tuple(args)) if arg)
- return sp.Popen(args, stdout=sp.PIPE, stderr=sp.PIPE, env=env)
def _str(text):
"""Convert to the best string format for this python version"""
@@ -182,6 +167,53 @@
return text
+class Process:
+ """Runs a program and orders the arguments for compatability.
+
+ a. keyword args are flags and always appear /before/ arguments for bsd
+ """
+ def __init__(self, cmd, *args, **flags):
+ cmd_args = tuple(shlex.split(cmd, posix=flags.pop('posix', POSIX)))
+ self.env = flags.pop('env', None)
+ for (key, value) in flags.items():
+ if len(key) == 1:
+ cmd_args += (f"-{key}",)
+ if value is not None:
+ cmd_args += (str(value),)
+ else:
+ cmd_args += (f"--{key}={value}",)
+ self.args = tuple(arg for arg in (cmd_args + tuple(args)) if arg)
+ self.has_run = False
+ self.stdout = None
+ self.stderr = None
+ self.returncode = None
+
+ def _run(self):
+ """Run this process and return the popen process object"""
+ return sp.Popen(self.args, stdout=sp.PIPE, stderr=sp.PIPE,
env=self.env)
+
+ def run(self):
+ """Run this process and store whatever is returned"""
+ process = self._run()
+ (out, err) = process.communicate()
+ self.returncode = process.returncode
+ self.stdout = out.decode("utf-8")
+ self.stderr = err.decode("utf-8")
+ return self
+
+ def __str__(self):
+ return self.stdout.strip()
+
+ def __repr__(self):
+ return f"Process({self.args})"
+
+ def __int__(self):
+ return self.returncode
+
+ def __eq__(self, other):
+ return str(self) == other
+
+
class CronTab:
"""
Crontab object which can access any time based cron using the standard.
@@ -252,7 +284,7 @@
elif name == 'crons' and value:
raise AttributeError("You can NOT set crons attribute directly")
else:
- super(CronTab, self).__setattr__(name, value)
+ super().__setattr__(name, value)
def read(self, filename=None):
"""
@@ -273,12 +305,10 @@
lines = fhl.readlines()
elif self.user:
- (out, err) = open_pipe(self.cron_command, l='',
**self.user_opt).communicate()
- if err and 'no crontab for' in str(err):
- pass
- elif err:
- raise IOError("Read crontab %s: %s" % (self.user, err))
- lines = out.decode('utf-8').split("\n")
+ process = Process(self.cron_command, l='', **self.user_opt).run()
+ if process.stderr and 'no crontab for' not in process.stderr:
+ raise IOError(f"Read crontab {self.user}: {process.stderr}")
+ lines = process.stdout.split("\n")
self.lines = lines
@@ -305,8 +335,8 @@
cron_id = self.crons.index(before)
line_id = self.lines.index(before)
- except ValueError:
- raise ValueError("Can not find CronItem in crontab to insert
before")
+ except ValueError as err:
+ raise ValueError("Can not find CronItem in crontab to insert
before") from err
if item.is_valid():
item.env.update(self._parked_env)
@@ -349,7 +379,7 @@
return
if self.filen:
- fileh = open(self.filen, 'wb')
+ fileh = open(self.filen, 'wb') # pylint:
disable=consider-using-with
else:
filed, path = tempfile.mkstemp()
fileh = os.fdopen(filed, 'wb')
@@ -363,11 +393,11 @@
os.unlink(path)
raise IOError("Please specify user or filename to write.")
- proc = open_pipe(self.cron_command, path, **self.user_opt)
+ proc = Process(self.cron_command, path, **self.user_opt)._run()
ret = proc.wait()
if ret != 0:
- raise IOError("Program Error: {} returned {}: {}".format(
- self.cron_command, ret, proc.stderr.read()))
+ msg = proc.stderr.read()
+ raise IOError(f"Program Error: {self.cron_command} returned
{ret}: {msg}")
proc.stdout.close()
proc.stderr.close()
os.unlink(path)
@@ -383,17 +413,17 @@
if ret not in [None, -1]:
yield ret
- def run_scheduler(self, timeout=-1, **kwargs):
+ def run_scheduler(self, timeout=-1, cadence=60, warp=False):
"""Run the CronTab as an internal scheduler (generator)"""
count = 0
while count != timeout:
now = datetime.now()
- if 'warp' in kwargs:
+ if warp:
now += timedelta(seconds=count * 60)
for value in self.run_pending(now=now):
yield value
- sleep(kwargs.get('cadence', 60))
+ sleep(cadence)
count += 1
def render(self, errors=False, specials=True):
@@ -413,7 +443,7 @@
elif not errors:
crons.append('# DISABLED LINE\n# ' + line)
else:
- raise ValueError("Invalid line: %s" % line)
+ raise ValueError(f"Invalid line: {line}")
elif isinstance(line, CronItem):
if not line.is_valid() and not errors:
line.enabled = False
@@ -421,12 +451,12 @@
# Environment variables are attached to cron lines so order will
# always work no matter how you add lines in the middle of the stack.
- result = str(self.env) + u'\n'.join(crons)
- if result and result[-1] not in (u'\n', u'\r'):
- result += u'\n'
+ result = str(self.env) + '\n'.join(crons)
+ if result and result[-1] not in ('\n', '\r'):
+ result += '\n'
return result
- def new(self, command='', comment='', user=None, pre_comment=False,
before=None):
+ def new(self, command='', comment='', user=None, pre_comment=False,
before=None): # pylint: disable=too-many-arguments
"""
Create a new CronItem and append it to the cron.
@@ -541,12 +571,12 @@
def __repr__(self):
kind = 'System ' if self._user is False else ''
if self.filen:
- return "<%sCronTab '%s'>" % (kind, self.filen)
+ return f"<{kind}CronTab '{self.filen}'>"
if self.user and not self.user_opt:
return "<My CronTab>"
if self.user:
- return "<User CronTab '%s'>" % self.user
- return "<Unattached %sCronTab>" % kind
+ return f"<User CronTab '{self.user}'>"
+ return f"<Unattached {kind}CronTab>"
def __iter__(self):
"""Return generator so we can track jobs after removal"""
@@ -695,23 +725,24 @@
if not self.is_valid() and self.enabled:
raise ValueError('Refusing to render invalid crontab.'
' Disable to continue.')
- command = _str(self.command).replace(u'%', u'\\%')
+ command = _str(self.command).replace('%', '\\%')
user = ''
if self.cron and self.cron.user is False:
if not self.user:
raise ValueError("Job to system-cron format, no user set!")
user = self.user + ' '
- result = u"%s %s%s" % (self.slices.render(specials=specials), user,
command)
+ rend = self.slices.render(specials=specials)
+ result = f"{rend} {user}{command}"
if self.stdin:
result += ' %' + self.stdin.replace('\n', '%')
if not self.enabled:
- result = u"# " + result
+ result = "# " + result
if self.comment:
comment = self.comment = _str(self.comment)
if self.marker:
- comment = u"#%s: %s" % (self.marker, comment)
+ comment = f"#{self.marker}: {comment}"
else:
- comment = u"# " + comment
+ comment = "# " + comment
if SYSTEMV or self.pre_comment or self.stdin:
result = comment + "\n" + result
@@ -757,6 +788,34 @@
"""
return self.slices.frequency(year=year)
+ def frequency_at_hour(self, year=None, month=None, day=None, hour=None):
+ """Returns the number of times this item will execute in a given hour
+ (defaults to this hour)
+ """
+ return self.slices.frequency_at_hour(year=year, month=month, day=day,
hour=hour)
+
+ def frequency_at_day(self, year=None, month=None, day=None):
+ """Returns the number of times this item will execute in a given day
+ (defaults to today)
+ """
+ return self.slices.frequency_at_day(year=year, month=month, day=day)
+
+ def frequency_at_month(self, year=None, month=None):
+ """Returns the number of times this item will execute in a given month
+ (defaults to this month)
+ """
+ return self.slices.frequency_at_month(year=year, month=month)
+
+ def frequency_at_year(self, year=None):
+ """Returns the number of times this item will execute in a given year
+ (defaults to this year)
+ """
+ return self.slices.frequency_at_year(year=year)
+
+ def frequency(self, year=None):
+ """Return frequence per year times frequency per day"""
+ return self.frequency_per_year(year=year) * self.frequency_per_day()
+
def frequency_per_year(self, year=None):
"""Returns the number of /days/ this item will execute on in a year
(defaults to this year)
@@ -789,10 +848,10 @@
env = os.environ.copy()
env.update(self.env.all())
shell = self.env.get('SHELL', SHELL)
- (out, err) = open_pipe(shell, '-c', self.command,
env=env).communicate()
- if err:
- LOG.error(err.decode("utf-8"))
- return out.decode("utf-8").strip()
+ process = Process(shell, '-c', self.command, env=env).run()
+ if process.stderr:
+ LOG.error(process.stderr)
+ return process
def schedule(self, date_from=None):
"""Return a croniter schedule if available."""
@@ -801,9 +860,9 @@
try:
# Croniter is an optional import
from croniter.croniter import croniter # pylint:
disable=import-outside-toplevel
- except ImportError:
+ except ImportError as err:
raise ImportError("Croniter not available. Please install croniter"
- " python module via pip or your package manager")
+ " python module via pip or your package
manager") from err
return croniter(self.slices.clean_render(), date_from,
ret_type=datetime)
def description(self, **kw):
@@ -814,9 +873,9 @@
"""
try:
from cron_descriptor import ExpressionDescriptor # pylint:
disable=import-outside-toplevel
- except ImportError:
+ except ImportError as err:
raise ImportError("cron_descriptor not available. Please install"\
- "cron_descriptor python module via pip or your package manager")
+ "cron_descriptor python module via pip or your package manager")
from err
exdesc = ExpressionDescriptor(self.slices.clean_render(), **kw)
return exdesc.get_description()
@@ -874,7 +933,7 @@
return self.slices[4]
def __repr__(self):
- return "<CronItem '%s'>" % str(self)
+ return f"<CronItem '{self}'>"
def __len__(self):
return len(str(self))
@@ -922,7 +981,7 @@
def year(self):
"""Special every year target"""
if self.unit > 1:
- raise ValueError("Invalid value '%s', outside 1 year" % self.unit)
+ raise ValueError(f"Invalid value '{self.unit}', outside 1 year")
self.slices.setall('@yearly')
@@ -932,7 +991,7 @@
month requency and finally day of the week frequency.
"""
def __init__(self, *args):
- super(CronSlices, self).__init__([CronSlice(info) for info in S_INFO])
+ super().__init__([CronSlice(info) for info in S_INFO])
self.special = None
self.setall(*args)
self.is_valid = self.is_self_valid
@@ -977,7 +1036,8 @@
# It might be possible to later understand timedelta objects
# but there's no convincing mathematics to do the conversion yet.
if not isinstance(value, (list, tuple)):
- raise ValueError("Unknown type: {}".format(type(value).__name__))
+ typ = type(value).__name__
+ raise ValueError(f"Unknown type: {typ}")
return value, None
@staticmethod
@@ -986,10 +1046,10 @@
key = value.lstrip('@').lower()
if value.count(' ') == 4:
return value.strip().split(' '), None
- if key in SPECIALS.keys():
+ if key in SPECIALS:
return SPECIALS[key].split(' '), '@' + key
if value.startswith('@'):
- raise ValueError("Unknown special '{}'".format(value))
+ raise ValueError(f"Unknown special '{value}'")
return [value], None
def clean_render(self):
@@ -1006,7 +1066,7 @@
if not SYSTEMV and specials is True:
for (name, value) in SPECIALS.items():
if value == slices and name not in SPECIAL_IGNORE:
- return "@%s" % name
+ return f"@{name}"
return slices
def clear(self):
@@ -1045,6 +1105,82 @@
"""Returns the number of times this item will execute in any hour"""
return len(self[0])
+ def frequency_at_year(self, year=None):
+ """Returns the number of /days/ this item will execute
+ in a given year (default is this year)"""
+ if not year:
+ year = date.today().year
+
+ total = 0
+ for month in range(1, 13):
+ total += self.frequency_at_month(year, month)
+ return total
+
+ def frequency_at_month(self, year=None, month=None):
+ """Returns the number of times this item will execute in given month
+ (default: current month)
+ """
+ if year is None and month is None:
+ year = date.today().year
+ month = date.today().month
+ elif year is None or month is None:
+ raise ValueError(
+ f"One of more arguments undefined: year={year}, month={month}")
+
+ total = 0
+ if month in self[3]:
+ # Calculate amount of days of specific month
+ days = monthrange(year, month)[1]
+ for day in range(1, days + 1):
+ total += self.frequency_at_day(year, month, day)
+ return total
+
+ def frequency_at_day(self, year=None, month=None, day=None):
+ """Returns the number of times this item will execute in a day
+ (default: any executed day)
+ """
+ # If arguments provided, all needs to be provided
+ test_none = [x is None for x in [year, month, day]]
+
+ if all(test_none):
+ return len(self[0]) * len(self[1])
+
+ if any(test_none):
+ raise ValueError(
+ f"One of more arguments undefined: year={year}, month={month},
day={day}")
+
+ total = 0
+ if day in self[2]:
+ for hour in range(24):
+ total += self.frequency_at_hour(year, month, day, hour)
+ return total
+
+ def frequency_at_hour(self, year=None, month=None, day=None, hour=None):
+ """Returns the number of times this item will execute in a hour
+ (default: any executed hour)
+ """
+ # If arguments provided, all needs to be provided
+ test_none = [x is None for x in [year, month, day, hour]]
+
+ if all(test_none):
+ return len(self[0])
+
+ if any(test_none):
+ raise ValueError(
+ f"One of more arguments undefined: year={year}, month={month},
day={day}, hour={hour}")
+
+ result = 0
+ weekday = date(year, month, day).weekday()
+
+ # Check if scheduled for execution at defined moment
+ if hour in self[1] and \
+ day in self[2] and \
+ month in self[3] and \
+ ((weekday + 1) % 7) in self[4]:
+ result = len(self[0])
+
+ return result
+
def __str__(self):
return self.render()
@@ -1098,7 +1234,7 @@
continue
self.parts.append(self.parse_value(part, sunday=0))
- def render(self, resolve=False, specials=True):
+ def render(self, resolve=False):
"""Return the slice rendered as a crontab.
resolve - return integer values instead of enums (default False)
@@ -1109,7 +1245,7 @@
return _render_values(self.parts, ',', resolve)
def __repr__(self):
- return "<CronSlice '%s'>" % str(self)
+ return f"<CronSlice '{self}'>"
def __eq__(self, value):
return str(self) == str(value)
@@ -1182,10 +1318,10 @@
val = self.min
try:
out = get_cronvalue(val, self.enum)
- except ValueError:
- raise ValueError("Unrecognised %s: '%s'" % (self.name, val))
- except KeyError:
- raise KeyError("No enumeration for %s: '%s'" % (self.name, val))
+ except ValueError as err:
+ raise ValueError(f"Unrecognised {self.name}: '{val}'") from err
+ except KeyError as err:
+ raise KeyError(f"No enumeration for {self.name}: '{val}'") from err
if self.max == 6 and int(out) == 7:
if sunday is not None:
@@ -1193,7 +1329,7 @@
raise SundayError("Detected Sunday as 7 instead of 0!")
if int(out) < self.min or int(out) > self.max:
- raise ValueError("'{1}', not in {0.min}-{0.max} for
{0.name}".format(self, val))
+ raise ValueError(f"'{val}', not in {self.min}-{self.max} for
{self.name}")
return out
@@ -1240,7 +1376,7 @@
return value.render(resolve)
if resolve:
return str(int(value))
- return str(u'{:02d}'.format(value) if ZERO_PAD else value)
+ return str(f'{value:02d}' if ZERO_PAD else value)
class CronRange:
@@ -1286,11 +1422,11 @@
self.dangling = 0
self.vto = self.slice.parse_value(vto, sunday=6)
if self.vto < self.vfrom:
- raise ValueError("Bad range '{0.vfrom}-{0.vto}'".format(self))
+ raise ValueError(f"Bad range '{self.vfrom}-{self.vto}'")
elif value == '*':
self.all()
else:
- raise ValueError('Unknown cron range value "%s"' % value)
+ raise ValueError(f'Unknown cron range value "{value}"')
def all(self):
"""Set this slice to all units between the miniumum and maximum"""
@@ -1306,7 +1442,7 @@
else:
value = _render_values([self.vfrom, self.vto], '-', resolve)
if self.seq != 1:
- value += "/%d" % self.seq
+ value += f"/{self.seq:d}"
if value != '*' and SYSTEMV:
value = ','.join([str(val) for val in self.range()])
return value
@@ -1344,7 +1480,7 @@
"""
def __init__(self, *args, **kw):
self.job = kw.pop('job', None)
- super(OrderedVariableList, self).__init__(*args, **kw)
+ super().__init__(*args, **kw)
@property
def previous(self):
@@ -1370,10 +1506,10 @@
def __getitem__(self, key):
previous = self.previous
if key in self:
- return super(OrderedVariableList, self).__getitem__(key)
+ return super().__getitem__(key)
if previous is not None:
return previous.all()[key]
- raise KeyError("Environment Variable '%s' not found." % key)
+ raise KeyError(f"Environment Variable '{key}' not found.")
def __str__(self):
"""Constructs to variable list output used in cron jobs"""
@@ -1383,7 +1519,7 @@
if self.previous.all().get(key, None) == value:
continue
if ' ' in str(value) or value == '':
- value = '"%s"' % value
- ret.append("%s=%s" % (key, str(value)))
+ value = f'"{value}"'
+ ret.append(f"{key}={value}")
ret.append('')
return "\n".join(ret)
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore'
old/python-crontab-2.7.1/python_crontab.egg-info/PKG-INFO
new/python-crontab-3.0.0/python_crontab.egg-info/PKG-INFO
--- old/python-crontab-2.7.1/python_crontab.egg-info/PKG-INFO 2022-12-22
05:48:55.000000000 +0100
+++ new/python-crontab-3.0.0/python_crontab.egg-info/PKG-INFO 2023-07-13
16:53:01.000000000 +0200
@@ -1,6 +1,6 @@
Metadata-Version: 2.1
Name: python-crontab
-Version: 2.7.1
+Version: 3.0.0
Summary: Python Crontab API
Home-page: https://gitlab.com/doctormo/python-crontab/
Author: Martin Owens
@@ -26,8 +26,9 @@
Provides: crontab
Provides: crontabs
Provides: cronlog
-Provides-Extra: cron-description
+Description-Content-Type: text/x-rst
Provides-Extra: cron-schedule
+Provides-Extra: cron-description
License-File: COPYING
License-File: AUTHORS
@@ -345,7 +346,9 @@
tab = CronTab(tabfile='MyScripts.tab')
for result in tab.run_scheduler():
- print("This was printed to stdout by the process.")
+ print("Return code: {result.returncode}")
+ print("Standard Out: {result.stdout}")
+ print("Standard Err: {result.stderr}")
Do not do this, it won't work because it returns generator function::
@@ -361,15 +364,20 @@
Frequency Calculation
=====================
-
Every job's schedule has a frequency. We can attempt to calculate the number
-of times a job would execute in a give amount of time. We have three simple
-methods::
+of times a job would execute in a give amount of time. We have two variants
+`frequency_per_*` and `frequency_at_*` calculations. The `freqency_at_*`
+always returnes *times* a job would execute and is aware of leap years.
+
+
+`frequency_per_*`
+-----------------
+For `frequency_per_*` We have three simple methods::
job.setall("1,2 1,2 * * *")
job.frequency_per_day() == 4
-The per year frequency method will tell you how many days a year the
+The per year frequency method will tell you how many **days** a year the
job would execute::
job.setall("* * 1,2 1,2 *")
@@ -386,6 +394,43 @@
job > job2
job.slices == "*/5"
+
+`frequency_at_*`
+----------------
+For `frequency_at_*` We have four simple methods.
+
+The at per hour frequency method will tell you how many times the job would
+execute at a given hour::
+
+ job.setall("*/2 0 * * *")
+ job.frequency_at_hour() == 30
+ job.frequency_at_hour(year=2010, month=1, day=1, hour=0) == 30 # even
hour
+ job.frequency_at_hour(year=2010, month=1, day=1, hour=1) == 0 # odd hour
+
+The at day frequency method parameterized tells you how many times the job
+would execute at a given day::
+
+ job.setall("0 0 * * 1,2")
+ job.frequency_at_day(year=2010, month=1, day=18) == 24 # Mon Jan 18th 2020
+ job.frequency_at_day(year=2010, month=1, day=21) == 0 # Thu Jan 21th 2020
+
+The at month frequency method will tell you how many times the job would
+execute at a given month::
+
+ job.setall("0 0 * * *")
+ job.frequency_at_month() == <output_of_current_month>
+ job.frequency_at_month(year=2010, month=1) == 31
+ job.frequency_at_month(year=2010, month=2) == 28
+ job.frequency_at_month(year=2012, month=2) == 29 # leap year
+
+The at year frequency method will tell you how many times a year the
+job would execute::
+
+ job.setall("* * 3,29 2 *")
+ job.frequency_at_year(year=2021) == 24
+ job.frequency_at_year(year=2024) == 48 # leap year
+
+
Log Functionality
=================
@@ -472,5 +517,3 @@
- Python 3 (3.7, 3.8, 3.10) tested, python 2.6, 2.7 removed from support.
- Windows support works for non-system crontabs only.
( see mem_cron and file_cron examples above for usage )
-
-
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/python-crontab-2.7.1/setup.py
new/python-crontab-3.0.0/setup.py
--- old/python-crontab-2.7.1/setup.py 2022-12-22 05:20:28.000000000 +0100
+++ new/python-crontab-3.0.0/setup.py 2023-07-13 16:52:41.000000000 +0200
@@ -40,6 +40,7 @@
release = RELEASE,
description = 'Python Crontab API',
long_description = description,
+ long_description_content_type = "text/x-rst",
author = 'Martin Owens',
url = 'https://gitlab.com/doctormo/python-crontab/',
author_email = 'docto...@gmail.com',
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/python-crontab-2.7.1/tests/test_compatibility.py
new/python-crontab-3.0.0/tests/test_compatibility.py
--- old/python-crontab-2.7.1/tests/test_compatibility.py 2020-05-17
19:13:48.000000000 +0200
+++ new/python-crontab-3.0.0/tests/test_compatibility.py 2023-07-13
15:44:06.000000000 +0200
@@ -116,9 +116,9 @@
def test_07_non_posix_shell(self):
"""Shell in windows environments is split correctly"""
- from crontab import open_pipe
+ from crontab import Process
winfile = os.path.join(TEST_DIR, 'data', "bash\\win.exe")
- pipe = open_pipe("{sys.executable} {winfile}".format(winfile=winfile,
sys=sys), 'SLASHED', posix=False)
+ pipe = Process("{sys.executable} {winfile}".format(winfile=winfile,
sys=sys), 'SLASHED', posix=False)._run()
self.assertEqual(pipe.wait(), 0, 'Windows shell command not found!')
(out, err) = pipe.communicate()
self.assertEqual(out, b'Double Glazing Installed:SLASHED\n')
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/python-crontab-2.7.1/tests/test_frequency.py
new/python-crontab-3.0.0/tests/test_frequency.py
--- old/python-crontab-2.7.1/tests/test_frequency.py 2022-12-22
05:40:45.000000000 +0100
+++ new/python-crontab-3.0.0/tests/test_frequency.py 2022-12-31
20:51:32.000000000 +0100
@@ -153,6 +153,42 @@
job.setall("*/2 * * * *")
self.assertEqual(job.frequency_per_hour(), 30)
+ def test_17_frequency_at_hour(self):
+ """Frequency at hour at given moment"""
+ job = self.crontab.new(command='at_hour')
+ job.setall("*/2 10 * * *")
+ self.assertEqual(job.frequency_at_hour(2021, 7, 9, 10), 30)
+ self.assertEqual(job.frequency_at_hour(2021, 7, 9, 11), 0)
+ self.assertEqual(job.frequency_at_hour(), 30)
+ self.assertRaises(ValueError, job.frequency_at_hour, 2021)
+
+ def test_18_frequency_at_day(self):
+ """Frequency per day at given moment"""
+ job = self.crontab.new(command='at_day')
+ job.setall("2,4 7 9,14 * *")
+ self.assertEqual(job.frequency_at_day(2021, 7, 9), 2)
+ self.assertEqual(job.frequency_at_day(2021, 7, 10), 0)
+ self.assertEqual(job.frequency_at_day(), 2)
+ self.assertRaises(ValueError, job.frequency_at_day, 2021)
+
+ def test_19_frequency_at_month(self):
+ """Frequency per month at moment"""
+ job = self.crontab.new(command='at_month')
+ job.setall("2,4 9 7,14 10,11 *")
+ self.assertEqual(job.frequency_at_month(2021, 10), 4)
+ self.assertEqual(job.frequency_at_month(2021, 12), 0)
+ self.assertEqual(job.frequency_at_month(), 0)
+ self.assertRaises(ValueError, job.frequency_at_month, 2021)
+
+ def test_20_frequency_at_year(self):
+ """Frequency at leap year day"""
+ job = self.crontab.new(command='at_year')
+ job.setall("0 * 3,29 2 *")
+ self.assertEqual(job.frequency_at_year(2021), 24)
+ self.assertEqual(job.frequency_at_year(2024), 48)
+ self.assertEqual(job.frequency_at_year(), 24)
+
+
if __name__ == '__main__':
test_support.run_unittest(
FrequencyTestCase,
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/python-crontab-2.7.1/tests/test_usage.py
new/python-crontab-3.0.0/tests/test_usage.py
--- old/python-crontab-2.7.1/tests/test_usage.py 2022-12-22
05:45:50.000000000 +0100
+++ new/python-crontab-3.0.0/tests/test_usage.py 2023-07-13
16:00:42.000000000 +0200
@@ -210,13 +210,14 @@
self.assertEqual(cronitem.render(specials=None), '@daily true')
self.assertEqual(cronitem.render(specials=False), '0 0 * * * true')
- def test_25_open_pipe(self):
+ def test_25_process(self):
"""Test opening pipes"""
- from crontab import open_pipe, CRON_COMMAND
- pipe = open_pipe(CRON_COMMAND, h=None, a='one', abc='two')
- (out, err) = pipe.communicate()
- self.assertEqual(err, b'')
- self.assertEqual(out, b'--abc=two|-a|-h|one\n')
+ from crontab import Process, CRON_COMMAND
+ process = Process(CRON_COMMAND, h=None, a='one', abc='two').run()
+ self.assertEqual(int(process), 0)
+ self.assertEqual(repr(process)[:8], "Process(")
+ self.assertEqual(process.stderr, '')
+ self.assertEqual(process.stdout, '--abc=two|-a|-h|one\n')
def test_07_zero_padding(self):
"""Can we get zero padded output"""
