279 lines
8.5 KiB
Go
279 lines
8.5 KiB
Go
package wshub
|
|
|
|
import (
|
|
"encoding/binary"
|
|
"math"
|
|
"testing"
|
|
)
|
|
|
|
// fillRing writes n samples at the given rate starting at t0.
|
|
func fillRing(rb *sigRing, t0 float64, rate float64, n int) {
|
|
ts := make([]float64, n)
|
|
vs := make([]float64, n)
|
|
for i := range ts {
|
|
ts[i] = t0 + float64(i)/rate
|
|
vs[i] = math.Sin(float64(i))
|
|
}
|
|
rb.write(ts, vs)
|
|
}
|
|
|
|
func TestRingGrowPreservesSamples(t *testing.T) {
|
|
rb := newSigRing(100)
|
|
// Overflow the ring so the retained window starts mid-buffer.
|
|
fillRing(rb, 0, 1000, 250)
|
|
|
|
beforeT, beforeV := rb.slice(-1e9, 1e9)
|
|
if len(beforeT) != 100 {
|
|
t.Fatalf("pre-grow fill = %d, want 100", len(beforeT))
|
|
}
|
|
if !rb.grow(1000) {
|
|
t.Fatal("grow(1000) returned false")
|
|
}
|
|
if rb.capacity() != 1000 {
|
|
t.Fatalf("capacity = %d, want 1000", rb.capacity())
|
|
}
|
|
afterT, afterV := rb.slice(-1e9, 1e9)
|
|
if len(afterT) != len(beforeT) {
|
|
t.Fatalf("post-grow fill = %d, want %d", len(afterT), len(beforeT))
|
|
}
|
|
for i := range beforeT {
|
|
if afterT[i] != beforeT[i] || afterV[i] != beforeV[i] {
|
|
t.Fatalf("sample %d changed across grow", i)
|
|
}
|
|
}
|
|
|
|
// Further writes must keep landing in order rather than wrapping early.
|
|
fillRing(rb, 1.0, 1000, 500)
|
|
if n, _ := rb.stats(); n != 600 {
|
|
t.Fatalf("fill after grow = %d, want 600", n)
|
|
}
|
|
|
|
// Shrinking is refused.
|
|
if rb.grow(10) {
|
|
t.Fatal("grow(10) shrank the ring")
|
|
}
|
|
}
|
|
|
|
func TestRingStatsMeasuresRate(t *testing.T) {
|
|
rb := newSigRing(10000)
|
|
fillRing(rb, 0, 1000, 1000) // 1 kHz
|
|
n, span := rb.stats()
|
|
if n != 1000 {
|
|
t.Fatalf("count = %d, want 1000", n)
|
|
}
|
|
rate := float64(n) / span
|
|
if math.Abs(rate-1001) > 5 { // n samples span (n-1) intervals
|
|
t.Fatalf("rate = %v, want ~1000", rate)
|
|
}
|
|
}
|
|
|
|
// A long trigger window must grow the rings to hold it: a fixed sample-count
|
|
// ring covers a fraction of a second at a high rate, which is what made 60 s
|
|
// captures come back with only their tail populated.
|
|
func TestRetuneRingsCoversTriggerWindow(t *testing.T) {
|
|
h := NewHub()
|
|
rb := newSigRing(6000) // 6 s at 1 kHz — far short of a 60 s window
|
|
fillRing(rb, 0, 1000, 6000)
|
|
h.rings["s1:sig"] = rb
|
|
|
|
h.trigger.SetConfig(trigConfig{signalKey: "s1:sig", edge: "rising",
|
|
windowSec: 60, prePercent: 20, mode: "normal"})
|
|
|
|
h.retuneRings(1000)
|
|
|
|
// 60 s at 1 kHz is 60 k samples: growing to the budget holds them verbatim.
|
|
if got := rb.capacity(); got < 60000 {
|
|
t.Fatalf("capacity = %d, want >= 60000 to hold a 60 s window", got)
|
|
}
|
|
if got := rb.bucketSize(); got != 1 {
|
|
t.Fatalf("bucket = %d, want 1: the window fits at full rate", got)
|
|
}
|
|
}
|
|
|
|
// Past the budget the window is kept by reducing resolution, not by dropping
|
|
// its head — the whole point of the min/max buckets.
|
|
func TestRetuneRingsBucketsWhenTheWindowExceedsTheBudget(t *testing.T) {
|
|
h := NewHub()
|
|
rb := newSigRing(1000)
|
|
fillRing(rb, 0, 1e6, 100_000) // 1 MSps
|
|
h.rings["s1:sig"] = rb
|
|
|
|
h.trigger.SetConfig(trigConfig{signalKey: "s1:sig", windowSec: 60, mode: "normal"})
|
|
h.retuneRings(1000)
|
|
|
|
if got := rb.capacity(); got != defaultRingPts {
|
|
t.Fatalf("capacity = %d, want the budget %d", got, defaultRingPts)
|
|
}
|
|
// 60 s at 1 MSps is 60 M samples in a 10 M-point buffer, so each stored
|
|
// pair must cover at least 12 source samples.
|
|
bucket := rb.bucketSize()
|
|
if bucket < 12 {
|
|
t.Fatalf("bucket = %d, too fine to fit 60 M samples in %d points", bucket, rb.capacity())
|
|
}
|
|
if covered := float64(rb.capacity()) / 2 * float64(bucket) / 1e6; covered < 60 {
|
|
t.Fatalf("buffer covers %.1f s, want the whole 60 s window", covered)
|
|
}
|
|
}
|
|
|
|
// A raised budget buys resolution back: the same window is held verbatim.
|
|
func TestRetuneRingsHonoursRaisedBudget(t *testing.T) {
|
|
h := NewHub()
|
|
h.SetRingBudget(80_000_000)
|
|
rb := newSigRing(1000)
|
|
fillRing(rb, 0, 1e6, 100_000) // 1 MSps
|
|
h.rings["s1:sig"] = rb
|
|
|
|
h.trigger.SetConfig(trigConfig{signalKey: "s1:sig", windowSec: 60, mode: "normal"})
|
|
h.retuneRings(1000)
|
|
|
|
if got := rb.bucketSize(); got != 1 {
|
|
t.Fatalf("bucket = %d, want 1: 60 M samples fit in an 80 M-point buffer", got)
|
|
}
|
|
}
|
|
|
|
func TestSetRingBudgetBounds(t *testing.T) {
|
|
h := NewHub()
|
|
h.SetRingBudget(0)
|
|
if got := h.ringBudget(); got != defaultRingPts {
|
|
t.Fatalf("ringBudget after 0 = %d, want the default %d", got, defaultRingPts)
|
|
}
|
|
// Never below the depth a freshly configured ring already has, or the
|
|
// budget would ask for a shrink the ring refuses anyway.
|
|
h.SetRingBudget(10)
|
|
if got := h.ringBudget(); got != ringCapInitial {
|
|
t.Fatalf("ringBudget after 10 = %d, want the floor %d", got, ringCapInitial)
|
|
}
|
|
}
|
|
|
|
func TestRetuneRingsIsThrottled(t *testing.T) {
|
|
h := NewHub()
|
|
h.SetRingBudget(250_000)
|
|
rb := newSigRing(250_000)
|
|
fillRing(rb, 0, 1e6, 100_000)
|
|
h.rings["s1:sig"] = rb
|
|
h.trigger.SetConfig(trigConfig{signalKey: "s1:sig", windowSec: 10, mode: "normal"})
|
|
|
|
h.retuneRings(100)
|
|
first := rb.bucketSize()
|
|
if first <= 1 {
|
|
t.Fatalf("bucket = %d, expected a reduction for 10 s at 1 MSps in 250 k points", first)
|
|
}
|
|
// Same second: the sweep must not run again even though a bigger window
|
|
// is now configured.
|
|
h.trigger.SetConfig(trigConfig{signalKey: "s1:sig", windowSec: 600, mode: "normal"})
|
|
h.retuneRings(100.5)
|
|
if rb.bucketSize() != first {
|
|
t.Fatalf("sweep ran inside the throttle window")
|
|
}
|
|
h.retuneRings(200)
|
|
if rb.bucketSize() <= first {
|
|
t.Fatalf("sweep did not run after the throttle window elapsed")
|
|
}
|
|
}
|
|
|
|
// With no trigger armed and no client saying otherwise, the rings are sized for
|
|
// the default live window — live mode needs the buffers just as much as a
|
|
// capture does.
|
|
func TestRetuneRingsSizesForTheLiveWindow(t *testing.T) {
|
|
h := NewHub()
|
|
rb := newSigRing(1000)
|
|
fillRing(rb, 0, 1e6, 100_000) // 1 MSps: 10 s does not fit in 1000 points
|
|
h.rings["s1:sig"] = rb
|
|
// No signal configured → trigger inactive, so the live window governs.
|
|
h.trigger.SetConfig(trigConfig{windowSec: 600, mode: "normal"})
|
|
|
|
h.retuneRings(100)
|
|
|
|
if got := rb.capacity(); got != defaultRingPts {
|
|
t.Fatalf("capacity = %d, want the budget %d", got, defaultRingPts)
|
|
}
|
|
// defaultLiveWindowSec at 1 MSps is exactly the budget, so no reduction.
|
|
if got := rb.bucketSize(); got != 1 {
|
|
t.Fatalf("bucket = %d, want 1 for the default live window", got)
|
|
}
|
|
}
|
|
|
|
func TestRingBucketForCoversTheWindow(t *testing.T) {
|
|
cases := []struct {
|
|
rate, window float64
|
|
capacity int
|
|
want int
|
|
}{
|
|
{1000, 10, 1_000_000, 1}, // 10 k samples in 1 M points: verbatim
|
|
{1e6, 10, 10_000_000, 1}, // exactly the budget: still verbatim
|
|
{1e6, 60, 10_000_000, 15}, // 60 M samples, 1.25x headroom
|
|
{1e6, 600, 10_000_000, 150}, // 600 s still fits, at 1/150 resolution
|
|
{0, 10, 1_000_000, 1}, // no rate measured yet
|
|
{1000, 0, 1_000_000, 1}, // no window
|
|
}
|
|
for _, c := range cases {
|
|
if got := ringBucketFor(c.rate, c.window, c.capacity); got != c.want {
|
|
t.Errorf("ringBucketFor(%v, %v, %d) = %d, want %d",
|
|
c.rate, c.window, c.capacity, got, c.want)
|
|
}
|
|
}
|
|
}
|
|
|
|
// decodeCapture pulls the per-signal point counts out of a v2 capture frame.
|
|
func decodeCapture(t *testing.T, buf []byte) map[string]int {
|
|
t.Helper()
|
|
if buf[0] != 2 {
|
|
t.Fatalf("frame version = %d, want 2", buf[0])
|
|
}
|
|
off := 1 + 8 + 8 + 8
|
|
nSig := int(binary.LittleEndian.Uint32(buf[off:]))
|
|
off += 4
|
|
out := make(map[string]int, nSig)
|
|
for i := 0; i < nSig; i++ {
|
|
kl := int(binary.LittleEndian.Uint16(buf[off:]))
|
|
off += 2
|
|
key := string(buf[off : off+kl])
|
|
off += kl
|
|
n := int(binary.LittleEndian.Uint32(buf[off:]))
|
|
off += 4
|
|
off += n * 16
|
|
out[key] = n
|
|
}
|
|
if off != len(buf) {
|
|
t.Fatalf("decoded %d of %d bytes", off, len(buf))
|
|
}
|
|
return out
|
|
}
|
|
|
|
// A 60 s window at a high rate is hundreds of megabytes raw; the capture frame
|
|
// must be decimated so it can actually reach a client.
|
|
func TestBuildTriggerCaptureDecimates(t *testing.T) {
|
|
h := NewHub()
|
|
rb := newSigRing(200000)
|
|
fillRing(rb, 0, 100000, 200000) // 2 s at 100 kSps
|
|
h.rings["s1:sig"] = rb
|
|
|
|
buf := h.buildTriggerCapture(1.0, 1.0, 1.0)
|
|
if buf == nil {
|
|
t.Fatal("no capture frame built")
|
|
}
|
|
counts := decodeCapture(t, buf)
|
|
n := counts["s1:sig"]
|
|
if n != trigCapturePts {
|
|
t.Fatalf("captured %d points, want the %d-point cap", n, trigCapturePts)
|
|
}
|
|
}
|
|
|
|
// Short captures must stay full resolution — decimation only kicks in above
|
|
// the cap.
|
|
func TestBuildTriggerCaptureKeepsSmallWindowsIntact(t *testing.T) {
|
|
h := NewHub()
|
|
rb := newSigRing(10000)
|
|
fillRing(rb, 0, 1000, 10000) // 10 s at 1 kHz
|
|
h.rings["s1:sig"] = rb
|
|
|
|
buf := h.buildTriggerCapture(1.0, 0.5, 0.5)
|
|
if buf == nil {
|
|
t.Fatal("no capture frame built")
|
|
}
|
|
counts := decodeCapture(t, buf)
|
|
if n := counts["s1:sig"]; n < 990 || n > 1010 {
|
|
t.Fatalf("captured %d points, want ~1000 undecimated", n)
|
|
}
|
|
}
|