tests/core/test_task_schedule.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """Tests for tasks.schedule (core 1e Phase 4): cron/every/at cadences.16 17 No I/O, no server: the three spec classes and ``TaskCadence``. Cron matches are18 checked by decoding the returned epoch back to a local ``datetime`` and19 asserting the field values, so the tests are timezone-agnostic (the parser20 evaluates in local time, like system cron).21 """22 23 from __future__ import annotations24 25 from datetime import datetime26 27 import pytest28 29 from genro_asgi.tasks.schedule import AtSpec, CronSpec, EverySpec, TaskCadence30 31 32 class TestEverySpec:33 """``"<n><unit>"`` -> seconds, with strict validation."""34 35 def test_units(self) -> None:36 assert EverySpec("30s").seconds == 3037 assert EverySpec("15m").seconds == 15 * 6038 assert EverySpec("2h").seconds == 2 * 360039 assert EverySpec("1d").seconds == 8640040 41 def test_whitespace_tolerated(self) -> None:42 assert EverySpec(" 45s ").seconds == 4543 44 @pytest.mark.parametrize("bad", ["", "s", "10", "10x", "1.5h", "-5m"])45 def test_malformed_raises(self, bad: str) -> None:46 with pytest.raises(ValueError):47 EverySpec(bad)48 49 def test_zero_interval_raises(self) -> None:50 with pytest.raises(ValueError, match="zero interval"):51 EverySpec("0s")52 53 def test_next_run_adds_the_interval(self) -> None:54 assert EverySpec("15m").get_next_run(1_000_000.0) == 1_000_000.0 + 15 * 6055 56 57 class TestAtSpec:58 """A list of ISO timestamps -> sorted epoch seconds."""59 60 def test_sorted_epochs(self) -> None:61 got = AtSpec(["2030-01-02T00:00:00", "2030-01-01T00:00:00"]).instants62 assert got == sorted(got)63 assert len(got) == 264 65 def test_empty_list(self) -> None:66 assert AtSpec([]).instants == []67 68 def test_not_a_list_raises(self) -> None:69 with pytest.raises(ValueError, match="want a list"):70 AtSpec("2030-01-01T00:00:00")71 72 def test_bad_timestamp_raises(self) -> None:73 with pytest.raises(ValueError, match="invalid at timestamp"):74 AtSpec(["not-a-date"])75 76 77 class TestCronSpec:78 """5-field cron parsing with ``* , - /`` and the dom/dow OR rule."""79 80 def test_field_count_enforced(self) -> None:81 with pytest.raises(ValueError, match="5 fields"):82 CronSpec("* * * *")83 84 def test_wildcards(self) -> None:85 spec = CronSpec("* * * * *")86 assert spec.minutes == set(range(60))87 assert spec.hours == set(range(24))88 assert not spec.dom_restricted and not spec.dow_restricted89 90 def test_list_range_step(self) -> None:91 spec = CronSpec("0,30 9-17 * * *")92 assert spec.minutes == {0, 30}93 assert spec.hours == set(range(9, 18))94 95 def test_step_opens_range_to_top(self) -> None:96 # "*/15" and "0/15" both -> {0,15,30,45}97 assert CronSpec("*/15 * * * *").minutes == {0, 15, 30, 45}98 assert CronSpec("0/15 * * * *").minutes == {0, 15, 30, 45}99 100 def test_dow_7_folds_onto_sunday(self) -> None:101 assert CronSpec("0 0 * * 7").weekdays == {0}102 103 @pytest.mark.parametrize("bad", ["61 * * * *", "* 24 * * *", "* * 0 * *", "* * * 13 *"])104 def test_out_of_range_raises(self, bad: str) -> None:105 with pytest.raises(ValueError, match="out of range"):106 CronSpec(bad)107 108 def test_bad_step_raises(self) -> None:109 with pytest.raises(ValueError, match="invalid cron step"):110 CronSpec("*/0 * * * *")111 112 def test_next_run_daily(self) -> None:113 # every day at 07:30 — the next match is a 07:30 local instant114 after = datetime(2030, 6, 15, 8, 0, 0).timestamp() # past 07:30 today115 got = datetime.fromtimestamp(CronSpec("30 7 * * *").get_next_run(after))116 assert (got.hour, got.minute) == (7, 30)117 assert got.date() == datetime(2030, 6, 16).date() # -> tomorrow118 119 def test_next_run_strictly_after(self) -> None:120 base = datetime(2030, 6, 15, 7, 30, 0)121 got = CronSpec("30 7 * * *").get_next_run(base.timestamp())122 assert got > base.timestamp() # never returns "now"123 124 def test_dom_or_dow_when_both_restricted(self) -> None:125 # "0 0 13 * 5" matches day-13 OR any Friday (system-cron OR rule)126 spec = CronSpec("0 0 13 * 5")127 assert spec.dom_restricted and spec.dow_restricted128 got = datetime.fromtimestamp(spec.get_next_run(datetime(2030, 6, 1).timestamp()))129 assert got.day == 13 or (got.weekday() + 1) % 7 == 5130 131 def test_impossible_date_raises(self) -> None:132 with pytest.raises(ValueError, match="no occurrence"):133 CronSpec("0 0 31 2 *").get_next_run(datetime(2030, 1, 1).timestamp())134 135 136 class TestTaskCadence:137 """The kind picks the spec class; the exhausted ``at`` answers None."""138 139 def test_every_adds_interval(self) -> None:140 now = 1_000_000.0141 assert TaskCadence("every", "15m").get_next_run(now) == now + 15 * 60142 143 def test_cron_delegates(self) -> None:144 now = datetime(2030, 6, 15, 8, 0, 0).timestamp()145 got = TaskCadence("cron", "30 7 * * *").get_next_run(now)146 assert got is not None and got > now147 148 def test_at_returns_first_future(self) -> None:149 now = datetime(2030, 6, 15).timestamp()150 spec = ["2030-06-14T00:00:00", "2030-06-16T00:00:00"]151 assert TaskCadence("at", spec).get_next_run(now) == datetime(2030, 6, 16).timestamp()152 153 def test_at_exhausted_returns_none(self) -> None:154 now = datetime(2030, 6, 15).timestamp()155 assert TaskCadence("at", ["2030-06-14T00:00:00"]).get_next_run(now) is None156 157 def test_unknown_kind_raises(self) -> None:158 with pytest.raises(ValueError, match="unknown schedule kind"):159 TaskCadence("weekly", "x")160 161 def test_malformed_spec_raises_at_construction(self) -> None:162 with pytest.raises(ValueError, match="invalid every spec"):163 TaskCadence("every", "10x")