|
15 | 15 | import time_machine |
16 | 16 | from frequenz.channels import Broadcast, SenderError |
17 | 17 |
|
18 | | -from frequenz.sdk.timeseries import Sample |
| 18 | +from frequenz.sdk.timeseries import UNIX_EPOCH, Sample |
19 | 19 | from frequenz.sdk.timeseries._quantities import Quantity |
20 | 20 | from frequenz.sdk.timeseries._resampling import ( |
21 | 21 | DEFAULT_BUFFER_LEN_MAX, |
@@ -1195,6 +1195,82 @@ async def test_timer_is_aligned( |
1195 | 1195 | resampling_fun_mock.reset_mock() |
1196 | 1196 |
|
1197 | 1197 |
|
| 1198 | +async def test_system_clock_changed_backwards( |
| 1199 | + fake_time: time_machine.Coordinates, |
| 1200 | + source_chan: Broadcast[Sample[Quantity]], |
| 1201 | +) -> None: |
| 1202 | + """Test that the resampler is able to handle system clock changes.""" |
| 1203 | + await _advance_time(fake_time, 600) |
| 1204 | + timestamp = datetime.now(timezone.utc) |
| 1205 | + |
| 1206 | + resampling_period_s = 2 |
| 1207 | + |
| 1208 | + config = ResamplerConfig( |
| 1209 | + resampling_period=timedelta(seconds=resampling_period_s), |
| 1210 | + max_data_age_in_periods=2.0, |
| 1211 | + initial_buffer_len=5, |
| 1212 | + ) |
| 1213 | + |
| 1214 | + resampler = Resampler(config) |
| 1215 | + |
| 1216 | + source_receiver = source_chan.new_receiver() |
| 1217 | + source_sender = source_chan.new_sender() |
| 1218 | + |
| 1219 | + sink_mock = AsyncMock(spec=Sink, return_value=True) |
| 1220 | + |
| 1221 | + resampler.add_timeseries("test", source_receiver, sink_mock) |
| 1222 | + |
| 1223 | + await source_sender.send( |
| 1224 | + Sample(timestamp + timedelta(seconds=1.0), value=Quantity(1.0)) |
| 1225 | + ) |
| 1226 | + await source_sender.send( |
| 1227 | + Sample(timestamp + timedelta(seconds=2.0), value=Quantity(1.0)) |
| 1228 | + ) |
| 1229 | + |
| 1230 | + await resampler.resample(one_shot=True) |
| 1231 | + sink_mock.assert_called_once_with( |
| 1232 | + Sample( |
| 1233 | + timestamp + timedelta(seconds=resampling_period_s), |
| 1234 | + Quantity(1.0), |
| 1235 | + ) |
| 1236 | + ) |
| 1237 | + |
| 1238 | + # go back in time by 10 minutes |
| 1239 | + travaller = time_machine.travel(UNIX_EPOCH) |
| 1240 | + travaller.start() |
| 1241 | + timestamp = datetime.now(timezone.utc) |
| 1242 | + |
| 1243 | + await resampler.resample(one_shot=True) |
| 1244 | + |
| 1245 | + # The resampler will trigger two periods from now since when the resync |
| 1246 | + # happens some time is already passed and the next calculated `window_end` |
| 1247 | + # will always be more then one period in the future |
| 1248 | + sink_mock.assert_called_with( |
| 1249 | + Sample( |
| 1250 | + timestamp + timedelta(seconds=resampling_period_s), |
| 1251 | + None, |
| 1252 | + ) |
| 1253 | + ) |
| 1254 | + |
| 1255 | + # we sending some samples and run the resampler again without advancing the time |
| 1256 | + await source_sender.send( |
| 1257 | + Sample(timestamp + timedelta(seconds=2.0), value=Quantity(1.0)) |
| 1258 | + ) |
| 1259 | + await source_sender.send( |
| 1260 | + Sample(timestamp + timedelta(seconds=3.0), value=Quantity(1.0)) |
| 1261 | + ) |
| 1262 | + |
| 1263 | + await resampler.resample(one_shot=True) |
| 1264 | + sink_mock.assert_called_with( |
| 1265 | + Sample( |
| 1266 | + timestamp + timedelta(seconds=resampling_period_s), |
| 1267 | + Quantity(1.0), |
| 1268 | + ) |
| 1269 | + ) |
| 1270 | + |
| 1271 | + travaller.stop() |
| 1272 | + |
| 1273 | + |
1198 | 1274 | def _get_buffer_len(resampler: Resampler, source_receiver: Source) -> int: |
1199 | 1275 | # pylint: disable=protected-access |
1200 | 1276 | blen = resampler._resamplers[source_receiver]._helper._buffer.maxlen |
|
0 commit comments