tests/pi_star_time_series: 35 KATs for temporal-signal kernel
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.
This commit is contained in:
parent
593550b1e3
commit
2585d3c718
1 changed files with 210 additions and 0 deletions
210
tests/test_pi_star_time_series.py
Normal file
210
tests/test_pi_star_time_series.py
Normal file
|
|
@ -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)
|
||||
Loading…
Add table
Add a link
Reference in a new issue