Skip to content

Commit 6423628

Browse files
llucaxllucax
committed
Add ClocksInfo tests
Signed-off-by: Leandro Lucarella <[email protected]>
1 parent a31fdcf commit 6423628

File tree

1 file changed

+253
-0
lines changed

1 file changed

+253
-0
lines changed

‎tests/timeseries/_resampling/wall_clock_timer/test_clocksinfo.py‎

Lines changed: 253 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,253 @@
1+
# License: MIT
2+
# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
3+
4+
"""Tests for the `ClocksInfo` class."""
5+
6+
import math
7+
import re
8+
from dataclasses import dataclass
9+
from datetime import datetime, timedelta, timezone
10+
11+
import pytest
12+
13+
from frequenz.sdk.timeseries._resampling._wall_clock_timer import ClocksInfo
14+
15+
_DEFAULT_MONOTONIC_REQUESTED_SLEEP = timedelta(seconds=1.0)
16+
_DEFAULT_MONOTONIC_TIME = 1234.5
17+
_DEFAULT_WALL_CLOCK_TIME = datetime(2023, 1, 1, tzinfo=timezone.utc)
18+
_DEFAULT_MONOTONIC_ELAPSED = timedelta(seconds=1.1)
19+
_DEFAULT_WALL_CLOCK_ELAPSED = timedelta(seconds=1.2)
20+
21+
22+
@pytest.mark.parametrize(
23+
"elapsed",
24+
[timedelta(seconds=0), timedelta(seconds=-1.0)],
25+
ids=["zero", "negative"],
26+
)
27+
def test_monotonic_requested_sleep_invalid(elapsed: timedelta) -> None:
28+
"""Test monotonic_requested_sleep with invalid values."""
29+
with pytest.raises(
30+
ValueError,
31+
match=r"^monotonic_requested_sleep must be strictly positive, not "
32+
+ re.escape(repr(elapsed))
33+
+ r"$",
34+
):
35+
_ = ClocksInfo(
36+
monotonic_requested_sleep=elapsed,
37+
monotonic_time=_DEFAULT_MONOTONIC_TIME,
38+
wall_clock_time=_DEFAULT_WALL_CLOCK_TIME,
39+
monotonic_elapsed=_DEFAULT_MONOTONIC_ELAPSED,
40+
wall_clock_elapsed=_DEFAULT_WALL_CLOCK_ELAPSED,
41+
)
42+
43+
44+
@pytest.mark.parametrize(
45+
"time",
46+
[float("-inf"), float("nan"), float("inf")],
47+
)
48+
def test_monotonic_time_invalid(time: float) -> None:
49+
"""Test monotonic_time with invalid values."""
50+
with pytest.raises(
51+
ValueError,
52+
match=rf"^monotonic_time must be a number, not {re.escape(repr(time))}$",
53+
):
54+
_ = ClocksInfo(
55+
monotonic_requested_sleep=_DEFAULT_MONOTONIC_REQUESTED_SLEEP,
56+
monotonic_time=time,
57+
wall_clock_time=_DEFAULT_WALL_CLOCK_TIME,
58+
monotonic_elapsed=_DEFAULT_MONOTONIC_ELAPSED,
59+
wall_clock_elapsed=_DEFAULT_WALL_CLOCK_ELAPSED,
60+
)
61+
62+
63+
@pytest.mark.parametrize(
64+
"elapsed",
65+
[timedelta(seconds=0), timedelta(seconds=-1.0)],
66+
ids=["zero", "negative"],
67+
)
68+
def test_monotonic_elapsed_invalid(elapsed: timedelta) -> None:
69+
"""Test monotonic_elapsed with invalid values."""
70+
with pytest.raises(
71+
ValueError,
72+
match="^monotonic_elapsed must be strictly positive, not "
73+
rf"{re.escape(repr(elapsed))}$",
74+
):
75+
_ = ClocksInfo(
76+
monotonic_requested_sleep=_DEFAULT_MONOTONIC_REQUESTED_SLEEP,
77+
monotonic_time=_DEFAULT_MONOTONIC_TIME,
78+
wall_clock_time=_DEFAULT_WALL_CLOCK_TIME,
79+
monotonic_elapsed=elapsed,
80+
wall_clock_elapsed=_DEFAULT_WALL_CLOCK_ELAPSED,
81+
)
82+
83+
84+
@pytest.mark.parametrize("wall_clock_factor", [float("nan"), 2.3])
85+
def test_clocks_info_construction(wall_clock_factor: float) -> None:
86+
"""Test that ClocksInfo can be constructed and attributes are set correctly."""
87+
monotonic_requested_sleep = timedelta(seconds=1.0)
88+
monotonic_time = 1234.5
89+
wall_clock_time = datetime(2023, 1, 1, tzinfo=timezone.utc)
90+
monotonic_elapsed = timedelta(seconds=1.1)
91+
wall_clock_elapsed = timedelta(seconds=1.2)
92+
93+
info = ClocksInfo(
94+
monotonic_requested_sleep=monotonic_requested_sleep,
95+
monotonic_time=monotonic_time,
96+
wall_clock_time=wall_clock_time,
97+
monotonic_elapsed=monotonic_elapsed,
98+
wall_clock_elapsed=wall_clock_elapsed,
99+
wall_clock_factor=wall_clock_factor,
100+
)
101+
102+
assert info.monotonic_requested_sleep == monotonic_requested_sleep
103+
assert info.monotonic_time == monotonic_time
104+
assert info.wall_clock_time == wall_clock_time
105+
assert info.monotonic_elapsed == monotonic_elapsed
106+
assert info.wall_clock_elapsed == wall_clock_elapsed
107+
108+
# Check in particular that using nan explicitly is the same as using the default
109+
# We test how the default is calculated in another test
110+
if math.isnan(wall_clock_factor):
111+
assert info == ClocksInfo(
112+
monotonic_requested_sleep=monotonic_requested_sleep,
113+
monotonic_time=monotonic_time,
114+
wall_clock_time=wall_clock_time,
115+
monotonic_elapsed=monotonic_elapsed,
116+
wall_clock_elapsed=wall_clock_elapsed,
117+
)
118+
else:
119+
assert info.wall_clock_factor == wall_clock_factor
120+
121+
122+
@pytest.mark.parametrize(
123+
"requested_sleep, monotonic_elapsed, expected_drift",
124+
[
125+
(timedelta(seconds=1.0), timedelta(seconds=1.1), timedelta(seconds=0.1)),
126+
(timedelta(seconds=1.0), timedelta(seconds=0.9), timedelta(seconds=-0.1)),
127+
(timedelta(seconds=1.0), timedelta(seconds=1.0), timedelta(seconds=0.0)),
128+
],
129+
ids=["positive", "negative", "no_drift"],
130+
)
131+
def test_monotonic_drift(
132+
requested_sleep: timedelta,
133+
monotonic_elapsed: timedelta,
134+
expected_drift: timedelta,
135+
) -> None:
136+
"""Test the monotonic_drift property."""
137+
info = ClocksInfo(
138+
monotonic_requested_sleep=requested_sleep,
139+
monotonic_time=_DEFAULT_MONOTONIC_TIME,
140+
wall_clock_time=_DEFAULT_WALL_CLOCK_TIME,
141+
monotonic_elapsed=monotonic_elapsed,
142+
wall_clock_elapsed=_DEFAULT_WALL_CLOCK_ELAPSED,
143+
)
144+
assert info.monotonic_drift == pytest.approx(expected_drift)
145+
146+
147+
@pytest.mark.parametrize(
148+
"wall_clock_elapsed, monotonic_elapsed, expected_jump",
149+
[
150+
(timedelta(seconds=1.0), timedelta(seconds=2.1), timedelta(seconds=-1.1)),
151+
(timedelta(seconds=1.0), timedelta(seconds=0.19), timedelta(seconds=0.81)),
152+
(timedelta(seconds=1.0), timedelta(seconds=1.0), timedelta(seconds=0.0)),
153+
],
154+
ids=["positive", "negative", "no_jump"],
155+
)
156+
def test_wall_clock_jump(
157+
wall_clock_elapsed: timedelta,
158+
monotonic_elapsed: timedelta,
159+
expected_jump: timedelta,
160+
) -> None:
161+
"""Test the wall_clock_jump property."""
162+
info = ClocksInfo(
163+
monotonic_requested_sleep=_DEFAULT_MONOTONIC_REQUESTED_SLEEP,
164+
monotonic_time=_DEFAULT_MONOTONIC_TIME,
165+
wall_clock_time=_DEFAULT_WALL_CLOCK_TIME,
166+
monotonic_elapsed=monotonic_elapsed,
167+
wall_clock_elapsed=wall_clock_elapsed,
168+
)
169+
assert info.wall_clock_jump == pytest.approx(expected_jump)
170+
171+
172+
@dataclass(kw_only=True, frozen=True)
173+
class _TestCaseWallClockFactor:
174+
"""Test case for wall clock factor calculation."""
175+
176+
id: str
177+
monotonic_elapsed: timedelta
178+
wall_clock_elapsed: timedelta
179+
expected_factor: float
180+
181+
182+
@pytest.mark.parametrize(
183+
"case",
184+
[
185+
_TestCaseWallClockFactor(
186+
id="wall_faster",
187+
monotonic_elapsed=timedelta(seconds=1.0),
188+
wall_clock_elapsed=timedelta(seconds=1.1),
189+
expected_factor=0.9090909090909091,
190+
),
191+
_TestCaseWallClockFactor(
192+
id="wall_slower",
193+
monotonic_elapsed=timedelta(seconds=1.0),
194+
wall_clock_elapsed=timedelta(seconds=0.9),
195+
expected_factor=1.11111111111111,
196+
),
197+
_TestCaseWallClockFactor(
198+
id="in_sync",
199+
monotonic_elapsed=timedelta(seconds=1.0),
200+
wall_clock_elapsed=timedelta(seconds=1.0),
201+
expected_factor=1.0,
202+
),
203+
_TestCaseWallClockFactor(
204+
id="wall_twice_as_fast",
205+
monotonic_elapsed=timedelta(seconds=0.5),
206+
wall_clock_elapsed=timedelta(seconds=1.0),
207+
expected_factor=0.5,
208+
),
209+
],
210+
ids=lambda case: case.id,
211+
)
212+
def test_wall_clock_factor(case: _TestCaseWallClockFactor) -> None:
213+
"""Test the calculate_wall_clock_factor method with valid inputs."""
214+
info = ClocksInfo(
215+
monotonic_requested_sleep=_DEFAULT_MONOTONIC_REQUESTED_SLEEP,
216+
monotonic_time=_DEFAULT_MONOTONIC_TIME,
217+
wall_clock_time=_DEFAULT_WALL_CLOCK_TIME,
218+
monotonic_elapsed=case.monotonic_elapsed,
219+
wall_clock_elapsed=case.wall_clock_elapsed,
220+
)
221+
assert info.wall_clock_factor == pytest.approx(case.expected_factor)
222+
assert info.wall_clock_to_monotonic(case.wall_clock_elapsed) == pytest.approx(
223+
case.monotonic_elapsed
224+
)
225+
226+
227+
@pytest.mark.parametrize(
228+
"elapsed",
229+
[timedelta(seconds=0), timedelta(seconds=-1.0)],
230+
ids=["zero", "negative"],
231+
)
232+
def test_wall_clock_factor_invalid_wall_clock_elapsed(
233+
elapsed: timedelta, caplog: pytest.LogCaptureFixture
234+
) -> None:
235+
"""Test that a warning is logged when wall_clock_elapsed is zero."""
236+
expected_log = (
237+
"The monotonic clock advanced 0:00:01, but the wall clock "
238+
f"stayed still or jumped back (elapsed: {elapsed})!"
239+
)
240+
with caplog.at_level("WARNING"):
241+
info = ClocksInfo(
242+
monotonic_requested_sleep=_DEFAULT_MONOTONIC_REQUESTED_SLEEP,
243+
monotonic_time=_DEFAULT_MONOTONIC_TIME,
244+
wall_clock_time=_DEFAULT_WALL_CLOCK_TIME,
245+
monotonic_elapsed=timedelta(seconds=1.0),
246+
wall_clock_elapsed=elapsed,
247+
)
248+
assert info.wall_clock_to_monotonic(timedelta(seconds=1.0)) == timedelta(
249+
seconds=10.0
250+
)
251+
assert info.wall_clock_factor == 10.0
252+
253+
assert expected_log in caplog.text

0 commit comments

Comments
 (0)