wehub-resource-sync
1363 lines
53 KiB
Python
1363 lines
53 KiB
Python
import asyncio
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
from ray._common.test_utils import async_wait_for_condition
|
|
from ray.serve._private.common import TimeStampedValue
|
|
from ray.serve._private.metrics_utils import (
|
|
InMemoryMetricsStore,
|
|
MetricsPusher,
|
|
aggregate_timeseries,
|
|
merge_instantaneous_total,
|
|
merge_timeseries_dicts,
|
|
time_weighted_average,
|
|
)
|
|
from ray.serve._private.test_utils import MockAsyncTimer
|
|
from ray.serve.config import AggregationFunction
|
|
|
|
|
|
class TestMetricsPusher:
|
|
@pytest.mark.asyncio
|
|
async def test_no_tasks(self):
|
|
"""Test that a metrics pusher can be started with zero tasks.
|
|
|
|
After a task is registered, it should work.
|
|
"""
|
|
val = 0
|
|
|
|
def inc():
|
|
nonlocal val
|
|
val += 1
|
|
|
|
metrics_pusher = MetricsPusher()
|
|
metrics_pusher.start()
|
|
assert len(metrics_pusher._tasks) == 0
|
|
|
|
metrics_pusher.register_or_update_task("inc", inc, 0.01)
|
|
|
|
async_wait_for_condition(lambda: val > 0, timeout=10)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_basic(self):
|
|
timer = MockAsyncTimer(0)
|
|
state = {"val": 0}
|
|
|
|
def task(s):
|
|
s["val"] += 1
|
|
|
|
metrics_pusher = MetricsPusher(async_sleep=timer.sleep)
|
|
metrics_pusher.start()
|
|
|
|
metrics_pusher.register_or_update_task("basic", lambda: task(state), 0.5)
|
|
for i in range(20):
|
|
await async_wait_for_condition(
|
|
lambda: timer.num_sleepers() == 1, retry_interval_ms=1
|
|
)
|
|
timer.advance(0.5)
|
|
await asyncio.sleep(0)
|
|
assert state["val"] == i + 1
|
|
|
|
await metrics_pusher.graceful_shutdown()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multiple_tasks(self):
|
|
timer = MockAsyncTimer(0)
|
|
|
|
state = {"A": 0, "B": 0, "C": 0}
|
|
|
|
def task(key, s):
|
|
s[key] += 1
|
|
|
|
metrics_pusher = MetricsPusher(async_sleep=timer.sleep)
|
|
metrics_pusher.start()
|
|
|
|
# Each task interval is different, and they don't divide each other.
|
|
metrics_pusher.register_or_update_task("A", lambda: task("A", state), 0.2)
|
|
metrics_pusher.register_or_update_task("B", lambda: task("B", state), 0.3)
|
|
metrics_pusher.register_or_update_task("C", lambda: task("C", state), 0.5)
|
|
|
|
times = sorted(
|
|
[(0, None, None)]
|
|
+ [(0.2 * (i + 1), "A", i + 2) for i in range(15)]
|
|
+ [(0.3 * (i + 1), "B", i + 2) for i in range(10)]
|
|
+ [(0.5 * (i + 1), "C", i + 2) for i in range(6)]
|
|
)
|
|
advances = [(j[0] - i[0], j[1], j[2]) for i, j in zip(times[:-1], times[1:])]
|
|
|
|
for t, key, expected in advances:
|
|
await async_wait_for_condition(
|
|
lambda: timer.num_sleepers() == 3, retry_interval_ms=1, timeout=1
|
|
)
|
|
timer.advance(t + 0.001)
|
|
await async_wait_for_condition(
|
|
lambda: state[key] == expected, retry_interval_ms=1, timeout=1
|
|
)
|
|
|
|
# At 7 seconds, tasks A, B, C should have executed 16, 11, and 7
|
|
# times respectively.
|
|
assert state["A"] == 16
|
|
assert state["B"] == 11
|
|
assert state["C"] == 7
|
|
await metrics_pusher.graceful_shutdown()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_task(self):
|
|
_start = {"A": 0}
|
|
timer = MockAsyncTimer(_start["A"])
|
|
state = {"A": 0, "B": 0}
|
|
|
|
def f(s):
|
|
s["A"] += 1
|
|
|
|
# Start metrics pusher and register task() with interval 1s.
|
|
# After (fake) 10s, the task should have executed 10 times
|
|
metrics_pusher = MetricsPusher(async_sleep=timer.sleep)
|
|
metrics_pusher.start()
|
|
|
|
# Give the metrics pusher thread opportunity to execute task
|
|
# The only thing that should be moving the timer forward is
|
|
# the metrics pusher thread. So if the timer has reached 11,
|
|
# the task should have at least executed 10 times.
|
|
metrics_pusher.register_or_update_task("my_task", lambda: f(state), 1)
|
|
for i in range(20):
|
|
await async_wait_for_condition(
|
|
lambda: timer.num_sleepers() == 1, retry_interval_ms=1
|
|
)
|
|
timer.advance(1)
|
|
await asyncio.sleep(0)
|
|
assert state["A"] == i + 1
|
|
|
|
def new_f(s):
|
|
s["B"] += 1
|
|
|
|
# Re-register new_f() with interval 50s.
|
|
metrics_pusher.register_or_update_task("my_task", lambda: new_f(state), 50)
|
|
for i in range(20):
|
|
await async_wait_for_condition(
|
|
lambda: timer.num_sleepers() == 1, retry_interval_ms=1, timeout=1
|
|
)
|
|
timer.advance(50)
|
|
|
|
await asyncio.sleep(0)
|
|
assert state["B"] == i + 1
|
|
|
|
await metrics_pusher.graceful_shutdown()
|
|
|
|
|
|
def assert_timeseries_equal(actual, expected):
|
|
assert len(actual) == len(
|
|
expected
|
|
), f"Length mismatch: {len(actual)} vs {len(expected)}"
|
|
for i, (a, e) in enumerate(zip(actual, expected)):
|
|
assert (
|
|
# c_round is used in the Cython implementation, so we need to use a tolerance
|
|
abs(a.timestamp - e.timestamp)
|
|
< 1e-4
|
|
), f"Timestamp mismatch at {i}: {a.timestamp} vs {e.timestamp}"
|
|
assert a.value == e.value, f"Value mismatch at {i}: {a.value} vs {e.value}"
|
|
|
|
|
|
class TestInMemoryMetricsStore:
|
|
def test_basics(self):
|
|
s = InMemoryMetricsStore()
|
|
s.add_metrics_point({"m1": 1}, timestamp=1)
|
|
s.add_metrics_point({"m1": 2}, timestamp=2)
|
|
assert s.aggregate_avg(["m1"]) == (1.5, 1)
|
|
assert s.get_latest("m1") == 2
|
|
|
|
def test_out_of_order_insert(self):
|
|
s = InMemoryMetricsStore()
|
|
s.add_metrics_point({"m1": 1}, timestamp=1)
|
|
s.add_metrics_point({"m1": 5}, timestamp=5)
|
|
s.add_metrics_point({"m1": 3}, timestamp=3)
|
|
s.add_metrics_point({"m1": 2}, timestamp=2)
|
|
s.add_metrics_point({"m1": 4}, timestamp=4)
|
|
assert s.aggregate_avg(["m1"]) == (3, 1)
|
|
|
|
def test_window_start_timestamp(self):
|
|
s = InMemoryMetricsStore()
|
|
assert s.aggregate_avg(["m1"]) == (None, 0)
|
|
|
|
s.add_metrics_point({"m1": 1}, timestamp=2)
|
|
assert s.aggregate_avg(["m1"]) == (1, 1)
|
|
s.prune_keys_and_compact_data(10)
|
|
assert s.aggregate_avg(["m1"]) == (None, 0)
|
|
|
|
def test_multiple_metrics(self):
|
|
s = InMemoryMetricsStore()
|
|
s.add_metrics_point({"m1": 1, "m2": -1}, timestamp=1)
|
|
s.add_metrics_point({"m1": 2, "m2": -2}, timestamp=2)
|
|
assert s.aggregate_avg(["m1"]) == (1.5, 1)
|
|
assert s.aggregate_avg(["m2"]) == (-1.5, 1)
|
|
assert s.aggregate_avg(["m1", "m2"]) == (0, 2)
|
|
|
|
def test_empty_key_mix(self):
|
|
s = InMemoryMetricsStore()
|
|
s.add_metrics_point({"m1": 1}, timestamp=1)
|
|
assert s.aggregate_avg(["m1", "m2"]) == (1, 1)
|
|
assert s.aggregate_avg(["m2"]) == (None, 0)
|
|
|
|
def test_prune_keys_and_compact_data(self):
|
|
s = InMemoryMetricsStore()
|
|
s.add_metrics_point({"m1": 1, "m2": 2, "m3": 8, "m4": 5}, timestamp=1)
|
|
s.add_metrics_point({"m1": 2, "m2": 3, "m3": 8}, timestamp=2)
|
|
s.add_metrics_point({"m1": 2, "m2": 5}, timestamp=3)
|
|
s.prune_keys_and_compact_data(1.1)
|
|
assert set(s.data) == {"m1", "m2", "m3"}
|
|
assert len(s.data["m1"]) == 2 and s.data["m1"] == s._get_datapoints("m1", 1.1)
|
|
assert len(s.data["m2"]) == 2 and s.data["m2"] == s._get_datapoints("m2", 1.1)
|
|
assert len(s.data["m3"]) == 1 and s.data["m3"] == s._get_datapoints("m3", 1.1)
|
|
|
|
|
|
class TestAggregateTimeseries:
|
|
def test_aggregate_timeseries_empty(self):
|
|
assert aggregate_timeseries([], AggregationFunction.MEAN) is None
|
|
assert aggregate_timeseries([], AggregationFunction.MAX) is None
|
|
assert aggregate_timeseries([], AggregationFunction.MIN) is None
|
|
|
|
def test_aggregate_timeseries_mean(self):
|
|
assert (
|
|
aggregate_timeseries([TimeStampedValue(1.0, 5.0)], AggregationFunction.MEAN)
|
|
== 5.0
|
|
)
|
|
assert (
|
|
aggregate_timeseries(
|
|
[TimeStampedValue(1.0, 5.0), TimeStampedValue(2.0, 10.0)],
|
|
AggregationFunction.MEAN,
|
|
)
|
|
== 7.5
|
|
)
|
|
assert (
|
|
aggregate_timeseries(
|
|
[
|
|
TimeStampedValue(1.0, 5.0),
|
|
TimeStampedValue(2.0, 10.0),
|
|
TimeStampedValue(3.0, 15.0),
|
|
],
|
|
AggregationFunction.MEAN,
|
|
)
|
|
== 10.0
|
|
)
|
|
|
|
def test_aggregate_timeseries_max(self):
|
|
assert (
|
|
aggregate_timeseries([TimeStampedValue(1.0, 5.0)], AggregationFunction.MAX)
|
|
== 5.0
|
|
)
|
|
assert (
|
|
aggregate_timeseries(
|
|
[TimeStampedValue(1.0, 5.0), TimeStampedValue(2.0, 10.0)],
|
|
AggregationFunction.MAX,
|
|
)
|
|
== 10.0
|
|
)
|
|
assert (
|
|
aggregate_timeseries(
|
|
[
|
|
TimeStampedValue(1.0, 5.0),
|
|
TimeStampedValue(2.0, 10.0),
|
|
TimeStampedValue(3.0, 15.0),
|
|
],
|
|
AggregationFunction.MAX,
|
|
)
|
|
== 15.0
|
|
)
|
|
|
|
def test_aggregate_timeseries_min(self):
|
|
assert (
|
|
aggregate_timeseries([TimeStampedValue(1.0, 5.0)], AggregationFunction.MIN)
|
|
== 5.0
|
|
)
|
|
assert (
|
|
aggregate_timeseries(
|
|
[TimeStampedValue(1.0, 5.0), TimeStampedValue(2.0, 10.0)],
|
|
AggregationFunction.MIN,
|
|
)
|
|
== 5.0
|
|
)
|
|
assert (
|
|
aggregate_timeseries(
|
|
[
|
|
TimeStampedValue(1.0, 5.0),
|
|
TimeStampedValue(2.0, 10.0),
|
|
TimeStampedValue(3.0, 15.0),
|
|
],
|
|
AggregationFunction.MIN,
|
|
)
|
|
== 5.0
|
|
)
|
|
|
|
def test_exclude_early_partial_period_when_aligned_start_equals_last_timestamp(
|
|
self,
|
|
):
|
|
"""Edge case: aligned_start == merged_timeseries[-1].timestamp.
|
|
|
|
When one series has only one point and it's the latest, aligned_start equals
|
|
the last merged timestamp. window_start must exclude the partial period
|
|
(the earlier point from the other series).
|
|
"""
|
|
# s1 has one point at 100.1, s2 has one point at 100.2 (the latest)
|
|
# merged = [(100.1, 10), (100.2, 30)]; aligned_start = 100.2 = last timestamp
|
|
series1 = [TimeStampedValue(100.1, 10.0)]
|
|
series2 = [TimeStampedValue(100.2, 20.0)]
|
|
|
|
merged = merge_instantaneous_total([series1, series2])
|
|
assert merged[-1].timestamp == 100.2
|
|
aligned_start = max(ts[0].timestamp for ts in [series1, series2])
|
|
assert aligned_start == 100.2 # Edge case: aligned_start == last timestamp
|
|
|
|
last_window_s = 1.0
|
|
|
|
# With window_start=aligned_start: excludes 100.1, only [100.2, 101.2) at value 30
|
|
result_with_window = aggregate_timeseries(
|
|
merged,
|
|
AggregationFunction.MEAN,
|
|
last_window_s=last_window_s,
|
|
window_start=aligned_start,
|
|
)
|
|
assert result_with_window == 30.0
|
|
|
|
# Without window_start: includes 100.1 (partial period), gives lower average
|
|
result_without_window = aggregate_timeseries(
|
|
merged,
|
|
AggregationFunction.MEAN,
|
|
last_window_s=last_window_s,
|
|
window_start=None,
|
|
)
|
|
assert result_without_window < 30.0 # Includes the 10 from 100.1-100.2
|
|
|
|
|
|
class TestInstantaneousMerge:
|
|
"""Test the new instantaneous merge functionality."""
|
|
|
|
def test_merge_instantaneous_total_empty(self):
|
|
"""Test merge_instantaneous_total with empty input."""
|
|
result = merge_instantaneous_total([])
|
|
assert result == []
|
|
|
|
result = merge_instantaneous_total([[], []])
|
|
assert result == []
|
|
|
|
def test_merge_instantaneous_total_single_replica(self):
|
|
"""Test merge_instantaneous_total with single replica."""
|
|
series = [
|
|
TimeStampedValue(1.0, 5.0),
|
|
TimeStampedValue(2.0, 7.0),
|
|
TimeStampedValue(3.0, 3.0),
|
|
]
|
|
result = merge_instantaneous_total([series])
|
|
|
|
expected = [
|
|
TimeStampedValue(1.0, 5.0),
|
|
TimeStampedValue(2.0, 7.0),
|
|
TimeStampedValue(3.0, 3.0),
|
|
]
|
|
assert_timeseries_equal(result, expected)
|
|
|
|
def test_merge_instantaneous_total_two_replicas(self):
|
|
"""Test merge_instantaneous_total with two replicas."""
|
|
series1 = [
|
|
TimeStampedValue(1.0, 5.0),
|
|
TimeStampedValue(3.0, 7.0),
|
|
]
|
|
series2 = [
|
|
TimeStampedValue(2.0, 3.0),
|
|
TimeStampedValue(4.0, 1.0),
|
|
]
|
|
result = merge_instantaneous_total([series1, series2])
|
|
|
|
# Expected: t=1.0: +5 (total=5), t=2.0: +3 (total=8), t=3.0: +2 (total=10), t=4.0: -2 (total=8)
|
|
expected = [
|
|
TimeStampedValue(1.0, 5.0),
|
|
TimeStampedValue(2.0, 8.0),
|
|
TimeStampedValue(3.0, 10.0),
|
|
TimeStampedValue(4.0, 8.0),
|
|
]
|
|
assert_timeseries_equal(result, expected)
|
|
|
|
def test_merge_instantaneous_total_complex_scenario(self):
|
|
"""Test complex scenario matching the autoscaling example."""
|
|
# r1: starts at 5 (t=0.2), changes to 7 (t=0.8), then 6 (t=1.5)
|
|
series1 = [
|
|
TimeStampedValue(0.2, 5.0),
|
|
TimeStampedValue(0.8, 7.0),
|
|
TimeStampedValue(1.5, 6.0),
|
|
]
|
|
# r2: starts at 3 (t=0.1), changes to 4 (t=0.9), then 8 (t=1.2)
|
|
series2 = [
|
|
TimeStampedValue(0.1, 3.0),
|
|
TimeStampedValue(0.9, 4.0),
|
|
TimeStampedValue(1.2, 8.0),
|
|
]
|
|
result = merge_instantaneous_total([series1, series2])
|
|
|
|
expected = [
|
|
TimeStampedValue(0.1, 3.0), # r2 starts
|
|
TimeStampedValue(0.2, 8.0), # r1 starts: 3+5=8
|
|
TimeStampedValue(0.8, 10.0), # r1 changes: 8+(7-5)=10
|
|
TimeStampedValue(0.9, 11.0), # r2 changes: 10+(4-3)=11
|
|
TimeStampedValue(1.2, 15.0), # r2 changes: 11+(8-4)=15
|
|
TimeStampedValue(1.5, 14.0), # r1 changes: 15+(6-7)=14
|
|
]
|
|
assert_timeseries_equal(result, expected)
|
|
|
|
def test_time_weighted_average_empty(self):
|
|
"""Test time_weighted_average with empty series."""
|
|
result = time_weighted_average([], 0.0, 1.0)
|
|
assert result is None
|
|
|
|
def test_time_weighted_average_no_overlap(self):
|
|
"""Test time_weighted_average with no data overlap."""
|
|
series = [TimeStampedValue(2.0, 5.0)]
|
|
result = time_weighted_average(series, 0.0, 1.0)
|
|
assert result == 0.0 # Default value before first point
|
|
|
|
def test_time_weighted_average_constant_value(self):
|
|
"""Test time_weighted_average with constant value."""
|
|
series = [TimeStampedValue(0.5, 10.0)]
|
|
result = time_weighted_average(series, 1.0, 2.0)
|
|
assert result == 10.0
|
|
|
|
def test_time_weighted_average_step_function(self):
|
|
"""Test time_weighted_average with step function."""
|
|
series = [
|
|
TimeStampedValue(0.0, 5.0),
|
|
TimeStampedValue(1.0, 10.0),
|
|
TimeStampedValue(2.0, 15.0),
|
|
]
|
|
# Average over [0.5, 1.5): 0.5s at value 5, 0.5s at value 10
|
|
result = time_weighted_average(series, 0.5, 1.5)
|
|
expected = (5.0 * 0.5 + 10.0 * 0.5) / 1.0
|
|
assert abs(result - expected) < 1e-10
|
|
|
|
def test_time_weighted_average_none_window_start(self):
|
|
"""Test time_weighted_average with None window_start."""
|
|
series = [
|
|
TimeStampedValue(1.0, 5.0),
|
|
TimeStampedValue(2.0, 10.0),
|
|
TimeStampedValue(3.0, 15.0),
|
|
]
|
|
# Should use full series from start (t=1.0) to window_end (t=2.5)
|
|
result = time_weighted_average(series, None, 2.5)
|
|
# 1.0s at value 5 (from 1.0 to 2.0), 0.5s at value 10 (from 2.0 to 2.5)
|
|
expected = (5.0 * 1.0 + 10.0 * 0.5) / 1.5
|
|
assert abs(result - expected) < 1e-10
|
|
|
|
def test_time_weighted_average_none_window_end(self):
|
|
"""Test time_weighted_average with None window_end."""
|
|
series = [
|
|
TimeStampedValue(1.0, 5.0),
|
|
TimeStampedValue(2.0, 10.0),
|
|
TimeStampedValue(3.0, 15.0),
|
|
]
|
|
# Should use from window_start (t=1.5) to end of series (t=3.0+1.0=4.0)
|
|
result = time_weighted_average(series, 1.5, None)
|
|
# 0.5s at value 5 (from 1.5 to 2.0), 1.0s at value 10 (from 2.0 to 3.0), 1.0s at value 15 (from 3.0 to 4.0)
|
|
expected = (5.0 * 0.5 + 10.0 * 1.0 + 15.0 * 1.0) / 2.5
|
|
assert abs(result - expected) < 1e-10
|
|
|
|
def test_time_weighted_average_both_none(self):
|
|
"""Test time_weighted_average with both window_start and window_end None."""
|
|
series = [
|
|
TimeStampedValue(1.0, 5.0),
|
|
TimeStampedValue(2.0, 10.0),
|
|
TimeStampedValue(3.0, 15.0),
|
|
]
|
|
# Should use full series from t=1.0 to t=3.0+1.0=4.0
|
|
result = time_weighted_average(series, None, None)
|
|
# 1.0s at value 5, 1.0s at value 10, 1.0s at value 15
|
|
expected = (5.0 * 1.0 + 10.0 * 1.0 + 15.0 * 1.0) / 3.0
|
|
assert abs(result - expected) < 1e-10
|
|
|
|
def test_time_weighted_average_single_point_none_bounds(self):
|
|
"""Test time_weighted_average with single point and None bounds."""
|
|
series = [TimeStampedValue(2.0, 10.0)]
|
|
result = time_weighted_average(series, None, None)
|
|
# Single point with 1.0s duration (from 2.0 to 3.0)
|
|
assert result == 10.0
|
|
|
|
def test_time_weighted_average_custom_last_window_s(self):
|
|
"""Test time_weighted_average with custom last_window_s parameter."""
|
|
series = [
|
|
TimeStampedValue(1.0, 5.0),
|
|
TimeStampedValue(2.0, 10.0),
|
|
TimeStampedValue(3.0, 15.0),
|
|
]
|
|
|
|
# Test with last_window_s=2.0 (double the default)
|
|
result_2s = time_weighted_average(series, None, None, last_window_s=2.0)
|
|
# Should use from t=1.0 to t=3.0+2.0=5.0
|
|
# 1.0s at value 5 (from 1.0 to 2.0), 1.0s at value 10 (from 2.0 to 3.0), 2.0s at value 15 (from 3.0 to 5.0)
|
|
expected_2s = (5.0 * 1.0 + 10.0 * 1.0 + 15.0 * 2.0) / 4.0
|
|
assert abs(result_2s - expected_2s) < 1e-10
|
|
|
|
# Test with last_window_s=0.5 (half the default)
|
|
result_0_5s = time_weighted_average(series, None, None, last_window_s=0.5)
|
|
# Should use from t=1.0 to t=3.0+0.5=3.5
|
|
# 1.0s at value 5 (from 1.0 to 2.0), 1.0s at value 10 (from 2.0 to 3.0), 0.5s at value 15 (from 3.0 to 3.5)
|
|
expected_0_5s = (5.0 * 1.0 + 10.0 * 1.0 + 15.0 * 0.5) / 2.5
|
|
assert abs(result_0_5s - expected_0_5s) < 1e-10
|
|
|
|
# Test with window_start specified but window_end None - should still use last_window_s
|
|
result_with_start = time_weighted_average(series, 1.5, None, last_window_s=3.0)
|
|
# Should use from t=1.5 to t=3.0+3.0=6.0
|
|
# 0.5s at value 5 (from 1.5 to 2.0), 1.0s at value 10 (from 2.0 to 3.0), 3.0s at value 15 (from 3.0 to 6.0)
|
|
expected_with_start = (5.0 * 0.5 + 10.0 * 1.0 + 15.0 * 3.0) / 4.5
|
|
assert abs(result_with_start - expected_with_start) < 1e-10
|
|
|
|
# Test that last_window_s is ignored when window_end is explicitly provided
|
|
result_explicit_end = time_weighted_average(
|
|
series, None, 4.0, last_window_s=10.0
|
|
)
|
|
# Should use from t=1.0 to t=4.0 (ignoring last_window_s=10.0)
|
|
# 1.0s at value 5 (from 1.0 to 2.0), 1.0s at value 10 (from 2.0 to 3.0), 1.0s at value 15 (from 3.0 to 4.0)
|
|
expected_explicit_end = (5.0 * 1.0 + 10.0 * 1.0 + 15.0 * 1.0) / 3.0
|
|
assert abs(result_explicit_end - expected_explicit_end) < 1e-10
|
|
|
|
def test_merge_timeseries_dicts_instantaneous_basic(self):
|
|
"""Test merge_timeseries_dicts basic functionality with instantaneous approach."""
|
|
s1 = InMemoryMetricsStore()
|
|
s2 = InMemoryMetricsStore()
|
|
|
|
s1.add_metrics_point({"metric1": 5, "metric2": 10}, timestamp=1.0)
|
|
s1.add_metrics_point({"metric1": 7}, timestamp=2.0)
|
|
|
|
s2.add_metrics_point({"metric1": 3, "metric3": 20}, timestamp=1.5)
|
|
|
|
result = merge_timeseries_dicts(s1.data, s2.data)
|
|
|
|
# metric1: s1 starts at 5 (t=1.0), s2 starts at 3 (t=1.5), s1 changes to 7 (t=2.0)
|
|
expected_metric1 = [
|
|
TimeStampedValue(1.0, 5.0),
|
|
TimeStampedValue(1.5, 8.0), # 5+3=8
|
|
TimeStampedValue(2.0, 10.0), # 3+(7-5)=10
|
|
]
|
|
assert_timeseries_equal(result["metric1"], expected_metric1)
|
|
|
|
# metric2: only from s1
|
|
expected_metric2 = [TimeStampedValue(1.0, 10.0)]
|
|
assert_timeseries_equal(result["metric2"], expected_metric2)
|
|
|
|
# metric3: only from s2
|
|
expected_metric3 = [TimeStampedValue(1.5, 20.0)]
|
|
assert_timeseries_equal(result["metric3"], expected_metric3)
|
|
|
|
def test_merge_instantaneous_vs_windowed_comparison(self):
|
|
"""Compare instantaneous merge vs windowed approach."""
|
|
# Create test data that highlights the difference
|
|
s1 = InMemoryMetricsStore()
|
|
s2 = InMemoryMetricsStore()
|
|
|
|
# Replica 1: 10 requests at t=0.1, then 5 at t=0.9
|
|
s1.add_metrics_point({"requests": 10}, timestamp=0.1)
|
|
s1.add_metrics_point({"requests": 5}, timestamp=0.9)
|
|
|
|
# Replica 2: 3 requests at t=0.5, then 8 at t=1.1
|
|
s2.add_metrics_point({"requests": 3}, timestamp=0.5)
|
|
s2.add_metrics_point({"requests": 8}, timestamp=1.1)
|
|
|
|
# Instantaneous approach
|
|
instantaneous = merge_timeseries_dicts(s1.data, s2.data)
|
|
|
|
# Instantaneous should have: t=0.1: 10, t=0.5: 13, t=0.9: 8, t=1.1: 13
|
|
expected_instantaneous = [
|
|
TimeStampedValue(0.1, 10.0),
|
|
TimeStampedValue(0.5, 13.0), # 10+3=13
|
|
TimeStampedValue(0.9, 8.0), # 3+(5-10)=8
|
|
TimeStampedValue(1.1, 13.0), # 5+(8-3)=13
|
|
]
|
|
assert_timeseries_equal(instantaneous["requests"], expected_instantaneous)
|
|
|
|
def test_instantaneous_merge_handles_zero_deltas(self):
|
|
"""Test that zero deltas are properly filtered out."""
|
|
series1 = [
|
|
TimeStampedValue(1.0, 5.0),
|
|
TimeStampedValue(2.0, 5.0), # No change
|
|
TimeStampedValue(3.0, 7.0),
|
|
]
|
|
series2 = [
|
|
TimeStampedValue(1.5, 3.0),
|
|
TimeStampedValue(2.5, 3.0), # No change
|
|
]
|
|
|
|
result = merge_instantaneous_total([series1, series2])
|
|
|
|
# Should skip zero deltas
|
|
expected = [
|
|
TimeStampedValue(1.0, 5.0),
|
|
TimeStampedValue(1.5, 8.0), # 5+3=8
|
|
TimeStampedValue(3.0, 10.0), # 8+(7-5)=10
|
|
]
|
|
assert_timeseries_equal(result, expected)
|
|
|
|
def test_instantaneous_merge_with_epoch_times(self):
|
|
"""Test instantaneous merge with realistic epoch timestamps."""
|
|
|
|
# Use realistic epoch times (around current time)
|
|
base_time = 1703980800.0 # December 30, 2023 16:00:00 UTC
|
|
|
|
# Simulate 3 replicas reporting metrics over a 30-second period
|
|
replica1_series = [
|
|
TimeStampedValue(base_time + 0.0, 12.0), # t=0s: 12 running requests
|
|
TimeStampedValue(base_time + 5.2, 15.0), # t=5.2s: increased to 15
|
|
TimeStampedValue(base_time + 18.7, 8.0), # t=18.7s: dropped to 8
|
|
TimeStampedValue(base_time + 25.1, 11.0), # t=25.1s: back up to 11
|
|
]
|
|
|
|
replica2_series = [
|
|
TimeStampedValue(base_time + 1.3, 7.0), # t=1.3s: 7 running requests
|
|
TimeStampedValue(base_time + 8.9, 9.0), # t=8.9s: increased to 9
|
|
TimeStampedValue(base_time + 22.4, 4.0), # t=22.4s: dropped to 4
|
|
]
|
|
|
|
replica3_series = [
|
|
TimeStampedValue(base_time + 3.1, 5.0), # t=3.1s: 5 running requests
|
|
TimeStampedValue(base_time + 12.6, 8.0), # t=12.6s: increased to 8
|
|
TimeStampedValue(base_time + 20.8, 6.0), # t=20.8s: dropped to 6
|
|
TimeStampedValue(base_time + 28.3, 9.0), # t=28.3s: increased to 9
|
|
]
|
|
|
|
# Merge all replicas
|
|
result = merge_instantaneous_total(
|
|
[replica1_series, replica2_series, replica3_series]
|
|
)
|
|
|
|
# Expected timeline of instantaneous totals:
|
|
expected = [
|
|
TimeStampedValue(base_time + 0.0, 12.0), # r1 starts: 12
|
|
TimeStampedValue(base_time + 1.3, 19.0), # r2 starts: 12+7=19
|
|
TimeStampedValue(base_time + 3.1, 24.0), # r3 starts: 19+5=24
|
|
TimeStampedValue(base_time + 5.2, 27.0), # r1 changes: 24+(15-12)=27
|
|
TimeStampedValue(base_time + 8.9, 29.0), # r2 changes: 27+(9-7)=29
|
|
TimeStampedValue(base_time + 12.6, 32.0), # r3 changes: 29+(8-5)=32
|
|
TimeStampedValue(base_time + 18.7, 25.0), # r1 changes: 32+(8-15)=25
|
|
TimeStampedValue(base_time + 20.8, 23.0), # r3 changes: 25+(6-8)=23
|
|
TimeStampedValue(base_time + 22.4, 18.0), # r2 changes: 23+(4-9)=18
|
|
TimeStampedValue(base_time + 25.1, 21.0), # r1 changes: 18+(11-8)=21
|
|
TimeStampedValue(base_time + 28.3, 24.0), # r3 changes: 21+(9-6)=24
|
|
]
|
|
|
|
assert_timeseries_equal(result, expected)
|
|
|
|
# Test time-weighted average over different intervals
|
|
# Full series average
|
|
full_avg = time_weighted_average(result, None, None)
|
|
assert full_avg is not None
|
|
assert full_avg > 0
|
|
|
|
# Average over first 10 seconds
|
|
early_avg = time_weighted_average(result, base_time, base_time + 10.0)
|
|
assert early_avg is not None
|
|
|
|
# Average over last 10 seconds
|
|
late_avg = time_weighted_average(result, base_time + 20.0, base_time + 30.0)
|
|
assert late_avg is not None
|
|
|
|
# Verify the averages make sense relative to each other
|
|
# (early period has higher values, so early_avg should be > late_avg)
|
|
assert early_avg > late_avg
|
|
|
|
print(f"Full series average: {full_avg:.2f}")
|
|
print(f"Early period average (0-10s): {early_avg:.2f}")
|
|
print(f"Late period average (20-30s): {late_avg:.2f}")
|
|
|
|
def test_merge_instantaneous_total_timestamp_rounding(self):
|
|
"""Test that timestamps are rounded to 10ms precision."""
|
|
series1 = [
|
|
TimeStampedValue(1.001234, 5.0), # Should round to 1.00
|
|
TimeStampedValue(2.005678, 7.0), # Should round to 2.01
|
|
TimeStampedValue(3.009999, 3.0), # Should round to 3.01
|
|
]
|
|
series2 = [
|
|
TimeStampedValue(1.504321, 2.0), # Should round to 1.50
|
|
TimeStampedValue(2.008765, 4.0), # Should round to 2.01
|
|
]
|
|
|
|
result = merge_instantaneous_total([series1, series2])
|
|
|
|
# Verify timestamps are rounded to 2 decimal places (10ms precision)
|
|
expected_timestamps = [1.00, 1.50, 2.01, 3.01]
|
|
actual_timestamps = [point.timestamp for point in result]
|
|
|
|
assert len(actual_timestamps) == len(expected_timestamps)
|
|
for actual, expected in zip(actual_timestamps, expected_timestamps):
|
|
# c_round is used in the Cython implementation, so we need to use a tolerance
|
|
assert abs(actual - expected) < 1e-4, f"Expected {expected}, got {actual}"
|
|
|
|
# Verify values are correct with rounded timestamps
|
|
expected = [
|
|
TimeStampedValue(1.00, 5.0), # series1 starts
|
|
TimeStampedValue(1.50, 7.0), # series2 starts: 5+2=7
|
|
TimeStampedValue(
|
|
2.01, 11.0
|
|
), # s1 becomes 7, s2 becomes 4. Total: 7 + 4 = 11.0
|
|
TimeStampedValue(3.01, 7.0), # series1 changes: 11+(3-7)=7
|
|
]
|
|
assert_timeseries_equal(result, expected)
|
|
|
|
def test_merge_instantaneous_total_combine_same_timestamp(self):
|
|
"""Test that datapoints with same rounded timestamp are combined."""
|
|
# Create series where multiple events round to the same timestamp
|
|
series1 = [
|
|
TimeStampedValue(1.001, 5.0), # Rounds to 1.00
|
|
TimeStampedValue(1.004, 7.0), # Also rounds to 1.00
|
|
TimeStampedValue(2.000, 10.0), # Rounds to 2.00
|
|
]
|
|
series2 = [
|
|
TimeStampedValue(1.002, 3.0), # Rounds to 1.00
|
|
TimeStampedValue(1.005, 4.0), # Also rounds to 1.00
|
|
]
|
|
|
|
result = merge_instantaneous_total([series1, series2])
|
|
|
|
# Should only have unique rounded timestamps
|
|
timestamps = [point.timestamp for point in result]
|
|
assert timestamps == [
|
|
1.00,
|
|
2.00,
|
|
], f"Expected [1.00, 2.00], got {timestamps}"
|
|
|
|
# The value at 1.00 should be the final state after all changes at that rounded time
|
|
# Order of events at rounded timestamp 1.00:
|
|
# - series1: 0->5 (t=1.001)
|
|
# - series2: 0->3 (t=1.002)
|
|
# - series1: 5->7 (t=1.004)
|
|
# - series2: 3->4 (t=1.005)
|
|
# Final state: series1=7, series2=4, total=11
|
|
expected = [
|
|
TimeStampedValue(1.00, 11.0), # Final combined state at rounded timestamp
|
|
TimeStampedValue(2.00, 14.0), # series1 changes: 11+(10-7)=14
|
|
]
|
|
assert_timeseries_equal(result, expected)
|
|
|
|
def test_merge_instantaneous_total_edge_cases_rounding(self):
|
|
"""Test edge cases for timestamp rounding and combination."""
|
|
# Test rounding edge cases with TWO series to trigger merge logic
|
|
series1 = [
|
|
TimeStampedValue(1.004999, 5.0), # Should round to 1.0
|
|
TimeStampedValue(1.005000, 7.0), # Should round to 1.0 (round half to even)
|
|
TimeStampedValue(1.005001, 9.0), # Should round to 1.01
|
|
]
|
|
series2 = [
|
|
TimeStampedValue(0.5, 2.0), # Add a second series to trigger merge
|
|
]
|
|
|
|
result = merge_instantaneous_total([series1, series2])
|
|
|
|
# With merge, timestamps should be rounded
|
|
# series2 contributes value at 0.5 (rounds to 0.5)
|
|
# Then series1 points get merged with rounding
|
|
expected_timestamps = [0.5, 1.0, 1.01]
|
|
actual_timestamps = [point.timestamp for point in result]
|
|
assert actual_timestamps == expected_timestamps
|
|
|
|
# Values should reflect the changes
|
|
# 0.5: series2 adds 2.0 -> total = 2.0
|
|
# 1.0: series1 adds 5.0 at 1.004999, then changes to 7.0 at 1.005000 -> total = 2.0 + 7.0 = 9.0
|
|
# 1.01: series1 changes to 9.0 at 1.005001 -> total = 2.0 + 9.0 = 11.0
|
|
expected = [
|
|
TimeStampedValue(0.5, 2.0),
|
|
TimeStampedValue(1.0, 9.0),
|
|
TimeStampedValue(1.01, 11.0),
|
|
]
|
|
assert_timeseries_equal(result, expected)
|
|
|
|
def test_merge_instantaneous_total_no_changes_filtered(self):
|
|
"""Test that zero-change events are filtered even with rounding."""
|
|
# Use multiple replicas to trigger the merge logic (single replica returns as-is)
|
|
series1 = [
|
|
TimeStampedValue(1.001, 5.0), # Rounds to 1.00
|
|
TimeStampedValue(1.004, 5.0), # Also rounds to 1.00, no change
|
|
TimeStampedValue(2.000, 7.0), # Rounds to 2.00, change
|
|
]
|
|
series2 = [
|
|
TimeStampedValue(0.5, 1.0), # Add a second series to trigger merge
|
|
]
|
|
|
|
result = merge_instantaneous_total([series1, series2])
|
|
|
|
# Should only include points where value actually changed
|
|
# 0.5: series2 adds 1.0 -> total = 1.0
|
|
# 1.00: series1 adds 5.0 (both 1.001 and 1.004 round to 1.00, same value) -> total = 6.0
|
|
# 2.00: series1 changes to 7.0 -> total = 8.0
|
|
expected = [
|
|
TimeStampedValue(0.5, 1.0),
|
|
TimeStampedValue(1.0, 6.0),
|
|
TimeStampedValue(2.0, 8.0),
|
|
]
|
|
assert_timeseries_equal(result, expected)
|
|
|
|
|
|
class TestCythonImplementationEdgeCases:
|
|
"""Test edge cases and stress scenarios for the Cython/C++ implementation."""
|
|
|
|
def test_merge_large_number_of_replicas(self):
|
|
"""Test merging a large number of replicas (stress test for heap)."""
|
|
# Create 100 replicas with overlapping timestamps
|
|
replicas = []
|
|
for i in range(100):
|
|
series = [
|
|
TimeStampedValue(float(i) * 0.1, float(i)),
|
|
TimeStampedValue(float(i) * 0.1 + 5.0, float(i) + 10),
|
|
]
|
|
replicas.append(series)
|
|
|
|
result = merge_instantaneous_total(replicas)
|
|
|
|
# Many points will be merged due to rounding (149 unique rounded timestamps)
|
|
assert len(result) == 149
|
|
# All timestamps should be sorted
|
|
timestamps = [p.timestamp for p in result]
|
|
assert timestamps == sorted(timestamps)
|
|
# First value should be from replica at index 1 (i=1, timestamp=0.1, value=1.0)
|
|
assert result[0].timestamp == 0.1
|
|
assert result[0].value == 1.0
|
|
|
|
def test_merge_very_long_single_series(self):
|
|
"""Test merging a very long single series (10k points)."""
|
|
series = [
|
|
TimeStampedValue(float(i) * 0.01, float(i % 100)) for i in range(10000)
|
|
]
|
|
result = merge_instantaneous_total([series])
|
|
|
|
# Single series should return as-is
|
|
assert len(result) == len(series)
|
|
assert result == series
|
|
|
|
def test_merge_many_points_same_timestamp(self):
|
|
"""Test handling many points with identical timestamps."""
|
|
series1 = [
|
|
TimeStampedValue(1.0, 5.0),
|
|
TimeStampedValue(1.0, 10.0), # Same timestamp
|
|
TimeStampedValue(1.0, 15.0), # Same timestamp
|
|
]
|
|
series2 = [
|
|
TimeStampedValue(1.0, 20.0),
|
|
TimeStampedValue(1.0, 25.0), # Same timestamp
|
|
]
|
|
|
|
result = merge_instantaneous_total([series1, series2])
|
|
|
|
# All should be merged into single point at rounded timestamp 1.0
|
|
assert len(result) == 1
|
|
# First point should have timestamp 1.0
|
|
assert result[0].timestamp == 1.0
|
|
|
|
def test_merge_with_very_small_timestamps(self):
|
|
"""Test handling of very small timestamp values."""
|
|
series1 = [
|
|
TimeStampedValue(1e-10, 5.0),
|
|
TimeStampedValue(2e-10, 7.0),
|
|
]
|
|
series2 = [
|
|
TimeStampedValue(1.5e-10, 3.0),
|
|
]
|
|
|
|
result = merge_instantaneous_total([series1, series2])
|
|
|
|
# Should handle without crashing
|
|
assert isinstance(result, list)
|
|
# All very small timestamps should round to 0.0
|
|
assert all(p.timestamp == 0.0 for p in result)
|
|
|
|
def test_merge_with_very_large_timestamps(self):
|
|
"""Test handling of very large timestamp values (e.g., Unix epoch times)."""
|
|
base = 1.7e9 # Year 2023+
|
|
series1 = [
|
|
TimeStampedValue(base + 0.1, 5.0),
|
|
TimeStampedValue(base + 5.0, 10.0),
|
|
]
|
|
series2 = [
|
|
TimeStampedValue(base + 2.0, 3.0),
|
|
TimeStampedValue(base + 7.0, 8.0),
|
|
]
|
|
|
|
result = merge_instantaneous_total([series1, series2])
|
|
|
|
# Should handle large timestamps correctly
|
|
assert len(result) == 4
|
|
expected = [
|
|
TimeStampedValue(base + 0.1, 5.0),
|
|
TimeStampedValue(base + 2.0, 8.0),
|
|
TimeStampedValue(base + 5.0, 13.0),
|
|
TimeStampedValue(base + 7.0, 18.0),
|
|
]
|
|
assert_timeseries_equal(result, expected)
|
|
|
|
def test_merge_with_negative_values(self):
|
|
"""Test handling of negative metric values."""
|
|
series1 = [
|
|
TimeStampedValue(1.0, -5.0),
|
|
TimeStampedValue(2.0, -3.0),
|
|
]
|
|
series2 = [
|
|
TimeStampedValue(1.5, -2.0),
|
|
TimeStampedValue(2.5, -1.0),
|
|
]
|
|
|
|
result = merge_instantaneous_total([series1, series2])
|
|
|
|
# Should handle negative values correctly
|
|
expected = [
|
|
TimeStampedValue(1.0, -5.0),
|
|
TimeStampedValue(1.5, -7.0), # -5 + -2
|
|
TimeStampedValue(2.0, -5.0), # -2 + (-3 - (-5))
|
|
TimeStampedValue(2.5, -4.0), # -3 + (-1 - (-2))
|
|
]
|
|
assert_timeseries_equal(result, expected)
|
|
|
|
def test_merge_with_zero_values(self):
|
|
"""Test handling of zero values in series."""
|
|
series1 = [
|
|
TimeStampedValue(1.0, 0.0),
|
|
TimeStampedValue(2.0, 5.0),
|
|
TimeStampedValue(3.0, 0.0),
|
|
]
|
|
series2 = [
|
|
TimeStampedValue(1.5, 0.0),
|
|
TimeStampedValue(2.5, 0.0),
|
|
]
|
|
|
|
result = merge_instantaneous_total([series1, series2])
|
|
|
|
# Zero-delta changes are filtered out, so only value-changing points remain
|
|
# t=1.0: series1 starts at 0 (total=0, but filtered as starting from 0)
|
|
# t=1.5: series2 starts at 0 (total=0, no change from implied 0, filtered)
|
|
# t=2.0: series1 changes to 5 (total=5, delta=+5)
|
|
# t=3.0: series1 changes to 0 (total=0, delta=-5)
|
|
expected = [
|
|
TimeStampedValue(2.0, 5.0),
|
|
TimeStampedValue(3.0, 0.0),
|
|
]
|
|
assert_timeseries_equal(result, expected)
|
|
|
|
def test_merge_alternating_series(self):
|
|
"""Test merging series with perfectly alternating timestamps."""
|
|
series1 = [
|
|
TimeStampedValue(1.0, 5.0),
|
|
TimeStampedValue(3.0, 7.0),
|
|
TimeStampedValue(5.0, 9.0),
|
|
]
|
|
series2 = [
|
|
TimeStampedValue(2.0, 3.0),
|
|
TimeStampedValue(4.0, 4.0),
|
|
TimeStampedValue(6.0, 2.0),
|
|
]
|
|
|
|
result = merge_instantaneous_total([series1, series2])
|
|
|
|
expected = [
|
|
TimeStampedValue(1.0, 5.0),
|
|
TimeStampedValue(2.0, 8.0), # 5 + 3
|
|
TimeStampedValue(3.0, 10.0), # 3 + 7
|
|
TimeStampedValue(4.0, 11.0), # 7 + 4
|
|
TimeStampedValue(5.0, 13.0), # 4 + 9
|
|
TimeStampedValue(6.0, 11.0), # 9 + 2
|
|
]
|
|
assert_timeseries_equal(result, expected)
|
|
|
|
def test_merge_with_floating_point_precision(self):
|
|
"""Test handling of floating point precision edge cases."""
|
|
series1 = [
|
|
TimeStampedValue(1.0 + 1e-15, 5.0),
|
|
TimeStampedValue(2.0 + 2e-15, 7.0),
|
|
]
|
|
series2 = [
|
|
TimeStampedValue(1.5 + 1e-15, 3.0),
|
|
TimeStampedValue(2.5 + 2e-15, 4.0),
|
|
]
|
|
|
|
result = merge_instantaneous_total([series1, series2])
|
|
|
|
# Should handle floating point precision correctly
|
|
assert len(result) == 4
|
|
# Timestamps should be properly rounded
|
|
timestamps = [p.timestamp for p in result]
|
|
assert all(isinstance(t, float) for t in timestamps)
|
|
|
|
def test_time_weighted_average_with_zero_duration_window(self):
|
|
"""Test time_weighted_average with zero-duration window."""
|
|
series = [
|
|
TimeStampedValue(1.0, 5.0),
|
|
TimeStampedValue(2.0, 10.0),
|
|
]
|
|
# Window where start == end should return 0 or None
|
|
result = time_weighted_average(series, 1.5, 1.5)
|
|
# Implementation should handle this gracefully
|
|
assert result is None or result == 0.0
|
|
|
|
def test_time_weighted_average_with_negative_window(self):
|
|
"""Test time_weighted_average with window_end < window_start."""
|
|
series = [
|
|
TimeStampedValue(1.0, 5.0),
|
|
TimeStampedValue(2.0, 10.0),
|
|
]
|
|
result = time_weighted_average(series, 2.0, 1.0)
|
|
# Should return None for invalid window
|
|
assert result is None
|
|
|
|
def test_time_weighted_average_very_large_values(self):
|
|
"""Test time_weighted_average with very large metric values."""
|
|
series = [
|
|
TimeStampedValue(1.0, 1e12),
|
|
TimeStampedValue(2.0, 2e12),
|
|
TimeStampedValue(3.0, 3e12),
|
|
]
|
|
result = time_weighted_average(series, None, None)
|
|
|
|
# Should handle large values without overflow
|
|
# Expected: (1e12*1 + 2e12*1 + 3e12*1) / 3 = 2e12
|
|
assert result == pytest.approx(2e12, rel=1e-9)
|
|
|
|
def test_time_weighted_average_very_small_values(self):
|
|
"""Test time_weighted_average with very small metric values."""
|
|
series = [
|
|
TimeStampedValue(1.0, 1e-12),
|
|
TimeStampedValue(2.0, 2e-12),
|
|
TimeStampedValue(3.0, 3e-12),
|
|
]
|
|
result = time_weighted_average(series, None, None)
|
|
|
|
# Should handle very small values without underflow
|
|
# Expected: (1e-12*1 + 2e-12*1 + 3e-12*1) / 3 = 2e-12
|
|
assert result == pytest.approx(2e-12, rel=1e-6)
|
|
|
|
def test_time_weighted_average_negative_values(self):
|
|
"""Test time_weighted_average with negative metric values.
|
|
|
|
Regression test: Ensures negative results are returned correctly,
|
|
not incorrectly treated as None/invalid.
|
|
"""
|
|
# All negative values
|
|
series = [
|
|
TimeStampedValue(1.0, -5.0),
|
|
TimeStampedValue(2.0, -10.0),
|
|
TimeStampedValue(3.0, -15.0),
|
|
]
|
|
result = time_weighted_average(series, None, None)
|
|
|
|
# Expected: (-5*1 + -10*1 + -15*1) / 3 = -10.0
|
|
assert result == pytest.approx(-10.0, rel=1e-9)
|
|
|
|
# Mixed positive and negative values with negative average
|
|
series_mixed = [
|
|
TimeStampedValue(1.0, -20.0),
|
|
TimeStampedValue(2.0, 5.0),
|
|
TimeStampedValue(3.0, -10.0),
|
|
]
|
|
result_mixed = time_weighted_average(series_mixed, None, None)
|
|
|
|
# Expected: (-20*1 + 5*1 + -10*1) / 3 = -25/3 ≈ -8.333...
|
|
assert result_mixed == pytest.approx(-25.0 / 3.0, rel=1e-9)
|
|
|
|
# Single negative value
|
|
series_single = [TimeStampedValue(1.0, -0.5)]
|
|
result_single = time_weighted_average(series_single, None, None)
|
|
assert result_single == pytest.approx(-0.5, rel=1e-9)
|
|
|
|
# Negative value exactly at -1.0 (the old sentinel value)
|
|
series_sentinel = [TimeStampedValue(1.0, -1.0)]
|
|
result_sentinel = time_weighted_average(series_sentinel, None, None)
|
|
assert result_sentinel == pytest.approx(-1.0, rel=1e-9)
|
|
|
|
def test_time_weighted_average_negative_timestamps(self):
|
|
"""Test time_weighted_average with negative timestamps.
|
|
|
|
Regression test: Ensures negative window_start and window_end values
|
|
are used as actual timestamps, not incorrectly treated as the sentinel
|
|
value for None. The Cython implementation uses negative infinity as the
|
|
sentinel for None, so any valid float (including -1.0) works as a boundary.
|
|
"""
|
|
# Series with negative timestamps
|
|
series = [
|
|
TimeStampedValue(-10.0, 5.0),
|
|
TimeStampedValue(-5.0, 10.0),
|
|
TimeStampedValue(0.0, 15.0),
|
|
]
|
|
|
|
# Test with explicit negative window_start
|
|
result_neg_start = time_weighted_average(series, -8.0, -2.0)
|
|
# [-8.0, -5.0): value = 5.0, duration = 3.0
|
|
# [-5.0, -2.0): value = 10.0, duration = 3.0
|
|
expected = (5.0 * 3.0 + 10.0 * 3.0) / 6.0
|
|
assert result_neg_start == pytest.approx(expected, rel=1e-9)
|
|
|
|
# Test with negative window_end (but positive window_start)
|
|
series_pos = [
|
|
TimeStampedValue(-5.0, 20.0),
|
|
TimeStampedValue(-2.0, 30.0),
|
|
]
|
|
result_neg_end = time_weighted_average(series_pos, -5.0, -1.5)
|
|
# [-5.0, -2.0): value = 20.0, duration = 3.0
|
|
# [-2.0, -1.5): value = 30.0, duration = 0.5
|
|
expected_neg_end = (20.0 * 3.0 + 30.0 * 0.5) / 3.5
|
|
assert result_neg_end == pytest.approx(expected_neg_end, rel=1e-9)
|
|
|
|
# Test with both negative window boundaries
|
|
result_both_neg = time_weighted_average(series, -10.0, -5.0)
|
|
# [-10.0, -5.0): value = 5.0, duration = 5.0
|
|
assert result_both_neg == pytest.approx(5.0, rel=1e-9)
|
|
|
|
# Test that -1.0 can now be used as an actual window boundary
|
|
# (previously it was used as a sentinel for None, but now we use -inf)
|
|
series_for_minus_one = [
|
|
TimeStampedValue(-2.0, 50.0),
|
|
TimeStampedValue(-1.0, 100.0),
|
|
TimeStampedValue(0.0, 150.0),
|
|
]
|
|
# Using window_start=-1.0 should use -1.0 as actual start, not "use default"
|
|
result_minus_one_start = time_weighted_average(series_for_minus_one, -1.0, 0.5)
|
|
# [-1.0, 0.0): value = 100.0, duration = 1.0
|
|
# [0.0, 0.5): value = 150.0, duration = 0.5
|
|
expected_minus_one = (100.0 * 1.0 + 150.0 * 0.5) / 1.5
|
|
assert result_minus_one_start == pytest.approx(expected_minus_one, rel=1e-9)
|
|
|
|
# Also verify that None still works correctly for default behavior
|
|
series_for_none = [
|
|
TimeStampedValue(2.0, 100.0),
|
|
TimeStampedValue(3.0, 200.0),
|
|
]
|
|
result_none_start = time_weighted_average(series_for_none, None, 4.0)
|
|
# With None, should use first timestamp (2.0) as start
|
|
# [2.0, 3.0): value = 100.0, duration = 1.0
|
|
# [3.0, 4.0): value = 200.0, duration = 1.0
|
|
expected_none = (100.0 * 1.0 + 200.0 * 1.0) / 2.0
|
|
assert result_none_start == pytest.approx(expected_none, rel=1e-9)
|
|
|
|
def test_time_weighted_average_minus_one_boundary(self):
|
|
"""Regression test: -1.0 can be used as an actual window boundary.
|
|
|
|
Previously, the Cython implementation used -1.0 as a sentinel value for
|
|
None, which made it impossible to use -1.0 as an actual window boundary.
|
|
This was fixed by using negative infinity (-inf) as the sentinel instead.
|
|
"""
|
|
# Test -1.0 as window_start
|
|
series = [
|
|
TimeStampedValue(-2.0, 10.0),
|
|
TimeStampedValue(-1.0, 20.0),
|
|
TimeStampedValue(0.0, 30.0),
|
|
TimeStampedValue(1.0, 40.0),
|
|
]
|
|
result = time_weighted_average(series, -1.0, 1.0)
|
|
# [-1.0, 0.0): value = 20.0, duration = 1.0
|
|
# [0.0, 1.0): value = 30.0, duration = 1.0
|
|
expected = (20.0 * 1.0 + 30.0 * 1.0) / 2.0
|
|
assert result == pytest.approx(expected, rel=1e-9)
|
|
|
|
# Test -1.0 as window_end
|
|
result_end = time_weighted_average(series, -2.0, -1.0)
|
|
# [-2.0, -1.0): value = 10.0, duration = 1.0
|
|
assert result_end == pytest.approx(10.0, rel=1e-9)
|
|
|
|
# Test both boundaries as -1.0 (degenerate zero-width window)
|
|
result_both = time_weighted_average(series, -1.0, -1.0)
|
|
assert result_both is None # Zero-width window should return None
|
|
|
|
def test_time_weighted_average_with_long_series(self):
|
|
"""Test time_weighted_average with a very long series (stress test)."""
|
|
# Create a series with 10,000 points
|
|
series = [
|
|
TimeStampedValue(float(i) * 0.1, float(i % 100)) for i in range(10000)
|
|
]
|
|
|
|
result = time_weighted_average(series, None, None)
|
|
|
|
# Should compute correctly without performance issues
|
|
# The pattern repeats every 100 points, so average should be around 49.5
|
|
assert result == pytest.approx(49.5, rel=0.01)
|
|
|
|
def test_time_weighted_average_with_irregular_spacing(self):
|
|
"""Test time_weighted_average with irregularly spaced timestamps."""
|
|
series = [
|
|
TimeStampedValue(1.0, 10.0),
|
|
TimeStampedValue(1.001, 20.0), # Very close
|
|
TimeStampedValue(5.0, 30.0), # Big gap
|
|
TimeStampedValue(5.001, 40.0), # Very close again
|
|
TimeStampedValue(100.0, 50.0), # Huge gap
|
|
]
|
|
|
|
result = time_weighted_average(series, None, None)
|
|
|
|
# Should handle irregular spacing correctly
|
|
# Most weight should be on the 40.0 value due to long duration (94.999s)
|
|
# Expected: (10*0.001 + 20*3.999 + 30*0.001 + 40*94.999 + 50*1.0) / 100.0 ≈ 39.30
|
|
expected = (
|
|
10 * 0.001 + 20 * 3.999 + 30 * 0.001 + 40 * 94.999 + 50 * 1.0
|
|
) / 100.0
|
|
assert result == pytest.approx(expected, rel=1e-6)
|
|
|
|
def test_merge_with_single_point_series(self):
|
|
"""Test merging multiple single-point series."""
|
|
replicas = [
|
|
[TimeStampedValue(1.0, 5.0)],
|
|
[TimeStampedValue(2.0, 3.0)],
|
|
[TimeStampedValue(3.0, 7.0)],
|
|
[TimeStampedValue(4.0, 2.0)],
|
|
]
|
|
|
|
result = merge_instantaneous_total(replicas)
|
|
|
|
expected = [
|
|
TimeStampedValue(1.0, 5.0),
|
|
TimeStampedValue(2.0, 8.0),
|
|
TimeStampedValue(3.0, 15.0),
|
|
TimeStampedValue(4.0, 17.0),
|
|
]
|
|
assert_timeseries_equal(result, expected)
|
|
|
|
def test_merge_mixed_empty_and_nonempty(self):
|
|
"""Test merging with a mix of empty and non-empty series."""
|
|
replicas = [
|
|
[],
|
|
[TimeStampedValue(1.0, 5.0)],
|
|
[],
|
|
[],
|
|
[TimeStampedValue(2.0, 3.0)],
|
|
[],
|
|
]
|
|
|
|
result = merge_instantaneous_total(replicas)
|
|
|
|
expected = [
|
|
TimeStampedValue(1.0, 5.0),
|
|
TimeStampedValue(2.0, 8.0),
|
|
]
|
|
assert_timeseries_equal(result, expected)
|
|
|
|
def test_merge_series_with_duplicate_consecutive_values(self):
|
|
"""Test merging when series have duplicate consecutive values."""
|
|
series1 = [
|
|
TimeStampedValue(1.0, 5.0),
|
|
TimeStampedValue(2.0, 5.0), # No change
|
|
TimeStampedValue(3.0, 5.0), # No change
|
|
TimeStampedValue(4.0, 10.0), # Change
|
|
]
|
|
series2 = [
|
|
TimeStampedValue(1.5, 3.0),
|
|
TimeStampedValue(2.5, 3.0), # No change
|
|
TimeStampedValue(3.5, 3.0), # No change
|
|
]
|
|
|
|
result = merge_instantaneous_total([series1, series2])
|
|
|
|
# Should only emit points where total changes
|
|
expected = [
|
|
TimeStampedValue(1.0, 5.0),
|
|
TimeStampedValue(1.5, 8.0),
|
|
TimeStampedValue(4.0, 13.0),
|
|
]
|
|
assert_timeseries_equal(result, expected)
|
|
|
|
def test_merge_with_descending_values(self):
|
|
"""Test merging series with descending values."""
|
|
series1 = [
|
|
TimeStampedValue(1.0, 100.0),
|
|
TimeStampedValue(2.0, 80.0),
|
|
TimeStampedValue(3.0, 60.0),
|
|
]
|
|
series2 = [
|
|
TimeStampedValue(1.5, 50.0),
|
|
TimeStampedValue(2.5, 30.0),
|
|
TimeStampedValue(3.5, 10.0),
|
|
]
|
|
|
|
result = merge_instantaneous_total([series1, series2])
|
|
|
|
# Values can increase or decrease as series change
|
|
expected = [
|
|
TimeStampedValue(1.0, 100.0),
|
|
TimeStampedValue(1.5, 150.0), # 100 + 50
|
|
TimeStampedValue(2.0, 130.0), # 50 + (80-100)
|
|
TimeStampedValue(2.5, 110.0), # 80 + (30-50)
|
|
TimeStampedValue(3.0, 90.0), # 30 + (60-80)
|
|
TimeStampedValue(3.5, 70.0), # 60 + (10-30)
|
|
]
|
|
assert_timeseries_equal(result, expected)
|
|
|
|
def test_time_weighted_average_single_point_in_middle_of_window(self):
|
|
"""Test time_weighted_average with single point inside window."""
|
|
series = [TimeStampedValue(5.0, 42.0)]
|
|
|
|
# Window completely contains the point
|
|
result = time_weighted_average(series, 3.0, 7.0)
|
|
|
|
# Should use default value (0.0) before point, then 42.0 after
|
|
# [3.0, 5.0): value = 0.0, duration = 2.0
|
|
# [5.0, 7.0): value = 42.0, duration = 2.0
|
|
expected = (0.0 * 2.0 + 42.0 * 2.0) / 4.0
|
|
assert result == expected
|
|
|
|
def test_time_weighted_average_window_before_all_data(self):
|
|
"""Test time_weighted_average with window completely before all data."""
|
|
series = [
|
|
TimeStampedValue(10.0, 5.0),
|
|
TimeStampedValue(11.0, 10.0),
|
|
]
|
|
|
|
# Window is [5.0, 8.0), data starts at 10.0
|
|
result = time_weighted_average(series, 5.0, 8.0)
|
|
|
|
# Should use default value 0.0 for entire window
|
|
assert result == 0.0
|
|
|
|
def test_time_weighted_average_with_last_window_zero(self):
|
|
"""Test time_weighted_average with last_window_s = 0."""
|
|
series = [
|
|
TimeStampedValue(1.0, 5.0),
|
|
TimeStampedValue(2.0, 10.0),
|
|
]
|
|
|
|
# With last_window_s=0, the last point should have zero duration
|
|
result = time_weighted_average(series, None, None, last_window_s=0.0)
|
|
|
|
# Should only count the segment from 1.0 to 2.0
|
|
assert result == 5.0
|
|
|
|
def test_merge_timestamp_rounding_boundary_at_005(self):
|
|
"""Test timestamp rounding at the 0.005 boundary (rounds to even)."""
|
|
# Testing banker's rounding (round half to even)
|
|
series1 = [
|
|
TimeStampedValue(
|
|
1.005, 5.0
|
|
), # Rounds to 1.00 (or 1.01 depending on c_round)
|
|
TimeStampedValue(1.015, 7.0), # Rounds to 1.02 (or 1.01)
|
|
TimeStampedValue(1.025, 9.0), # Rounds to 1.02 (or 1.03)
|
|
]
|
|
series2 = [
|
|
TimeStampedValue(0.5, 2.0), # Add second series to trigger merge
|
|
]
|
|
|
|
result = merge_instantaneous_total([series1, series2])
|
|
|
|
# Verify timestamps are properly rounded
|
|
timestamps = [p.timestamp for p in result]
|
|
# Actual output: [0.5, 1.0, 1.01, 1.02]
|
|
assert len(timestamps) == 4
|
|
assert timestamps[0] == 0.5
|
|
# Verify all timestamps are properly rounded to 2 decimal places
|
|
for ts in timestamps:
|
|
assert ts == pytest.approx(round(ts, 2), abs=1e-10)
|
|
|
|
def test_merge_with_extreme_value_changes(self):
|
|
"""Test merging with extreme value changes."""
|
|
series1 = [
|
|
TimeStampedValue(1.0, 1e-10),
|
|
TimeStampedValue(2.0, 1e10), # Huge increase
|
|
]
|
|
series2 = [
|
|
TimeStampedValue(1.5, 1e10),
|
|
TimeStampedValue(2.5, 1e-10), # Huge decrease
|
|
]
|
|
|
|
result = merge_instantaneous_total([series1, series2])
|
|
|
|
# Should handle extreme value changes
|
|
assert len(result) == 4
|
|
# Check that values are computed correctly despite extreme changes
|
|
assert result[0].value == pytest.approx(1e-10, rel=1e-6)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(pytest.main(["-v", "-s", __file__]))
|