Add ChangePoints parity tests

otava/change_point_divisive/base.py

@@ -495,9 +495,25 @@ def select_metrics(self, m: list[str] | str):
         request efficient. This will loop over all ChangePointGroup s.
         Use ChangePointsByMetric if you need this to be fast.
         """
+        if not isinstance(m, list):
+            if not isinstance(m, str):
+                raise TypeError(
+                    "ChangePoints.select_metrics() takes as argument a str or a list of str."
+                )
+            m = [m]
+        if not all(isinstance(metric, str) for metric in m):
+            raise TypeError(
+                "ChangePoints.select_metrics() takes as argument a str or a list of str."
+            )
+        for metric in m:
+            if metric not in self:
+                raise KeyError(metric)
+
         filtered = ChangePoints()
         for cpg in self._change_points:
-            filtered.append(cpg.select_metrics(m))
+            selected = cpg.select_metrics(m)
+            if selected.changes:
+                filtered.append(selected)
         return filtered
 
     def get_change_points_for_metric(self, m: str):
@@ -546,9 +562,9 @@ class ChangePointsByTime(ChangePoints):
         to make it explicit that your code at that point explicitly wanted a collection of ChangePoints
         ordered by time.
 
-        The pivot() method will return a new object (a copy) holding the same data, but ordered by metrics
-        as the primary and optimized axis. The method by_metric() can be used for the same purpose. Note
-        that the method by_time() is a no-op and returns self, it doesn't even do a copy.
+        The pivot() method returns a new view holding the same ChangePoint objects, but ordered by metrics
+        as the primary and optimized axis. The method by_metric() can be used for the same purpose. Use
+        copy().pivot() for an isolated copy. The method by_time() is a no-op and returns self.
     """
     @classmethod
     def from_dict(cls, cps: dict):
@@ -568,9 +584,9 @@ class ChangePointsByMetric(ChangePoints):
         You can create empty instances of this class, or you can also use the factory method
         `ChangePoints.from_dict()` to get an instance of this type.
 
-        The pivot() method will return a new object (a copy) holding the same data, but ordered by time
-        as the primary and optimized axis. The method by_time() can be used for the same purpose. Note
-        that the method by_metric() is a no-op and returns self, it doesn't even do a copy.
+        The pivot() method returns a new view holding the same ChangePoint objects, but ordered by time
+        as the primary and optimized axis. The method by_time() can be used for the same purpose. Use
+        copy().pivot() for an isolated copy. The method by_metric() is a no-op and returns self.
     """
 
     def __init__(self):
@@ -609,7 +625,7 @@ def pivot(self):
                 intermediate.append(cpg)
         cp_by_time = ChangePointsByTime()
         for cpg in sorted(intermediate, key=lambda cpg: cpg.time):
-            cp_by_time.append(cpg)
+            cp_by_time.append(cpg.select_metrics(list(cpg.metrics())))
         return cp_by_time
 
     def append(self, cpg: ChangePointGroup):
@@ -641,7 +657,7 @@ def extend(self, cps: list[ChangePointGroup]):
                 if metric1 not in self._change_points:
                     self._change_points[metric1] = []
                 if (not self._change_points[metric1]) or cpg.time > self._change_points[metric1][-1].time:
-                    self._change_points[metric1].append(cpg)
+                    self._change_points[metric1].append(cpg.select_metrics(metric1))
                 else:
                     raise ValueError(
                         "ChangePoints.extend() can only be used such that time is monotonically increasing"
@@ -691,8 +707,14 @@ def select_metrics(self, m: list[str] | str):
         """
         if not isinstance(m, list):
             if not isinstance(m, str):
-                TypeError("ChangePoints.select_metrics() takes as argument a str or a list of str.")
+                raise TypeError(
+                    "ChangePoints.select_metrics() takes as argument a str or a list of str."
+                )
             m = [m]
+        if not all(isinstance(metric, str) for metric in m):
+            raise TypeError(
+                "ChangePoints.select_metrics() takes as argument a str or a list of str."
+            )
 
         filtered = ChangePointsByMetric()
         for metric in m:

pyproject.toml

@@ -64,6 +64,7 @@ dependencies = [
 
 [project.optional-dependencies]
 dev = [
+    "hypothesis>=6.0",
     "pytest>=9.0.1",
     "pytest-benchmark>=5.2.3",
     "pytz==2025.2",

tests/change_point_classes_test.py

@@ -28,6 +28,8 @@
 
 import numpy as np
 import pytest
+from hypothesis import example, given, settings
+from hypothesis import strategies as st
 
 from otava.change_point_divisive.base import (
     BaseStats,
@@ -59,6 +61,129 @@ def make_group(time, metric="m", index=3, commit="sha"):
     )
 
 
+SWARM_METRICS = ("errors", "latency", "throughput")
+GOLDEN_HISTORY = (
+    (10.0, "commit-a", (("errors", 11), ("latency", 12))),
+    (20.0, "commit-b", (("throughput", 21),)),
+    (30.0, "commit-c", (("latency", 31), ("throughput", 32))),
+)
+
+
+@st.composite
+def history_specs(draw):
+    times = draw(
+        st.lists(
+            st.integers(min_value=1, max_value=1_000),
+            min_size=1,
+            max_size=12,
+            unique=True,
+        ).map(sorted)
+    )
+    history = []
+    for row_number, time in enumerate(times):
+        metrics = draw(
+            st.sets(
+                st.sampled_from(SWARM_METRICS),
+                min_size=1,
+                max_size=len(SWARM_METRICS),
+            )
+        )
+        changes = tuple(
+            (metric, row_number * len(SWARM_METRICS) + SWARM_METRICS.index(metric))
+            for metric in sorted(metrics)
+        )
+        history.append((float(time), f"commit-{time}", changes))
+    return tuple(history)
+
+
+def make_history_groups(history):
+    return [
+        ChangePointGroup(
+            time=time,
+            attributes={"commit": commit},
+            changes={metric: make_cp(metric, index) for metric, index in changes},
+        )
+        for time, commit, changes in history
+    ]
+
+
+def group_snapshot(group):
+    return (
+        group.time,
+        group.commit(),
+        tuple(
+            (metric, group[metric].metric, group[metric].index)
+            for metric in sorted(group.metrics())
+        ),
+    )
+
+
+def history_snapshot(change_points):
+    return [group_snapshot(group) for group in change_points]
+
+
+def expected_history_snapshot(history):
+    return [
+        (time, commit, tuple((metric, metric, index) for metric, index in changes))
+        for time, commit, changes in history
+    ]
+
+
+def expected_metric_snapshots(history):
+    metrics = {metric for _, _, changes in history for metric, _ in changes}
+    return {
+        metric: [
+            (time, commit, ((metric, metric, dict(changes)[metric]),))
+            for time, commit, changes in history
+            if metric in dict(changes)
+        ]
+        for metric in metrics
+    }
+
+
+def metric_snapshots(change_points):
+    return {
+        metric: history_snapshot(groups) for metric, groups in change_points.items()
+    }
+
+
+def assert_read_api(change_points, expected, expected_by_metric):
+    assert history_snapshot(change_points) == expected
+    assert group_snapshot(change_points[0]) == expected[0]
+    assert change_points.metrics() == set(expected_by_metric)
+    assert metric_snapshots(change_points) == expected_by_metric
+    assert history_snapshot(change_points.pivot()) == expected
+    assert history_snapshot(change_points.by_time()) == expected
+    assert metric_snapshots(change_points.by_metric()) == expected_by_metric
+
+    for row in expected:
+        time, commit, changes = row
+        assert group_snapshot(change_points.at_timestamp(time)) == row
+        assert group_snapshot(change_points.at_timestamp(time + 0.00005)) == row
+        assert group_snapshot(change_points.at_commit(commit)) == row
+        for metric, _, _ in changes:
+            assert metric in change_points
+
+    for metric, metric_rows in expected_by_metric.items():
+        assert history_snapshot(change_points.select_metrics(metric)) == metric_rows
+        assert [cp.index for cp in change_points.get_change_points_for_metric(metric)] == [
+            row[2][0][2] for row in metric_rows
+        ]
+
+    assert "missing" not in change_points
+    with pytest.raises(KeyError):
+        change_points.select_metrics("missing")
+    with pytest.raises(KeyError):
+        change_points.select_metrics([next(iter(expected_by_metric)), "missing"])
+    for invalid in ({}, ("errors",), 1, [1], ["missing", 1]):
+        with pytest.raises(TypeError):
+            change_points.select_metrics(invalid)
+    with pytest.raises(LookupError):
+        change_points.at_timestamp(expected[-1][0] + 1_000)
+    with pytest.raises(LookupError):
+        change_points.at_commit("missing")
+
+
 # BaseStats
 def test_basestats_calculate_means_and_std():
     s = BaseStats.calculate([10.0, 10.0, 10.0], [20.0, 20.0, 20.0], 0.02)
@@ -695,6 +820,102 @@ def test_select_metrics():
         metrics = cpbm.select_metrics({})
 
 
+# ChangePointsByTime / ChangePointsByMetric semantic parity
+def test_change_point_views_match_golden_history():
+    expected = [
+        (10.0, "commit-a", (("errors", "errors", 11), ("latency", "latency", 12))),
+        (20.0, "commit-b", (("throughput", "throughput", 21),)),
+        (30.0, "commit-c", (("latency", "latency", 31), ("throughput", "throughput", 32))),
+    ]
+    expected_by_metric = {
+        "errors": [(10.0, "commit-a", (("errors", "errors", 11),))],
+        "latency": [
+            (10.0, "commit-a", (("latency", "latency", 12),)),
+            (30.0, "commit-c", (("latency", "latency", 31),)),
+        ],
+        "throughput": [
+            (20.0, "commit-b", (("throughput", "throughput", 21),)),
+            (30.0, "commit-c", (("throughput", "throughput", 32),)),
+        ],
+    }
+    by_time = ChangePointsByTime.from_list(make_history_groups(GOLDEN_HISTORY))
+    by_metric = ChangePointsByMetric.from_list(make_history_groups(GOLDEN_HISTORY))
+
+    assert_read_api(by_time, expected, expected_by_metric)
+    assert_read_api(by_metric, expected, expected_by_metric)
+    assert history_snapshot(by_time) == history_snapshot(by_metric)
+
+
+@pytest.mark.parametrize(
+    ("view_class", "conversion"),
+    [
+        (ChangePointsByTime, "pivot"),
+        (ChangePointsByTime, "by_metric"),
+        (ChangePointsByMetric, "pivot"),
+        (ChangePointsByMetric, "by_time"),
+    ],
+)
+def test_change_point_view_conversion_mutation_boundaries(view_class, conversion):
+    source = view_class.from_list([make_group(1.0, "latency")])
+    converted = getattr(source, conversion)()
+
+    converted.at_timestamp(1.0)["latency"].stats.mean_1 = -1.0
+    assert source.at_timestamp(1.0)["latency"].stats.mean_1 == -1.0
+
+    isolated = getattr(source.copy(), conversion)()
+    isolated.at_timestamp(1.0)["latency"].stats.mean_1 = -2.0
+    assert source.at_timestamp(1.0)["latency"].stats.mean_1 == -1.0
+
+
+@settings(max_examples=75, deadline=None)
+@given(history=history_specs())
+def test_change_point_view_read_api_swarm(history):
+    expected = expected_history_snapshot(history)
+    expected_by_metric = expected_metric_snapshots(history)
+    by_time = ChangePointsByTime.from_list(make_history_groups(history))
+    by_metric = ChangePointsByMetric.from_list(make_history_groups(history))
+
+    assert_read_api(by_time, expected, expected_by_metric)
+    assert_read_api(by_metric, expected, expected_by_metric)
+    assert history_snapshot(by_time) == history_snapshot(by_metric)
+
+    for change_points in (by_time, by_metric):
+        clone = change_points.copy()
+        assert type(clone) is type(change_points)
+        assert history_snapshot(clone) == expected
+
+        first_metric = expected[0][2][0][0]
+        clone.at_timestamp(expected[0][0])[first_metric].stats.mean_1 = -999.0
+        assert change_points.at_timestamp(expected[0][0])[first_metric].stats.mean_1 != -999.0
+
+
+@settings(max_examples=75, deadline=None)
+@given(history=history_specs(), split_hint=st.integers(min_value=0, max_value=100))
+@example(history=GOLDEN_HISTORY, split_hint=1)
+def test_change_point_view_mutation_swarm(history, split_hint):
+    expected = expected_history_snapshot(history)
+    split = split_hint % (len(history) + 1)
+    views = []
+
+    for view_class in (ChangePointsByTime, ChangePointsByMetric):
+        change_points = view_class()
+        for row in history[:split]:
+            change_points.append(make_history_groups((row,))[0])
+        change_points.extend(make_history_groups(history[split:]))
+        assert history_snapshot(change_points) == expected
+        views.append(change_points)
+
+    assert history_snapshot(views[0]) == history_snapshot(views[1])
+
+    last_time, _, last_changes = history[-1]
+    last_metric = last_changes[0][0]
+    for change_points in views:
+        with pytest.raises(ValueError):
+            change_points.append(
+                make_group(last_time - 0.5, last_metric, commit="out-of-order")
+            )
+
+
 # SignificanceTester helpers
 def test_tester_compare_returns_basestats():
     tester = SignificanceTester(0.05)