From 2585d3c71828f1ba147fe91c41f61a1284968f7b Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Sun, 10 May 2026 15:54:55 -0400 Subject: [PATCH] tests/pi_star_time_series: 35 KATs for temporal-signal kernel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Backfills partial-only coverage on arborist/pi_star/time_series.py (172 lines, 2026-05-10 zero-coverage sweep). KATs mirror the docstring's equivalence classes: timestamp/value jitter within Δ_t/Δ_v, out-of-order samples sort to canonical, duplicate timestamps collapse with last-value-wins. Pins banker's rounding (ties-to-even: 0.5→0, 1.5→2, 2.5→2). Pins int-vs-float dt/dv equivalence when integer-valued. Negative cones: missing required field, non-positive dt/dv (zero, negative, wrong type), malformed sample pairs, samples-not-array, non-object root, non-JSON, non- bytes. Plus projective contract: re-canonicalizing the text output raises PiStarError. --- tests/test_pi_star_time_series.py | 210 ++++++++++++++++++++++++++++++ 1 file changed, 210 insertions(+) create mode 100644 tests/test_pi_star_time_series.py diff --git a/tests/test_pi_star_time_series.py b/tests/test_pi_star_time_series.py new file mode 100644 index 0000000..2fdcd3e --- /dev/null +++ b/tests/test_pi_star_time_series.py @@ -0,0 +1,210 @@ +"""time-series-quantized@v1 π* tests. + +Temporal signal canonicalizer: JSON {dt, dv, samples} → quantized +`dt=...;dv=...;n=...;t0=...:v0|v1|...` text form. KATs mirror the +docstring's equivalence-class examples: jitter within Δ_t / Δ_v, +out-of-order samples, duplicate timestamps. Plus the invalid-input +cone (missing fields, non-positive dt/dv, malformed samples, +non-bytes, non-JSON, non-object root). + +Backfills the test-coverage gap for arborist/pi_star/time_series.py +identified during the 2026-05-10 zero-coverage sweep. +""" + +from __future__ import annotations + +import json + +import pytest + +from arborist.pi_star import get +from arborist.pi_star.protocol import PiStarError + + +@pytest.fixture +def ps(): + return get("time-series-quantized@v1") + + +def _doc(dt, dv, samples) -> bytes: + return json.dumps({"dt": dt, "dv": dv, "samples": samples}).encode("utf-8") + + +# --- registry presence + metadata ------------------------------------ + + +def test_registry_contains_time_series(): + ps = get("time-series-quantized@v1") + assert ps.name == "time-series-quantized" + assert ps.version == "v1" + assert ps.domain == "time-series" + + +# --- positive KATs --------------------------------------------------- + + +def test_basic_series(ps): + out = ps.canonicalize(_doc(1, 1, [[0, 0], [1, 1], [2, 2]])) + assert out == b"dt=1;dv=1;n=3;t0=0:0|1|2" + + +def test_empty_samples(ps): + out = ps.canonicalize(_doc(1, 1, [])) + assert out == b"dt=1;dv=1;n=0;t0=0:" + + +def test_single_sample(ps): + out = ps.canonicalize(_doc(1, 1, [[5, 42]])) + assert out == b"dt=1;dv=1;n=1;t0=5:42" + + +def test_float_dt_dv(ps): + out = ps.canonicalize(_doc(0.5, 0.1, [[0.0, 0.0], [0.5, 0.1], [1.0, 0.2]])) + # dt=0.5 with float repr; values quantized to int multiples of dv=0.1. + assert out.startswith(b"dt=0.5;dv=0.1;n=3;t0=0:") + assert b"0|1|2" in out + + +# --- equivalence classes preserved ----------------------------------- + + +def test_jitter_within_dt_collapses(ps): + """Timestamps within Δ_t/2 round to the same t_idx; last value wins.""" + a = ps.canonicalize(_doc(1, 1, [[0, 0], [1, 1]])) + b = ps.canonicalize(_doc(1, 1, [[0.2, 0], [1.1, 1]])) # jitter < dt/2 + assert a == b + + +def test_jitter_within_dv_collapses(ps): + """Values within Δ_v/2 quantize to the same v_idx.""" + a = ps.canonicalize(_doc(1, 1, [[0, 1], [1, 2]])) + b = ps.canonicalize(_doc(1, 1, [[0, 1.2], [1, 1.9]])) # values jitter < dv/2 + assert a == b + + +def test_out_of_order_samples_canonicalize(ps): + """Input order doesn't matter; samples are sorted by timestamp.""" + a = ps.canonicalize(_doc(1, 1, [[0, 10], [1, 20], [2, 30]])) + b = ps.canonicalize(_doc(1, 1, [[2, 30], [0, 10], [1, 20]])) + assert a == b + + +def test_duplicate_timestamps_last_wins(ps): + """Duplicate timestamps collapse — last value seen wins.""" + out = ps.canonicalize(_doc(1, 1, [[0, 10], [0, 20], [0, 30]])) + assert out == b"dt=1;dv=1;n=1;t0=0:30" + + +def test_int_vs_float_same_value(ps): + """1 (int) and 1.0 (float, integer-valued) encode identically for dt/dv.""" + a = ps.canonicalize(_doc(1, 1, [[0, 0]])) + b = ps.canonicalize(_doc(1.0, 1.0, [[0, 0]])) + assert a == b + + +# --- equivalence classes kept distinct ------------------------------- + + +def test_different_dt_distinct(ps): + a = ps.canonicalize(_doc(1, 1, [[0, 0], [1, 1]])) + b = ps.canonicalize(_doc(2, 1, [[0, 0], [1, 1]])) + assert a != b + + +def test_different_dv_distinct(ps): + a = ps.canonicalize(_doc(1, 1, [[0, 0], [1, 1]])) + b = ps.canonicalize(_doc(1, 2, [[0, 0], [1, 1]])) + assert a != b + + +def test_different_values_distinct(ps): + a = ps.canonicalize(_doc(1, 1, [[0, 1], [1, 2]])) + b = ps.canonicalize(_doc(1, 1, [[0, 1], [1, 3]])) + assert a != b + + +# --- rounding rule (banker's / ties-to-even) ------------------------- + + +def test_banker_rounding_half_to_even(ps): + """Python's `round()` uses ties-to-even — 0.5 → 0, 1.5 → 2.""" + out = ps.canonicalize(_doc(1, 1, [[0, 0.5], [1, 1.5], [2, 2.5]])) + # 0.5 → 0, 1.5 → 2, 2.5 → 2 + assert out == b"dt=1;dv=1;n=3;t0=0:0|2|2" + + +# --- determinism ----------------------------------------------------- + + +def test_repeated_canonicalize_is_stable(ps): + raw = _doc(0.1, 0.01, [[0.0, 1.23], [0.1, 4.56], [0.2, 7.89]]) + a = ps.canonicalize(raw) + b = ps.canonicalize(raw) + c = ps.canonicalize(raw) + assert a == b == c + + +# --- invalid-input cone ---------------------------------------------- + + +def test_non_bytes_raises(ps): + with pytest.raises(PiStarError, match="expects bytes"): + ps.canonicalize('{"dt":1,"dv":1,"samples":[]}') # type: ignore[arg-type] + + +def test_non_json_raises(ps): + with pytest.raises(PiStarError, match="not valid JSON"): + ps.canonicalize(b"not json at all") + + +def test_non_object_root_raises(ps): + with pytest.raises(PiStarError, match="must be a JSON object"): + ps.canonicalize(b"[1, 2, 3]") + with pytest.raises(PiStarError, match="must be a JSON object"): + ps.canonicalize(b"42") + + +@pytest.mark.parametrize("missing", ["dt", "dv", "samples"]) +def test_missing_field_raises(ps, missing): + obj = {"dt": 1, "dv": 1, "samples": []} + obj.pop(missing) + with pytest.raises(PiStarError, match=f"missing required field {missing!r}"): + ps.canonicalize(json.dumps(obj).encode()) + + +@pytest.mark.parametrize("bad_dt", [0, -1, -0.5, "1", None, [1]]) +def test_non_positive_dt_raises(ps, bad_dt): + obj = {"dt": bad_dt, "dv": 1, "samples": []} + with pytest.raises(PiStarError, match="dt must be positive"): + ps.canonicalize(json.dumps(obj).encode()) + + +@pytest.mark.parametrize("bad_dv", [0, -1, -0.5, "1", None]) +def test_non_positive_dv_raises(ps, bad_dv): + obj = {"dt": 1, "dv": bad_dv, "samples": []} + with pytest.raises(PiStarError, match="dv must be positive"): + ps.canonicalize(json.dumps(obj).encode()) + + +def test_samples_not_array_raises(ps): + obj = {"dt": 1, "dv": 1, "samples": "not an array"} + with pytest.raises(PiStarError, match="samples must be a JSON array"): + ps.canonicalize(json.dumps(obj).encode()) + + +def test_malformed_sample_pair_raises(ps): + """Each sample must be a [timestamp, value] pair of exactly 2 numbers.""" + for bad_sample in ([1], [1, 2, 3], "not_a_pair", 42, [None, 1], [1, "v"]): + obj = {"dt": 1, "dv": 1, "samples": [bad_sample]} + with pytest.raises(PiStarError): + ps.canonicalize(json.dumps(obj).encode()) + + +# --- projective (not invertible) ------------------------------------- + + +def test_canonical_output_is_not_json(ps): + """Re-canonicalizing the canonical text form raises — by design.""" + canonical = ps.canonicalize(_doc(1, 1, [[0, 0]])) + with pytest.raises(PiStarError, match="not valid JSON"): + ps.canonicalize(canonical)