Godstrument/test_fixes.py

#!/usr/bin/env python3
"""Self-check for the review fixes. Run: python3 test_fixes.py
Asserts the behaviours that were broken now hold. No framework, no fixtures."""
import math

import json
import os

from normalize import AdaptiveNormalizer, ImpulseEnvelope
from transform import Tweak, TweakBank
from hub import Signal, Hub, _safe_name, _ffloat


def test_control_signals_pass_through():
    # a knob/hand signal (midi./tof./hand./light.) with no norm spec must NOT be
    # adaptively rescaled (that pinned held knobs to 0.5 and froze the planet).
    s = Signal("midi.cc.20", {})
    assert s.norm.mode == "fixed", "control input should be fixed 0..1"
    assert abs(s.norm(1.0) - 1.0) < 1e-9 and abs(s.norm(0.0)) < 1e-9
    # a bare event (fire.event) must trigger at full magnitude, not 0.5
    e = Signal("fire.event", {"type": "event"})
    assert e.norm.mode == "fixed"
    assert abs(e.norm(1.0) - 1.0) < 1e-9, "constant-magnitude event must reach 1.0"
    # a world signal with no spec keeps adaptive behaviour
    w = Signal("crypto.vel", {})
    assert w.norm.mode == "minmax"


def test_normalizer_collapses_on_repeats():
    # identical values collapse to 0.5 (hi==lo) — this is WHY the hub must not
    # re-feed the window with a worker's re-emitted duplicates.
    n = AdaptiveNormalizer(100, "minmax")
    for _ in range(50):
        out = n(0.7)
    assert abs(out - 0.5) < 1e-9


def test_hub_value_dedup_gate():
    # Replicates the hub's inbox gate: workers re-emit their last value on a
    # cadence (fresh timestamp each time), so gating on arrival time would still
    # pad the window with duplicates. Gating on the VALUE keeps only real changes.
    n = AdaptiveNormalizer(600, "minmax")
    last = object()
    # 10 datagrams, each with a distinct arrival tick, but only 2 distinct values
    stream = [(100.0, i) for i in range(5)] + [(200.0, i) for i in range(5, 10)]
    for raw, _rt in stream:
        if last != raw:              # the gate hub.control_loop now uses
            n(raw)
            last = raw
    assert len(n.buf) == 2, f"re-emits leaked into the window: {len(n.buf)}"


def test_group_freeze_is_per_member():
    # one group Tweak applied to 3 members must hold EACH at its own value.
    g = Tweak({"freeze": 1.0})
    vals = {"a": 0.8, "b": 0.2, "c": 0.55}
    held = {k: g.apply(v, k) for k, v in vals.items()}
    assert held == vals, f"freeze must hold each member; got {held}"
    # and a second frame keeps holding the same per-member values
    held2 = {k: g.apply(0.0, k) for k in vals}
    assert held2 == vals, f"freeze must persist per member; got {held2}"


def test_group_smooth_is_per_member():
    g = Tweak({"smooth": 0.9})
    # push two members apart; each must chase its own target, not a shared blend
    for _ in range(200):
        a = g.apply(1.0, "a")
        b = g.apply(0.0, "b")
    assert a > 0.9 and b < 0.1, f"per-member smooth failed: a={a} b={b}"


def test_envelope_peak_hold():
    env = ImpulseEnvelope(attack=0.01, decay=1.0)
    env.trigger(1.0, t=0.0)
    big = env.value(0.1)            # a large event ringing
    env.trigger(0.1, t=0.1)        # a small event arrives mid-decay
    after = env.value(0.1)
    assert after >= big - 1e-9, f"small event chopped the big one: {big}->{after}"
    # a negative magnitude must never zero a ringing envelope
    env.trigger(-5.0, t=0.11)
    assert env.value(0.11) > 0.5, "negative-mag event zeroed the swell"
    # but when nothing is ringing, a fresh event still triggers
    env2 = ImpulseEnvelope(0.01, 1.0)
    env2.trigger(1.0, t=0.0)
    assert env2.value(0.02) > 0.9


def test_safe_name_blocks_traversal():
    assert _safe_name("../../etc/passwd") == "passwd"
    assert _safe_name("/absolute/secret") == "secret"
    assert _safe_name("my spell-2") == "my spell-2"
    assert _safe_name("..") == ""
    assert _safe_name(None) == ""


def test_ffloat_rejects_non_finite():
    assert _ffloat(float("nan"), 0.5) == 0.5
    assert _ffloat(float("inf"), 0.5) == 0.5
    assert _ffloat("1.25") == 1.25
    assert _ffloat(None, 0.3) == 0.3


def test_spell_saves_muted_sources():
    # a source muted by clicking its sphere must be considered "active" so
    # dump_state persists it into a spell.
    cfg = {"groups": {}, "tweaks": {}, "controls": []}
    tb = TweakBank(cfg)
    tb.toggle("weather.temp", "mute")            # mute a single source
    state = tb.dump_state()
    assert "weather.temp" in state["sources"], "muted source dropped from spell"


def test_regroup_drops_ghost_membership():
    cfg = {"groups": {"x": {"members": ["a", "b", "c", "d"]}},
           "tweaks": {}, "controls": []}
    tb = TweakBank(cfg)
    tb.groups["x"].set("mute", 1.0)              # mute the group
    tb.add_group("x", ["a", "b"])                # redefine with fewer members
    # c and d must no longer be silenced by group x
    assert not tb.is_muted("c"), "ghost membership kept muting a removed source"
    assert tb.is_muted("a"), "remaining member should still be muted"


def test_readonly_ignores_commands():
    # a public (read-only) hub must ignore every incoming ws command
    cfg = json.load(open(os.path.join(os.path.dirname(__file__), "config.json")))
    os.environ["GODSTRUMENT_READONLY"] = "1"
    try:
        h = Hub(cfg)
    finally:
        os.environ.pop("GODSTRUMENT_READONLY", None)
    assert h.readonly
    before = [dict(r) for r in h.matrix.routes]
    h._handle_cmd({"cmd": "add_route", "source": "x", "dest": "glitch", "amount": 1})
    h._handle_cmd({"cmd": "set", "target": "@planet", "param": "mute", "value": 1})
    h._handle_cmd({"cmd": "load_patch", "name": "../../etc/passwd"})
    assert [dict(r) for r in h.matrix.routes] == before, "readonly leaked a route change"
    assert not h.tweaks.groups["planet"].p["mute"], "readonly leaked a mute"


def main():
    tests = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
    for t in tests:
        t()
        print(f"  ok  {t.__name__}")
    print(f"\n{len(tests)} checks passed.")


if __name__ == "__main__":
    main()