Add stuck initiatives audit report

This commit is contained in:
Ezra
2026-04-03 22:42:06 +00:00
parent dc3d975c2f
commit 56aa692d1c
1267 changed files with 1263232 additions and 0 deletions

View File

@@ -0,0 +1,41 @@
from . import croniter as cron_m
from .croniter import (
DAY_FIELD,
HOUR_FIELD,
MINUTE_FIELD,
MONTH_FIELD,
OVERFLOW32B_MODE,
SECOND_FIELD,
UTC_DT,
YEAR_FIELD,
CroniterBadCronError,
CroniterBadDateError,
CroniterBadTypeRangeError,
CroniterError,
CroniterNotAlphaError,
CroniterUnsupportedSyntaxError,
croniter,
croniter_range,
datetime_to_timestamp,
)
__all__ = [
"DAY_FIELD",
"HOUR_FIELD",
"MINUTE_FIELD",
"MONTH_FIELD",
"OVERFLOW32B_MODE",
"SECOND_FIELD",
"UTC_DT",
"YEAR_FIELD",
"CroniterBadCronError",
"CroniterBadDateError",
"CroniterBadTypeRangeError",
"CroniterError",
"CroniterNotAlphaError",
"CroniterUnsupportedSyntaxError",
"cron_m",
"croniter",
"croniter_range",
"datetime_to_timestamp",
]

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,11 @@
import typing
import unittest
class TestCase(unittest.TestCase):
"""
We use this base class for all the tests in this package.
If necessary, we can put common utility or setup code in here.
"""
maxDiff: typing.Optional[int] = 10**10

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,69 @@
#!/usr/bin/env python
"""
All related DST croniter tests are isolated here.
"""
# -*- coding: utf-8 -*-
import os
import time
import unittest
from datetime import datetime
from croniter import cron_m, croniter
from croniter.tests import base
ORIG_OVERFLOW32B_MODE = cron_m.OVERFLOW32B_MODE
class CroniterDST138Test(base.TestCase):
"""
See https://github.com/kiorky/croniter/issues/138.
"""
_tz = "UTC"
def setUp(self):
self._time = os.environ.setdefault("TZ", "")
self.base = datetime(2024, 1, 25, 4, 46)
self.iter = croniter("*/5 * * * *", self.base)
self.results = [
datetime(2024, 1, 25, 4, 50),
datetime(2024, 1, 25, 4, 55),
datetime(2024, 1, 25, 5, 0),
]
self.tzname, self.timezone = time.tzname, time.timezone
def tearDown(self):
cron_m.OVERFLOW32B_MODE = ORIG_OVERFLOW32B_MODE
if not self._time:
del os.environ["TZ"]
else:
os.environ["TZ"] = self._time
time.tzset()
def test_issue_138_dt_to_ts_32b(self):
"""
test local tz, forcing 32b mode.
"""
self._test(m32b=True)
def test_issue_138_dt_to_ts_n(self):
"""
test local tz, forcing non 32b mode.
"""
self._test(m32b=False)
def _test(self, tz="UTC", m32b=True):
cron_m.OVERFLOW32B_MODE = m32b
os.environ["TZ"] = tz
time.tzset()
res = [self.iter.get_next(datetime) for i in range(3)]
self.assertEqual(res, self.results)
class CroniterDST138TestLocal(CroniterDST138Test):
_tz = "UTC-8"
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,557 @@
import random
import unittest
import uuid
from datetime import datetime, timedelta
from croniter import CroniterBadCronError, CroniterNotAlphaError, croniter
from croniter.tests import base
class CroniterHashBase(base.TestCase):
epoch = datetime(2020, 1, 1, 0, 0)
hash_id = "hello"
def _test_iter(
self, definition, expectations, delta, epoch=None, hash_id=None, next_type=None
):
if epoch is None:
epoch = self.epoch
if hash_id is None:
hash_id = self.hash_id
if next_type is None:
next_type = datetime
if not isinstance(expectations, (list, tuple)):
expectations = (expectations,)
obj = croniter(definition, epoch, hash_id=hash_id)
testval = obj.get_next(next_type)
self.assertIn(testval, expectations)
if delta is not None:
self.assertEqual(obj.get_next(next_type), testval + delta)
class CroniterHashTest(CroniterHashBase):
def test_hash_hourly(self):
"""Test manually-defined hourly"""
self._test_iter("H * * * *", datetime(2020, 1, 1, 0, 10), timedelta(hours=1))
def test_hash_daily(self):
"""Test manually-defined daily"""
self._test_iter("H H * * *", datetime(2020, 1, 1, 11, 10), timedelta(days=1))
def test_hash_weekly(self):
"""Test manually-defined weekly"""
# croniter 1.0.5 changes the defined weekly range from (0, 6)
# to (0, 7), to match cron's behavior that Sunday is 0 or 7.
# This changes the hash, so test for either.
self._test_iter(
"H H * * H",
(datetime(2020, 1, 3, 11, 10), datetime(2020, 1, 5, 11, 10)),
timedelta(weeks=1),
)
def test_hash_monthly(self):
"""Test manually-defined monthly"""
self._test_iter("H H H * *", datetime(2020, 1, 1, 11, 10), timedelta(days=31))
def test_hash_yearly(self):
"""Test manually-defined yearly"""
self._test_iter("H H H H *", datetime(2020, 9, 1, 11, 10), timedelta(days=365))
def test_hash_second(self):
"""Test seconds
If a sixth field is provided, seconds are included in the datetime()
"""
self._test_iter("H H * * * H", datetime(2020, 1, 1, 11, 10, 32), timedelta(days=1))
def test_hash_year(self):
"""Test years
provide a seventh field as year
"""
self._test_iter("H H * * * H H", datetime(2066, 1, 1, 11, 10, 32), timedelta(days=1))
def test_hash_id_change(self):
"""Test a different hash_id returns different results given same definition and epoch"""
self._test_iter("H H * * *", datetime(2020, 1, 1, 11, 10), timedelta(days=1))
self._test_iter(
"H H * * *", datetime(2020, 1, 1, 0, 24), timedelta(days=1), hash_id="different id"
)
def test_hash_epoch_change(self):
"""Test a different epoch returns different results given same definition and hash_id"""
self._test_iter("H H * * *", datetime(2020, 1, 1, 11, 10), timedelta(days=1))
self._test_iter(
"H H * * *",
datetime(2011, 11, 12, 11, 10),
timedelta(days=1),
epoch=datetime(2011, 11, 11, 11, 11),
)
def test_hash_range(self):
"""Test a hashed range definition"""
self._test_iter("H H H(3-5) * *", datetime(2020, 1, 5, 11, 10), timedelta(days=31))
self._test_iter(
"H H * * * 0 H(2025-2030)", datetime(2029, 1, 1, 11, 10), timedelta(days=1)
)
def test_hash_division(self):
"""Test a hashed division definition"""
self._test_iter("H H/3 * * *", datetime(2020, 1, 1, 2, 10), timedelta(hours=3))
self._test_iter(
"H H H H * H H/2", datetime(2020, 9, 1, 11, 10, 32), timedelta(days=365 * 2)
)
def test_hash_range_division(self):
"""Test a hashed range + division definition"""
self._test_iter("H(30-59)/10 H * * *", datetime(2020, 1, 1, 11, 30), timedelta(minutes=10))
def test_hash_invalid_range(self):
"""Test validation logic for range_begin and range_end values"""
try:
self._test_iter(
"H(11-10) H * * *", datetime(2020, 1, 1, 11, 31), timedelta(minutes=10)
)
except CroniterBadCronError as ex:
self.assertEqual(str(ex), "Range end must be greater than range begin")
def test_hash_id_bytes(self):
"""Test hash_id as a bytes object"""
self._test_iter(
"H H * * *",
datetime(2020, 1, 1, 14, 53),
timedelta(days=1),
hash_id=b"\x01\x02\x03\x04",
)
def test_hash_float(self):
"""Test result as a float object"""
self._test_iter("H H * * *", 1577877000.0, (60 * 60 * 24), next_type=float)
def test_invalid_definition(self):
"""Test an invalid definition raises CroniterNotAlphaError"""
with self.assertRaises(CroniterNotAlphaError):
croniter("X X * * *", self.epoch, hash_id=self.hash_id)
def test_invalid_hash_id_type(self):
"""Test an invalid hash_id type raises TypeError"""
with self.assertRaises(TypeError):
croniter("H H * * *", self.epoch, hash_id={1: 2})
def test_invalid_divisor(self):
"""Test an invalid divisor type raises CroniterBadCronError"""
with self.assertRaises(CroniterBadCronError):
croniter("* * H/0 * *", self.epoch, hash_id=self.hash_id)
class CroniterWordAliasTest(CroniterHashBase):
def test_hash_word_midnight(self):
"""Test built-in @midnight
@midnight is actually up to 3 hours after midnight, not exactly midnight
"""
self._test_iter("@midnight", datetime(2020, 1, 1, 2, 10, 32), timedelta(days=1))
def test_hash_word_hourly(self):
"""Test built-in @hourly"""
self._test_iter("@hourly", datetime(2020, 1, 1, 0, 10, 32), timedelta(hours=1))
def test_hash_word_daily(self):
"""Test built-in @daily"""
self._test_iter("@daily", datetime(2020, 1, 1, 11, 10, 32), timedelta(days=1))
def test_hash_word_weekly(self):
"""Test built-in @weekly"""
# croniter 1.0.5 changes the defined weekly range from (0, 6)
# to (0, 7), to match cron's behavior that Sunday is 0 or 7.
# This changes the hash, so test for either.
self._test_iter(
"@weekly",
(datetime(2020, 1, 3, 11, 10, 32), datetime(2020, 1, 5, 11, 10, 32)),
timedelta(weeks=1),
)
def test_hash_word_monthly(self):
"""Test built-in @monthly"""
self._test_iter("@monthly", datetime(2020, 1, 1, 11, 10, 32), timedelta(days=31))
def test_hash_word_yearly(self):
"""Test built-in @yearly"""
self._test_iter("@yearly", datetime(2020, 9, 1, 11, 10, 32), timedelta(days=365))
def test_hash_word_annually(self):
"""Test built-in @annually
@annually is the same as @yearly
"""
obj_annually = croniter("@annually", self.epoch, hash_id=self.hash_id)
obj_yearly = croniter("@yearly", self.epoch, hash_id=self.hash_id)
self.assertEqual(obj_annually.get_next(datetime), obj_yearly.get_next(datetime))
self.assertEqual(obj_annually.get_next(datetime), obj_yearly.get_next(datetime))
class CroniterHashExpanderBase(base.TestCase):
def setUp(self):
_rd = random.Random()
_rd.seed(100)
self.HASH_IDS = [uuid.UUID(int=_rd.getrandbits(128)).bytes for _ in range(350)]
class CroniterHashExpanderExpandMinutesTest(CroniterHashExpanderBase):
MIN_VALUE = 0
MAX_VALUE = 59
TOTAL = 60
def test_expand_minutes(self):
minutes = set()
expression = "H * * * *"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
minutes.add(expanded[0][0][0])
assert len(minutes) == self.TOTAL
assert min(minutes) == self.MIN_VALUE
assert max(minutes) == self.MAX_VALUE
def test_expand_minutes_range_2_minutes(self):
minutes = set()
expression = "H/2 * * * *"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
_minutes = expanded[0][0]
assert len(_minutes) == 30
minutes.update(_minutes)
assert len(minutes) == self.TOTAL
assert min(minutes) == self.MIN_VALUE
assert max(minutes) == self.MAX_VALUE
def test_expand_minutes_range_3_minutes(self):
minutes = set()
expression = "H/3 * * * *"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
_minutes = expanded[0][0]
assert len(_minutes) == 20
minutes.update(_minutes)
assert len(minutes) == self.TOTAL
assert min(minutes) == self.MIN_VALUE
assert max(minutes) == self.MAX_VALUE
def test_expand_minutes_range_15_minutes(self):
minutes = set()
expression = "H/15 * * * *"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
_minutes = expanded[0][0]
assert len(_minutes) == 4
minutes.update(_minutes)
assert len(minutes) == self.TOTAL
assert min(minutes) == self.MIN_VALUE
assert max(minutes) == self.MAX_VALUE
def test_expand_minutes_with_full_range(self):
minutes = set()
expression = "H(0-59) * * * *"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
minutes.add(expanded[0][0][0])
assert len(minutes) == self.TOTAL
assert min(minutes) == self.MIN_VALUE
assert max(minutes) == self.MAX_VALUE
class CroniterHashExpanderExpandHoursTest(CroniterHashExpanderBase):
MIN_VALUE = 0
MAX_VALUE = 23
TOTAL = 24
def test_expand_hours(self):
hours = set()
expression = "H H * * *"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
hours.add(expanded[0][1][0])
assert len(hours) == self.TOTAL
assert min(hours) == self.MIN_VALUE
assert max(hours) == self.MAX_VALUE
def test_expand_hours_range_every_2_hours(self):
hours = set()
expression = "H H/2 * * *"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
_hours = expanded[0][1]
assert len(_hours) == 12
hours.update(_hours)
assert len(hours) == self.TOTAL
assert min(hours) == self.MIN_VALUE
assert max(hours) == self.MAX_VALUE
def test_expand_hours_range_4_hours(self):
hours = set()
expression = "H H/4 * * *"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
_hours = expanded[0][1]
assert len(_hours) == 6
hours.update(_hours)
assert len(hours) == self.TOTAL
assert min(hours) == self.MIN_VALUE
assert max(hours) == self.MAX_VALUE
def test_expand_hours_range_8_hours(self):
hours = set()
expression = "H H/8 * * *"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
_hours = expanded[0][1]
assert len(_hours) == 3
hours.update(_hours)
assert len(hours) == self.TOTAL
assert min(hours) == self.MIN_VALUE
assert max(hours) == self.MAX_VALUE
def test_expand_hours_range_10_hours(self):
hours = set()
expression = "H H/10 * * *"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
_hours = expanded[0][1]
assert len(_hours) in {2, 3}
hours.update(_hours)
assert len(hours) == self.TOTAL
assert min(hours) == self.MIN_VALUE
assert max(hours) == self.MAX_VALUE
def test_expand_hours_range_12_hours(self):
hours = set()
expression = "H H/12 * * *"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
_hours = expanded[0][1]
assert len(_hours) == 2
hours.update(_hours)
assert len(hours) == self.TOTAL
assert min(hours) == self.MIN_VALUE
assert max(hours) == self.MAX_VALUE
def test_expand_hours_with_full_range(self):
minutes = set()
expression = "* H(0-23) * * *"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
minutes.add(expanded[0][1][0])
assert len(minutes) == self.TOTAL
assert min(minutes) == self.MIN_VALUE
assert max(minutes) == self.MAX_VALUE
class CroniterHashExpanderExpandMonthDaysTest(CroniterHashExpanderBase):
MIN_VALUE = 1
MAX_VALUE = 31
TOTAL = 31
def test_expand_month_days(self):
month_days = set()
expression = "H H H * *"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
month_days.add(expanded[0][2][0])
assert len(month_days) == self.TOTAL
assert min(month_days) == self.MIN_VALUE
assert max(month_days) == self.MAX_VALUE
def test_expand_month_days_range_2_days(self):
month_days = set()
expression = "0 0 H/2 * *"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
_days = expanded[0][2]
assert len(_days) in {15, 16}
month_days.update(_days)
assert len(month_days) == self.TOTAL
assert min(month_days) == self.MIN_VALUE
assert max(month_days) == self.MAX_VALUE
def test_expand_month_days_range_5_days(self):
month_days = set()
expression = "H H H/5 * *"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
_days = expanded[0][2]
assert len(_days) in {6, 7}
month_days.update(_days)
assert len(month_days) == self.TOTAL
assert min(month_days) == self.MIN_VALUE
assert max(month_days) == self.MAX_VALUE
def test_expand_month_days_range_12_days(self):
month_days = set()
expression = "H H H/12 * *"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
_days = expanded[0][2]
assert len(_days) in {2, 3}
month_days.update(_days)
assert len(month_days) == self.TOTAL
assert min(month_days) == self.MIN_VALUE
assert max(month_days) == self.MAX_VALUE
def test_expand_month_days_with_full_range(self):
month_days = set()
expression = "* * H(1-31) * *"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
month_days.add(expanded[0][2][0])
assert len(month_days) == self.TOTAL
assert min(month_days) == self.MIN_VALUE
assert max(month_days) == self.MAX_VALUE
class CroniterHashExpanderExpandMonthTest(CroniterHashExpanderBase):
MIN_VALUE = 1
MAX_VALUE = 12
TOTAL = 12
def test_expand_month_days(self):
month_days = set()
expression = "H H * H *"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
month_days.add(expanded[0][3][0])
assert len(month_days) == self.TOTAL
assert min(month_days) == self.MIN_VALUE
assert max(month_days) == self.MAX_VALUE
def test_expand_month_days_range_2_months(self):
months = set()
expression = "H H * H/2 *"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
_months = expanded[0][3]
assert len(_months) == 6
months.update(_months)
assert len(months) == self.TOTAL
assert min(months) == self.MIN_VALUE
assert max(months) == self.MAX_VALUE
def test_expand_month_days_range_3_months(self):
months = set()
expression = "H H * H/3 *"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
_months = expanded[0][3]
assert len(_months) == 4
months.update(_months)
assert len(months) == self.TOTAL
assert min(months) == self.MIN_VALUE
assert max(months) == self.MAX_VALUE
def test_expand_month_days_range_5_months(self):
months = set()
expression = "H H * H/5 *"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
_months = expanded[0][3]
assert len(_months) in {2, 3}
months.update(_months)
assert len(months) == self.TOTAL
assert min(months) == self.MIN_VALUE
assert max(months) == self.MAX_VALUE
def test_expand_months_with_full_range(self):
months = set()
expression = "* * * H(1-12) *"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
months.add(expanded[0][3][0])
assert len(months) == self.TOTAL
assert min(months) == self.MIN_VALUE
assert max(months) == self.MAX_VALUE
class CroniterHashExpanderExpandWeekDays(CroniterHashExpanderBase):
MIN_VALUE = 0
MAX_VALUE = 6
TOTAL = 7
def test_expand_week_days(self):
week_days = set()
expression = "H H * * H"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
week_days.add(expanded[0][4][0])
assert len(week_days) == self.TOTAL
assert min(week_days) == self.MIN_VALUE
assert max(week_days) == self.MAX_VALUE
def test_expand_week_days_range_2_days(self):
days = set()
expression = "H H * * H/2"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
_days = expanded[0][4]
assert len(_days) in {3, 4}
days.update(_days)
assert len(days) == self.TOTAL
assert min(days) == self.MIN_VALUE
assert max(days) == self.MAX_VALUE
def test_expand_week_days_range_4_days(self):
days = set()
expression = "H H * * H/4"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
_days = expanded[0][4]
assert len(_days) in {1, 2}
days.update(_days)
assert len(days) == self.TOTAL
assert min(days) == self.MIN_VALUE
assert max(days) == self.MAX_VALUE
def test_expand_week_days_with_full_range(self):
days = set()
expression = "* * * * H(0-6)"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
days.add(expanded[0][4][0])
assert len(days) == self.TOTAL
assert min(days) == self.MIN_VALUE
assert max(days) == self.MAX_VALUE
class CroniterHashExpanderExpandYearsTest(CroniterHashExpanderBase):
def test_expand_years_by_division(self):
years = set()
year_min, year_max = croniter.RANGES[6]
expression = "* * * * * * H/10"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
assert len(expanded[0][6]) == 13
years.update(expanded[0][6])
assert len(years) == year_max - year_min + 1
assert min(years) == year_min
assert max(years) == year_max
def test_expand_years_by_range(self):
years = set()
expression = "* * * * * * H(2020-2030)"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
years.add(expanded[0][6][0])
assert len(years) == 11
assert min(years) == 2020
assert max(years) == 2030
def test_expand_years_by_range_and_division(self):
years = set()
expression = "* * * * * * H(2020-2050)/10"
for hash_id in self.HASH_IDS:
expanded = croniter.expand(expression, hash_id=hash_id)
years.update(expanded[0][6])
assert len(years) == 31
assert min(years) == 2020
assert max(years) == 2050
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,49 @@
import unittest
from datetime import datetime, timedelta
from croniter import croniter
from croniter.tests import base
class CroniterRandomTest(base.TestCase):
epoch = datetime(2020, 1, 1, 0, 0)
def test_random(self):
"""Test random definition"""
obj = croniter("R R * * *", self.epoch)
result_1 = obj.get_next(datetime)
self.assertGreaterEqual(result_1, datetime(2020, 1, 1, 0, 0))
self.assertLessEqual(result_1, datetime(2020, 1, 1, 0, 0) + timedelta(days=1))
result_2 = obj.get_next(datetime)
self.assertGreaterEqual(result_2, datetime(2020, 1, 2, 0, 0))
self.assertLessEqual(result_2, datetime(2020, 1, 2, 0, 0) + timedelta(days=1))
def test_random_range(self):
"""Test random definition within a range"""
obj = croniter("R R R(10-20) * *", self.epoch)
result_1 = obj.get_next(datetime)
self.assertGreaterEqual(result_1, datetime(2020, 1, 10, 0, 0))
self.assertLessEqual(result_1, datetime(2020, 1, 10, 0, 0) + timedelta(days=11))
result_2 = obj.get_next(datetime)
self.assertGreaterEqual(result_2, datetime(2020, 2, 10, 0, 0))
self.assertLessEqual(result_2, datetime(2020, 2, 10, 0, 0) + timedelta(days=11))
def test_random_float(self):
"""Test random definition, float result"""
obj = croniter("R R * * *", self.epoch)
result_1 = obj.get_next(float)
self.assertGreaterEqual(result_1, 1577836800.0)
self.assertLessEqual(result_1, 1577836800.0 + (60 * 60 * 24))
result_2 = obj.get_next(float)
self.assertGreaterEqual(result_2, 1577923200.0)
self.assertLessEqual(result_2, 1577923200.0 + (60 * 60 * 24))
def test_random_with_year(self):
obj = croniter("* * * * * * R(2025-2030)", self.epoch)
result = obj.get_next(datetime)
self.assertGreaterEqual(result.year, 2025)
self.assertLessEqual(result.year, 2030)
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,228 @@
#!/usr/bin/env python
import unittest
import zoneinfo
from datetime import datetime
import pytz
from croniter import (
CroniterBadCronError,
CroniterBadDateError,
CroniterBadTypeRangeError,
croniter,
croniter_range,
)
from croniter.tests import base
class mydatetime(datetime):
"""."""
class CroniterRangeTest(base.TestCase):
def test_1day_step(self):
start = datetime(2016, 12, 2)
stop = datetime(2016, 12, 10)
fwd = list(croniter_range(start, stop, "0 0 * * *"))
self.assertEqual(len(fwd), 9)
self.assertEqual(fwd[0], start)
self.assertEqual(fwd[-1], stop)
# Test the same, but in reverse
rev = list(croniter_range(stop, start, "0 0 * * *"))
self.assertEqual(len(rev), 9)
# Ensure forward/reverse are a mirror image
rev.reverse()
self.assertEqual(fwd, rev)
def test_1day_step_no_ends(self):
# Test without ends (exclusive)
start = datetime(2016, 12, 2)
stop = datetime(2016, 12, 10)
fwd = list(croniter_range(start, stop, "0 0 * * *", exclude_ends=True))
self.assertEqual(len(fwd), 7)
self.assertNotEqual(fwd[0], start)
self.assertNotEqual(fwd[-1], stop)
# Test the same, but in reverse
rev = list(croniter_range(stop, start, "0 0 * * *", exclude_ends=True))
self.assertEqual(len(rev), 7)
self.assertNotEqual(fwd[0], stop)
self.assertNotEqual(fwd[-1], start)
def test_1month_step(self):
start = datetime(1982, 1, 1)
stop = datetime(1983, 12, 31)
res = list(croniter_range(start, stop, "0 0 1 * *"))
self.assertEqual(len(res), 24)
self.assertEqual(res[0], start)
self.assertEqual(res[5].day, 1)
self.assertEqual(res[-1], datetime(1983, 12, 1))
def test_1minute_step_float(self):
start = datetime(2000, 1, 1, 0, 0)
stop = datetime(2000, 1, 1, 0, 1)
res = list(croniter_range(start, stop, "* * * * *", ret_type=float))
self.assertEqual(len(res), 2)
self.assertEqual(res[0], 946684800.0)
self.assertEqual(res[-1] - res[0], 60)
def test_auto_ret_type(self):
data = [
(datetime(2019, 1, 1), datetime(2020, 1, 1), datetime),
(1552252218.0, 1591823311.0, float),
]
for start, stop, rtype in data:
ret = list(croniter_range(start, stop, "0 0 * * *"))
self.assertIsInstance(ret[0], rtype)
def test_input_type_exceptions(self):
dt_start1 = datetime(2019, 1, 1)
dt_stop1 = datetime(2020, 1, 1)
f_start1 = 1552252218.0
f_stop1 = 1591823311.0
# Mix start/stop types
with self.assertRaises(TypeError):
list(croniter_range(dt_start1, f_stop1, "0 * * * *"), ret_type=datetime)
with self.assertRaises(TypeError):
list(croniter_range(f_start1, dt_stop1, "0 * * * *"))
def test_timezone_dst_pytz(self):
"""Test across DST transition, which technically is a timzone change in pytz."""
tz = pytz.timezone("America/New_York")
start = tz.localize(datetime(2020, 10, 30))
stop = tz.localize(datetime(2020, 11, 10))
res = list(croniter_range(start, stop, "0 0 * * *"))
self.assertNotEqual(res[0].tzinfo, res[-1].tzinfo)
self.assertEqual(len(res), 12)
def test_extra_hour_day_prio(self):
"""Test New York jumps forward: 2020-03-08 02:00 -> 03:00 (UTC-5 -> UTC-4)."""
tz = zoneinfo.ZoneInfo("America/New_York")
cron = "0 3 * * *"
start = datetime(2020, 3, 7, tzinfo=tz)
end = datetime(2020, 3, 11, tzinfo=tz)
ret = [i.isoformat() for i in croniter_range(start, end, cron)]
self.assertEqual(
ret,
[
"2020-03-07T03:00:00-05:00",
"2020-03-08T03:00:00-04:00",
"2020-03-09T03:00:00-04:00",
"2020-03-10T03:00:00-04:00",
],
)
def test_extra_hour_day_prio_pytz(self):
"""Test New York jumps forward: 2020-03-08 02:00 -> 03:00 (UTC-5 -> UTC-4)."""
def datetime_tz(*args, **kw):
"""Defined this in another branch. single-use-version"""
tzinfo = kw.pop("tzinfo")
return tzinfo.localize(datetime(*args))
tz = pytz.timezone("America/New_York")
cron = "0 3 * * *"
start = datetime_tz(2020, 3, 7, tzinfo=tz)
end = datetime_tz(2020, 3, 11, tzinfo=tz)
ret = [i.isoformat() for i in croniter_range(start, end, cron)]
self.assertEqual(
ret,
[
"2020-03-07T03:00:00-05:00",
"2020-03-08T03:00:00-04:00",
"2020-03-09T03:00:00-04:00",
"2020-03-10T03:00:00-04:00",
],
)
def test_issue145_getnext(self):
# Example of quarterly event cron schedule
start = datetime(2020, 9, 24)
cron = "0 13 8 1,4,7,10 wed"
with self.assertRaises(CroniterBadDateError):
it = croniter(cron, start, day_or=False, max_years_between_matches=1)
it.get_next()
# New functionality (0.3.35) allowing croniter to find spare matches of cron
# patterns across multiple years
it = croniter(cron, start, day_or=False, max_years_between_matches=5)
self.assertEqual(it.get_next(datetime), datetime(2025, 1, 8, 13))
def test_issue145_range(self):
cron = "0 13 8 1,4,7,10 wed"
matches = list(
croniter_range(datetime(2020, 1, 1), datetime(2020, 12, 31), cron, day_or=False)
)
self.assertEqual(len(matches), 3)
self.assertEqual(matches[0], datetime(2020, 1, 8, 13))
self.assertEqual(matches[1], datetime(2020, 4, 8, 13))
self.assertEqual(matches[2], datetime(2020, 7, 8, 13))
# No matches within this range; therefore expect empty list
matches = list(
croniter_range(datetime(2020, 9, 30), datetime(2020, 10, 30), cron, day_or=False)
)
self.assertEqual(len(matches), 0)
def test_croniter_range_derived_class(self):
# trivial example extending croniter
class croniter_nosec(croniter):
"""Like croniter, but it forbids second-level cron expressions."""
@classmethod
def expand(cls, expr_format, *args, **kwargs):
if len(expr_format.split()) == 6:
raise CroniterBadCronError("Expected 'min hour day mon dow'")
return croniter.expand(expr_format, *args, **kwargs)
cron = "0 13 8 1,4,7,10 wed"
matches = list(
croniter_range(
datetime(2020, 1, 1),
datetime(2020, 12, 31),
cron,
day_or=False,
_croniter=croniter_nosec,
)
)
self.assertEqual(len(matches), 3)
cron = "0 1 8 1,15,L wed 15,45"
with self.assertRaises(CroniterBadCronError):
# Should fail using the custom class that forbids the seconds expression
croniter_nosec(cron)
with self.assertRaises(CroniterBadCronError):
# Should similarly fail because the custom class rejects seconds expr
i = croniter_range(
datetime(2020, 1, 1), datetime(2020, 12, 31), cron, _croniter=croniter_nosec
)
next(i)
def test_dt_types(self):
start = mydatetime(2020, 9, 24)
stop = datetime(2020, 9, 28)
try:
list(croniter_range(start, stop, "0 0 * * *"))
except CroniterBadTypeRangeError:
self.fail("should not be triggered")
def test_configure_second_location(self):
start = datetime(2016, 12, 2, 0, 0, 0)
stop = datetime(2016, 12, 2, 0, 1, 0)
fwd = list(croniter_range(start, stop, "*/20 * * * * *", second_at_beginning=True))
self.assertEqual(len(fwd), 4)
self.assertEqual(fwd[0], start)
self.assertEqual(fwd[-1], stop)
def test_year_range(self):
start = datetime(2010, 1, 1)
stop = datetime(2030, 1, 1)
fwd = list(croniter_range(start, stop, "0 0 1 1 ? 0 2020-2024,2028"))
self.assertEqual(len(fwd), 6)
self.assertEqual(fwd[0], datetime(2020, 1, 1))
self.assertEqual(fwd[-1], datetime(2028, 1, 1))
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,111 @@
#!/usr/bin/env python
import os
import unittest
import zoneinfo
from datetime import datetime
from timeit import Timer
from croniter import croniter
from croniter.tests import base
class CroniterSpeedTest(base.TestCase):
def run_long_test(self, iterations=1):
dt = datetime(2010, 1, 23, 12, 18)
itr = croniter("*/1 * * * *", dt)
for i in range(iterations): # ~ 58
itr.get_next()
itr = croniter("*/5 * * * *", dt)
for i in range(iterations):
itr.get_next()
dt = datetime(2010, 1, 24, 12, 2)
itr = croniter("0 */3 * * *", dt)
for i in range(iterations):
itr.get_next()
dt = datetime(2010, 2, 24, 12, 9)
itr = croniter("0 0 */3 * *", dt)
for i in range(iterations):
itr.get_next(datetime)
# test leap year
dt = datetime(1996, 2, 27)
itr = croniter("0 0 * * *", dt)
for i in range(iterations):
itr.get_next(datetime)
dt2 = datetime(2000, 2, 27)
itr2 = croniter("0 0 * * *", dt2)
for i in range(iterations):
itr2.get_next(datetime)
dt = datetime(2010, 2, 25)
itr = croniter("0 0 * * sat", dt)
for i in range(iterations):
itr.get_next(datetime)
dt = datetime(2010, 1, 25)
itr = croniter("0 0 1 * wed", dt)
for i in range(iterations):
itr.get_next(datetime)
dt = datetime(2010, 1, 25)
itr = croniter("0 0 1 * *", dt)
for i in range(iterations):
itr.get_next()
dt = datetime(2010, 8, 25, 15, 56)
itr = croniter("*/1 * * * *", dt)
for i in range(iterations):
itr.get_prev(datetime)
dt = datetime(2010, 8, 25, 15, 0)
itr = croniter("*/1 * * * *", dt)
for i in range(iterations):
itr.get_prev(datetime)
dt = datetime(2010, 8, 25, 0, 0)
itr = croniter("*/1 * * * *", dt)
for i in range(iterations):
itr.get_prev(datetime)
dt = datetime(2010, 8, 25, 15, 56)
itr = croniter("0 0 * * sat,sun", dt)
for i in range(iterations):
itr.get_prev(datetime)
dt = datetime(2010, 2, 25)
itr = croniter("0 0 * * 7", dt)
for i in range(iterations):
itr.get_prev(datetime)
# dst regression test
tz = zoneinfo.ZoneInfo("Europe/Bucharest")
offsets = set()
dst_cron = "15 0,3 * 3 *"
dst_iters = int(2 * 31 * (iterations / 40))
dt = datetime(2010, 1, 25, tzinfo=tz)
itr = croniter(dst_cron, dt)
for i in range(dst_iters):
d = itr.get_next(datetime)
offsets.add(d.utcoffset())
itr = croniter(dst_cron, dt)
for i in range(dst_iters):
d = itr.get_prev(datetime)
offsets.add(d.utcoffset())
def test_not_long_time(self):
iterations = int(os.environ.get("CRONITER_TEST_SPEED_ITERATIONS", "40"))
globs = globals()
globs.update(locals())
t = Timer("self.run_long_test(iterations)", globals=globs)
limit = 80
ret = t.timeit(limit)
self.assertTrue(ret < limit, f"Regression in croniter speed detected ({ret} {limit}).")
if __name__ == "__main__":
unittest.main()