Skip to content

Commit 4e00078

Browse files
MarenzMarenz
committed
Add tests for newly added functions
Signed-off-by: Mathias L. Baumann <[email protected]>
1 parent 677fbbc commit 4e00078

File tree

2 files changed

+278
-0
lines changed

2 files changed

+278
-0
lines changed

‎pyproject.toml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -95,6 +95,7 @@ dev-pytest = [
9595
"pytest-mock == 3.14.0",
9696
"pytest-asyncio == 0.24.0",
9797
"async-solipsism == 0.7",
98+
"time-machine == 2.15.0",
9899
"hypothesis == 6.116.0",
99100
"frequenz-client-dispatch[cli]",
100101
]

‎tests/test_dispatch.py‎

Lines changed: 277 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,277 @@
1+
# License: MIT
2+
# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
3+
4+
"""Tests for the Dispatch class methods using pytest parametrization."""
5+
6+
from dataclasses import replace
7+
from datetime import datetime, timedelta, timezone
8+
9+
import pytest
10+
import time_machine
11+
12+
from frequenz.client.common.microgrid.components import ComponentCategory
13+
from frequenz.client.dispatch.recurrence import Frequency, RecurrenceRule, Weekday
14+
from frequenz.client.dispatch.types import Dispatch, RunningState
15+
16+
# Define a fixed current time for testing to avoid issues with datetime.now()
17+
CURRENT_TIME = datetime(2023, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
18+
19+
20+
@pytest.fixture
21+
def dispatch_base() -> Dispatch:
22+
"""Fixture to create a base Dispatch instance."""
23+
return Dispatch(
24+
id=1,
25+
type="TypeA",
26+
start_time=CURRENT_TIME,
27+
duration=timedelta(minutes=20),
28+
selector=[ComponentCategory.BATTERY],
29+
active=True,
30+
dry_run=False,
31+
payload={},
32+
recurrence=RecurrenceRule(),
33+
create_time=CURRENT_TIME - timedelta(hours=1),
34+
update_time=CURRENT_TIME - timedelta(minutes=30),
35+
)
36+
37+
38+
@time_machine.travel(CURRENT_TIME)
39+
@pytest.mark.parametrize(
40+
"dispatch_type, requested_type, active, start_time_offset, duration, expected_state",
41+
[
42+
# Dispatch type does not match the requested type
43+
(
44+
"TypeA",
45+
"TypeB",
46+
True,
47+
timedelta(minutes=-10),
48+
timedelta(minutes=20),
49+
RunningState.DIFFERENT_TYPE,
50+
),
51+
# Dispatch is inactive
52+
(
53+
"TypeA1",
54+
"TypeA1",
55+
False,
56+
timedelta(minutes=-10),
57+
timedelta(minutes=20),
58+
RunningState.STOPPED,
59+
),
60+
# Current time is before the start time
61+
(
62+
"TypeA2",
63+
"TypeA2",
64+
True,
65+
timedelta(minutes=10),
66+
timedelta(minutes=20),
67+
RunningState.STOPPED,
68+
),
69+
# Dispatch with infinite duration
70+
(
71+
"TypeA3",
72+
"TypeA3",
73+
True,
74+
timedelta(minutes=-10),
75+
None,
76+
RunningState.RUNNING,
77+
),
78+
# Dispatch is currently running
79+
(
80+
"TypeA4",
81+
"TypeA4",
82+
True,
83+
timedelta(minutes=-10),
84+
timedelta(minutes=20),
85+
RunningState.RUNNING,
86+
),
87+
# Dispatch duration has passed
88+
(
89+
"TypeA5",
90+
"TypeA5",
91+
True,
92+
timedelta(minutes=-30),
93+
timedelta(minutes=20),
94+
RunningState.STOPPED,
95+
),
96+
],
97+
)
98+
# pylint: disable=too-many-arguments,too-many-positional-arguments
99+
def test_dispatch_running(
100+
dispatch_base: Dispatch,
101+
dispatch_type: str,
102+
requested_type: str,
103+
active: bool,
104+
start_time_offset: timedelta,
105+
duration: timedelta | None,
106+
expected_state: RunningState,
107+
) -> None:
108+
"""Test the running method of the Dispatch class."""
109+
dispatch = replace(
110+
dispatch_base,
111+
type=dispatch_type,
112+
start_time=CURRENT_TIME + start_time_offset,
113+
duration=duration,
114+
active=active,
115+
)
116+
117+
assert dispatch.running(requested_type) == expected_state
118+
119+
120+
@time_machine.travel(CURRENT_TIME)
121+
@pytest.mark.parametrize(
122+
"active, duration, start_time_offset, expected_until_offset",
123+
[
124+
# Dispatch is inactive
125+
(False, timedelta(minutes=20), timedelta(minutes=-10), None),
126+
# Dispatch with infinite duration (no duration)
127+
(True, None, timedelta(minutes=-10), None),
128+
# Current time is before the start time
129+
(True, timedelta(minutes=20), timedelta(minutes=10), timedelta(minutes=30)),
130+
# Dispatch is currently running
131+
(
132+
True,
133+
timedelta(minutes=20),
134+
timedelta(minutes=-10),
135+
timedelta(minutes=10),
136+
),
137+
],
138+
)
139+
def test_dispatch_until(
140+
dispatch_base: Dispatch,
141+
active: bool,
142+
duration: timedelta | None,
143+
start_time_offset: timedelta,
144+
expected_until_offset: timedelta | None,
145+
) -> None:
146+
"""Test the until property of the Dispatch class."""
147+
start_time = CURRENT_TIME + start_time_offset
148+
dispatch = replace(
149+
dispatch_base,
150+
active=active,
151+
duration=duration,
152+
start_time=start_time,
153+
)
154+
155+
if duration is None:
156+
with pytest.raises(ValueError):
157+
_ = dispatch.until
158+
return
159+
160+
expected_until = (
161+
CURRENT_TIME + expected_until_offset
162+
if expected_until_offset is not None
163+
else None
164+
)
165+
166+
assert dispatch.until == expected_until
167+
168+
169+
@time_machine.travel(CURRENT_TIME)
170+
@pytest.mark.parametrize(
171+
"recurrence, duration, start_time_offset, expected_next_run_offset",
172+
[
173+
# No recurrence and start time in the past
174+
(RecurrenceRule(), timedelta(minutes=20), timedelta(minutes=-10), None),
175+
# No recurrence and start time in the future
176+
(
177+
RecurrenceRule(),
178+
timedelta(minutes=20),
179+
timedelta(minutes=10),
180+
timedelta(minutes=10),
181+
),
182+
# Daily recurrence
183+
(
184+
RecurrenceRule(frequency=Frequency.DAILY, interval=1),
185+
timedelta(minutes=20),
186+
timedelta(minutes=-10),
187+
timedelta(days=1, minutes=-10),
188+
),
189+
# Weekly recurrence on Monday
190+
(
191+
RecurrenceRule(
192+
frequency=Frequency.WEEKLY, byweekdays=[Weekday.MONDAY], interval=1
193+
),
194+
timedelta(minutes=20),
195+
timedelta(minutes=-10),
196+
None, # We'll compute expected_next_run inside the test
197+
),
198+
],
199+
)
200+
def test_dispatch_next_run(
201+
dispatch_base: Dispatch,
202+
recurrence: RecurrenceRule,
203+
duration: timedelta | None,
204+
start_time_offset: timedelta,
205+
expected_next_run_offset: timedelta | None,
206+
) -> None:
207+
"""Test the next_run property of the Dispatch class."""
208+
start_time = CURRENT_TIME + start_time_offset
209+
dispatch = replace(
210+
dispatch_base,
211+
start_time=start_time,
212+
duration=duration,
213+
recurrence=recurrence,
214+
)
215+
216+
if recurrence.frequency == Frequency.WEEKLY:
217+
# Compute the next run based on the recurrence rule
218+
expected_next_run = recurrence.prepare(start_time).after(
219+
CURRENT_TIME, inc=False
220+
)
221+
elif expected_next_run_offset is not None:
222+
expected_next_run = CURRENT_TIME + expected_next_run_offset
223+
else:
224+
expected_next_run = None
225+
226+
assert dispatch.next_run == expected_next_run
227+
228+
229+
@time_machine.travel(CURRENT_TIME)
230+
@pytest.mark.parametrize(
231+
"after_offset, recurrence, duration, expected_next_run_after_offset",
232+
[
233+
# No recurrence
234+
(timedelta(minutes=10), RecurrenceRule(), timedelta(minutes=20), None),
235+
# Weekly recurrence, after current time
236+
(
237+
timedelta(days=2),
238+
RecurrenceRule(
239+
frequency=Frequency.WEEKLY, byweekdays=[Weekday.MONDAY], interval=1
240+
),
241+
timedelta(minutes=20),
242+
None, # We'll compute expected_next_run_after inside the test
243+
),
244+
# Daily recurrence
245+
(
246+
timedelta(minutes=10),
247+
RecurrenceRule(frequency=Frequency.DAILY, interval=1),
248+
timedelta(minutes=20),
249+
timedelta(days=1),
250+
),
251+
],
252+
)
253+
def test_dispatch_next_run_after(
254+
dispatch_base: Dispatch,
255+
after_offset: timedelta,
256+
recurrence: RecurrenceRule,
257+
duration: timedelta | None,
258+
expected_next_run_after_offset: timedelta | None,
259+
) -> None:
260+
"""Test the next_run_after method of the Dispatch class."""
261+
after = CURRENT_TIME + after_offset
262+
dispatch = replace(
263+
dispatch_base,
264+
recurrence=recurrence,
265+
duration=duration,
266+
)
267+
268+
if recurrence.frequency == Frequency.WEEKLY:
269+
expected_next_run_after = recurrence.prepare(dispatch.start_time).after(
270+
after, inc=True
271+
)
272+
elif expected_next_run_after_offset is not None:
273+
expected_next_run_after = CURRENT_TIME + expected_next_run_after_offset
274+
else:
275+
expected_next_run_after = None
276+
277+
assert dispatch.next_run_after(after) == expected_next_run_after

0 commit comments

Comments
 (0)