MARTe-Integrated-Components/Common/Client/go/wshub/trigger_test.go

package wshub

import (
	"encoding/json"
	"math"
	"testing"
	"time"
)

func TestParseSignalKey(t *testing.T) {
	cases := []struct {
		in   string
		base string
		idx  int
	}{
		{"src:sig", "src:sig", -1},
		{"src:sig[0]", "src:sig", 0},
		{"src:sig[3]", "src:sig", 3},
		{"src:sig[x]", "src:sig[x]", -1},
		{"src:sig]", "src:sig]", -1},
	}
	for _, c := range cases {
		base, idx := parseSignalKey(c.in)
		if base != c.base || idx != c.idx {
			t.Errorf("parseSignalKey(%q) = (%q,%d), want (%q,%d)",
				c.in, base, idx, c.base, c.idx)
		}
	}
}

func armed(key, edge string, thr float64) *triggerEngine {
	te := newTriggerEngine()
	te.SetConfig(trigConfig{signalKey: key, edge: edge, threshold: thr,
		windowSec: 1, prePercent: 20, mode: "normal", holdoffSec: autoRearmDelaySec})
	te.Arm()
	return te
}

func TestFeedRisingEdge(t *testing.T) {
	te := armed("src:sig", "rising", 0.5)
	te.feed("src:sig", 1, []float64{1, 2, 3, 4}, []float64{0, 0.2, 0.9, 1.0})
	if te.State() != trigCollecting {
		t.Fatalf("state = %q, want collecting", te.State())
	}
	// Fires at the sample that crossed, i.e. t=3.
	trigTime, pre, post, ok := te.dueCapture(1e9)
	if !ok || trigTime != 3 {
		t.Fatalf("dueCapture = (%v,%v), want trigTime 3", trigTime, ok)
	}
	if pre != 0.2 || post != 0.8 {
		t.Errorf("pre/post = %v/%v, want 0.2/0.8", pre, post)
	}
}

func TestFeedFallingEdgeIgnoresRising(t *testing.T) {
	te := armed("src:sig", "falling", 0.5)
	te.feed("src:sig", 1, []float64{1, 2, 3}, []float64{0, 0.9, 1.0})
	if te.State() != trigArmed {
		t.Fatalf("state = %q, want armed (no falling edge)", te.State())
	}
	te.feed("src:sig", 1, []float64{4, 5}, []float64{0.6, 0.1})
	if te.State() != trigCollecting {
		t.Fatalf("state = %q, want collecting", te.State())
	}
}

func TestFeedIgnoresOtherSignals(t *testing.T) {
	te := armed("src:sig", "rising", 0.5)
	te.feed("src:other", 1, []float64{1, 2}, []float64{0, 1})
	if te.State() != trigArmed {
		t.Fatalf("state = %q, want armed", te.State())
	}
}

func TestFeedArrayElementSelection(t *testing.T) {
	// 2-element signal, element-major: [e0,e1, e0,e1, ...]. Only element 1
	// crosses the threshold.
	te := armed("src:sig[1]", "rising", 0.5)
	tt := []float64{1, 1, 2, 2}
	vv := []float64{0, 0, 0, 1}
	te.feed("src:sig", 2, tt, vv)
	if te.State() != trigCollecting {
		t.Fatalf("state = %q, want collecting", te.State())
	}

	// Element 0 never crosses, so a config on [0] must not fire.
	te2 := armed("src:sig[0]", "rising", 0.5)
	te2.feed("src:sig", 2, tt, vv)
	if te2.State() != trigArmed {
		t.Fatalf("state = %q, want armed", te2.State())
	}
}

func TestForceUsesLastSampleTime(t *testing.T) {
	te := newTriggerEngine()
	te.SetConfig(trigConfig{signalKey: "src:sig", edge: "rising", threshold: 1e9,
		windowSec: 2, prePercent: 50, mode: "single"})
	te.Arm()
	te.feed("src:sig", 1, []float64{10, 11, 12}, []float64{0, 0, 0})
	if te.State() != trigArmed {
		t.Fatalf("state = %q, want armed (threshold unreachable)", te.State())
	}
	te.Force()
	// post = 1 s, so the capture waits for samples past t = 12 + 1 + 0.15.
	te.feed("src:sig", 1, []float64{13.2}, []float64{0})
	trigTime, pre, post, ok := te.dueCapture(1e9)
	if !ok || trigTime != 12 || pre != 1 || post != 1 {
		t.Fatalf("dueCapture = (%v,%v,%v,%v), want (12,1,1,true)",
			trigTime, pre, post, ok)
	}
}

func TestForceFromIdle(t *testing.T) {
	te := newTriggerEngine()
	te.SetConfig(trigConfig{signalKey: "src:sig", edge: "rising",
		windowSec: 1, prePercent: 20, mode: "normal"})
	te.Force()
	if te.State() != trigCollecting {
		t.Fatalf("state = %q, want collecting", te.State())
	}
}

func TestCaptureMarginDelaysExtraction(t *testing.T) {
	te := armed("src:sig", "rising", 0.5)
	te.feed("src:sig", 1, []float64{0, 1}, []float64{0, 1}) // fires at t=1
	// post = 0.8 s; capture is due once the samples reach 1 + 0.8 + 0.15.
	te.feed("src:sig", 1, []float64{1.9}, []float64{0})
	if _, _, _, ok := te.dueCapture(1e9); ok {
		t.Error("capture extracted before the margin elapsed")
	}
	te.feed("src:sig", 1, []float64{1.96}, []float64{0})
	if _, _, _, ok := te.dueCapture(1e9); !ok {
		t.Error("capture not extracted after the margin elapsed")
	}
}

// A stream whose timestamps run behind real time must still yield the whole
// window: measuring the post-window on the wall clock cut the capture short by
// exactly the lag (an 8 s lag turned a 60 s window into a 36 s one).
func TestCaptureWaitsForLaggingStream(t *testing.T) {
	te := armed("src:sig", "rising", 0.5)
	te.feed("src:sig", 1, []float64{0, 1}, []float64{0, 1}) // fires at t=1
	wallNow := float64(time.Now().UnixNano()) / 1e9

	// Wall clock is far past the post-window, but the samples are not.
	te.feed("src:sig", 1, []float64{1.5}, []float64{0})
	if _, _, _, ok := te.dueCapture(wallNow); ok {
		t.Error("capture extracted while the stream was still short of the window")
	}
	te.feed("src:sig", 1, []float64{2.0}, []float64{0})
	if _, _, _, ok := te.dueCapture(wallNow); !ok {
		t.Error("capture not extracted once the samples covered the window")
	}
}

// A dead stream must not leave the client stuck in "collecting" forever.
func TestCaptureCompletesWhenStreamStalls(t *testing.T) {
	te := armed("src:sig", "rising", 0.5)
	te.feed("src:sig", 1, []float64{0, 1}, []float64{0, 1}) // fires at t=1
	wallNow := float64(time.Now().UnixNano()) / 1e9

	if _, _, _, ok := te.dueCapture(wallNow + captureStallSec/2); ok {
		t.Error("capture extracted before the stall timeout")
	}
	if _, _, _, ok := te.dueCapture(wallNow + captureStallSec + 0.1); !ok {
		t.Error("capture not extracted after the stream stalled")
	}
}

func TestAutoRearmNormalMode(t *testing.T) {
	te := armed("src:sig", "rising", 0.5)
	te.feed("src:sig", 1, []float64{0, 1}, []float64{0, 1})
	te.markTriggered(100)
	if te.State() != trigTriggered {
		t.Fatalf("state = %q, want triggered", te.State())
	}
	if te.dueRearm(100.1) {
		t.Error("rearmed before the delay elapsed")
	}
	if !te.dueRearm(100.3) {
		t.Error("did not rearm after the delay elapsed")
	}
	if te.dueRearm(200) {
		t.Error("rearm was not consumed")
	}
}

func TestNoAutoRearmInSingleMode(t *testing.T) {
	te := newTriggerEngine()
	te.SetConfig(trigConfig{signalKey: "src:sig", edge: "rising", threshold: 0.5,
		windowSec: 1, prePercent: 20, mode: "single"})
	te.Arm()
	te.feed("src:sig", 1, []float64{0, 1}, []float64{0, 1})
	te.markTriggered(100)
	if te.dueRearm(200) {
		t.Error("single mode must not auto-rearm")
	}
}

func TestStoppedSuppressesRearm(t *testing.T) {
	te := armed("src:sig", "rising", 0.5)
	te.feed("src:sig", 1, []float64{0, 1}, []float64{0, 1})
	te.SetStopped(true)
	te.markTriggered(100)
	if te.dueRearm(200) {
		t.Error("stopped engine must not rearm")
	}
}

func TestSetConfigClamps(t *testing.T) {
	te := newTriggerEngine()
	te.SetConfig(trigConfig{signalKey: "s:x", windowSec: 1000, prePercent: 500, holdoffSec: 120})
	if cfg := te.Config(); cfg.windowSec != 600 || cfg.prePercent != 100 || cfg.holdoffSec != 60 {
		t.Errorf("upper clamp = %v/%v/%v, want 600/100/60", cfg.windowSec, cfg.prePercent, cfg.holdoffSec)
	}
	// The web UI's longest option must survive intact — it used to be clamped
	// to 60 s, so a 10 min capture silently came back one minute long.
	te.SetConfig(trigConfig{signalKey: "s:x", windowSec: 600, prePercent: 20, holdoffSec: 1})
	if cfg := te.Config(); cfg.windowSec != 600 {
		t.Errorf("windowSec = %v, want the requested 600", cfg.windowSec)
	}
	te.SetConfig(trigConfig{signalKey: "s:x", windowSec: 0, prePercent: -5, holdoffSec: -1})
	if cfg := te.Config(); cfg.windowSec != 1e-4 || cfg.prePercent != 0 || cfg.holdoffSec != 0 {
		t.Errorf("lower clamp = %v/%v/%v, want 1e-4/0/0", cfg.windowSec, cfg.prePercent, cfg.holdoffSec)
	}
}

func TestHoldoffControlsRearmDelay(t *testing.T) {
	te := newTriggerEngine()
	te.SetConfig(trigConfig{signalKey: "src:sig", edge: "rising", threshold: 0.5,
		windowSec: 1, prePercent: 20, mode: "normal", holdoffSec: 5})
	te.Arm()
	te.feed("src:sig", 1, []float64{0, 1}, []float64{0, 1})
	te.markTriggered(100)
	if te.dueRearm(104.9) {
		t.Error("rearmed before the configured holdoff elapsed")
	}
	if !te.dueRearm(105.1) {
		t.Error("did not rearm after the configured holdoff elapsed")
	}
}

func TestActiveTracksConfiguredSignal(t *testing.T) {
	te := newTriggerEngine()
	if te.Active() {
		t.Error("a fresh engine must not be active")
	}
	te.SetConfig(trigConfig{signalKey: "src:sig", windowSec: 1})
	if !te.Active() {
		t.Error("engine must be active once a signal is configured")
	}
	// Rings must keep filling after a capture completes, not just while armed.
	te.Disarm()
	if !te.Active() {
		t.Error("engine must stay active after disarm while a signal is set")
	}
}

// The armed→collecting transition happens inside feed(), on the ingest path,
// which the hub runs before triggerTick in the same loop iteration. Clients need
// that state — it carries trigTime and the latched window, without which they
// cannot draw the window filling and sit frozen until the capture arrives.
func TestCollectingIsBroadcast(t *testing.T) {
	h := NewHub()
	h.trigger.SetConfig(trigConfig{signalKey: "s1:sig", edge: "rising", threshold: 0,
		windowSec: 10, prePercent: 20, mode: "single"})
	h.trigger.Arm()
	h.triggerTick()
	drainStates(t, h)

	// Fire, but stay well inside the post-trigger window: the capture is still
	// seconds away and this is exactly when the client has nothing to draw.
	h.ingest("s1:sig", 1, []float64{5.0, 5.001}, []float64{-1, 1})
	h.triggerTick()

	states := drainStates(t, h)
	found := false
	for _, m := range states {
		if m["state"] == trigCollecting {
			found = true
			if m["trigTime"] != 5.001 {
				t.Errorf("collecting broadcast has trigTime %v, want 5.001", m["trigTime"])
			}
			if m["preSec"] != 2.0 || m["postSec"] != 8.0 {
				t.Errorf("collecting broadcast has pre=%v post=%v, want 2 and 8",
					m["preSec"], m["postSec"])
			}
		}
	}
	if !found {
		t.Fatalf("no collecting broadcast after the trigger fired, got %v", states)
	}
}

// setFill hands the engine a buffer span and a growth rate, as the hub's
// per-tick measurements would: a reference point and a second one a second
// later. It forgets any earlier measurement first, so the rate is the one
// asked for rather than a blend with it.
func setFill(te *triggerEngine, span, growth, now float64) {
	te.setBuffered(0, 0, false, false, now)
	te.setBuffered(span-growth, 0, false, true, now)
	te.setBuffered(span, 0, false, true, now+1)
}

// What has to hold is that the buffer spans the whole window by the time the
// capture is read, one post-window after the trigger fires — so whatever it
// will fill in on its own during that time need not be there yet.
func TestFillNeed(t *testing.T) {
	cases := []struct {
		window, prePercent, growth, want float64
	}{
		{100, 20, 1, 20},   // still filling: only the pre-window has to exist
		{100, 20, 0.5, 60}, // half speed: 40 s of the 80 s post-window fills in
		{100, 20, 0, 100},  // not growing at all: it must already be all there
		{100, 0, 0.9, 10},  // no pre-window, but the buffer still has to keep up
		{100, 100, 1, 100}, // all pre-window: nothing fills in after the trigger
	}
	for _, c := range cases {
		te := newTriggerEngine()
		te.SetConfig(trigConfig{signalKey: "src:sig", windowSec: c.window, prePercent: c.prePercent})
		setFill(te, 1e6, c.growth, 100) // span large enough not to matter
		te.mu.Lock()
		got := te.fillNeedLocked()
		te.mu.Unlock()
		if math.Abs(got-c.want) > 1e-6 {
			t.Errorf("fillNeed(window %v, pre %v%%, growth %v) = %v, want %v",
				c.window, c.prePercent, c.growth, got, c.want)
		}
	}
}

// A trigger that fires before its pre-window has been buffered can only produce
// a capture whose front half never existed. It must wait instead.
func TestFillGateHoldsFire(t *testing.T) {
	te := armed("src:sig", "rising", 0.5) // window 1 s, pre 20 % → 0.2 s needed
	setFill(te, 0.05, 1, 100)
	te.feed("src:sig", 1, []float64{1, 2}, []float64{0, 1})
	if te.State() != trigArmed {
		t.Fatalf("state = %q, want armed: only 0.05 s of the 0.2 s pre-window is buffered", te.State())
	}
	// The level was still tracked, so the next crossing is a real edge and not a
	// re-detection of the one that was held off.
	setFill(te, 0.25, 1, 200)
	te.feed("src:sig", 1, []float64{3, 4}, []float64{1, 1})
	if te.State() != trigArmed {
		t.Fatalf("state = %q, want armed: no crossing, the signal stayed high", te.State())
	}
	te.feed("src:sig", 1, []float64{5, 6}, []float64{0, 1})
	if te.State() != trigCollecting {
		t.Fatalf("state = %q, want collecting once the pre-window is buffered", te.State())
	}
	te.feed("src:sig", 1, []float64{7, 8}, []float64{1, 1}) // carry the sample clock past the window
	if trigTime, _, _, ok := te.dueCapture(1e9); !ok || trigTime != 6 {
		t.Errorf("dueCapture = (%v,%v), want trigTime 6", trigTime, ok)
	}
}

// A ring that is full and re-bucketing for a longer window fills slower than
// real time — it drops dense old samples to take sparse new ones — so more of
// the window has to be there before an edge may be accepted.
func TestFillGateAccountsForSlowGrowth(t *testing.T) {
	te := armed("src:sig", "rising", 0.5) // window 1 s, pre 20 % → post 0.8 s
	// At half speed only 0.4 s of the post-window fills in, so 0.6 s is needed.
	setFill(te, 0.5, 0.5, 100)
	te.feed("src:sig", 1, []float64{1, 2}, []float64{0, 1})
	if te.State() != trigArmed {
		t.Fatalf("state = %q, want armed: 0.5 s buffered of the 0.6 s needed", te.State())
	}
	// The same 0.5 s in a ring still filling at full speed is plenty: everything
	// after the trigger is yet to be recorded anyway.
	te2 := armed("src:sig", "rising", 0.5)
	setFill(te2, 0.5, 1, 100)
	te2.feed("src:sig", 1, []float64{1, 2}, []float64{0, 1})
	if te2.State() != trigCollecting {
		t.Fatalf("state = %q, want collecting: the buffer keeps up with the stream", te2.State())
	}
	setFill(te, 0.65, 0.5, 200)
	te.feed("src:sig", 1, []float64{3, 4}, []float64{0, 1})
	if te.State() != trigCollecting {
		t.Fatalf("state = %q, want collecting once the buffer will span the window", te.State())
	}
}

func TestFillGateInactiveWithoutMeasurement(t *testing.T) {
	// No ring for the configured signal: gating would leave the trigger armed
	// forever, which is worse than a short capture.
	te := armed("src:sig", "rising", 0.5)
	te.feed("src:sig", 1, []float64{1, 2}, []float64{0, 1})
	if te.State() != trigCollecting {
		t.Fatalf("state = %q, want collecting: nothing measured, so nothing to gate on", te.State())
	}
	// Nor is there anything to wait for when the buffer keeps up and the whole
	// window is still to come.
	te = newTriggerEngine()
	te.SetConfig(trigConfig{signalKey: "src:sig", edge: "rising", threshold: 0.5,
		windowSec: 1, prePercent: 0, mode: "normal"})
	te.Arm()
	setFill(te, 0, 1, 100)
	te.feed("src:sig", 1, []float64{1, 2}, []float64{0, 1})
	if te.State() != trigCollecting {
		t.Fatalf("state = %q, want collecting with a 0 %% pre-window", te.State())
	}
}

// Force is the user overriding the trigger, so it overrides the gate too.
func TestForceIgnoresFillGate(t *testing.T) {
	te := armed("src:sig", "rising", 0.5)
	setFill(te, 0, 0, 100)
	te.Force()
	if te.State() != trigCollecting {
		t.Fatalf("state = %q, want collecting", te.State())
	}
}

// seedFillNow is setFill against the real clock, for tests that then let the
// hub take its own measurements: its ticks land inside the growth measurement
// interval, so they refresh the span and leave the seeded rate alone.
func seedFillNow(te *triggerEngine, span, growth float64) {
	now := float64(time.Now().UnixNano()) / 1e9
	te.setBuffered(0, 0, false, false, now-1)
	te.setBuffered(span-growth, 0, false, true, now-1)
	te.setBuffered(span, 0, false, true, now)
}

// While it holds off, the trigger looks identical to one that is ignoring
// edges. The state broadcast has to say it is filling, and keep saying so.
func TestFillProgressIsBroadcast(t *testing.T) {
	h := NewHub()
	rb := newSigRing(1000)
	h.rings["s1:sig"] = rb
	h.trigger.SetConfig(trigConfig{signalKey: "s1:sig", edge: "rising", threshold: 0,
		windowSec: 10, prePercent: 50, mode: "single"}) // 5 s of pre-window
	h.trigger.Arm()

	rb.write([]float64{0, 1}, []float64{-1, -1})
	// Filling at the rate of the stream, so only the pre-window is needed.
	seedFillNow(h.trigger, 1, 1)
	h.triggerTick()
	states := drainStates(t, h)
	if len(states) == 0 {
		t.Fatal("no state broadcast while the trigger was filling")
	}
	last := states[len(states)-1]
	if last["state"] != trigArmed {
		t.Fatalf("state = %v, want armed", last["state"])
	}
	if f, _ := last["bufferFill"].(float64); f < 0.19 || f > 0.21 {
		t.Errorf("bufferFill = %v, want ~0.2 (1 s of 5 s)", last["bufferFill"])
	}
	if last["bufferNeedSec"] != 5.0 {
		t.Errorf("bufferNeedSec = %v, want 5", last["bufferNeedSec"])
	}

	// An edge now is ignored: there is no 5 s of history to capture.
	h.ingest("s1:sig", 1, []float64{1.5, 2.0}, []float64{-1, 1})
	if h.trigger.State() != trigArmed {
		t.Fatalf("state = %q, want armed: the pre-window is only 20 %% buffered", h.trigger.State())
	}

	// Progress is news even though the state has not moved.
	rb.write([]float64{2, 3}, []float64{-1, -1})
	h.triggerTick()
	if states = drainStates(t, h); len(states) == 0 {
		t.Fatal("no state broadcast as the pre-window filled further")
	}
	if f, _ := states[len(states)-1]["bufferFill"].(float64); f < 0.59 || f > 0.61 {
		t.Errorf("bufferFill = %v, want ~0.6 (3 s of 5 s)", states[len(states)-1]["bufferFill"])
	}

	// Full: the gate opens, the fill disappears from the message and the next
	// edge fires.
	rb.write([]float64{4, 5.2}, []float64{-1, -1})
	h.triggerTick()
	states = drainStates(t, h)
	if len(states) == 0 {
		t.Fatal("no state broadcast when the pre-window filled")
	}
	if _, ok := states[len(states)-1]["bufferFill"]; ok {
		t.Errorf("bufferFill still reported once the pre-window is buffered: %v", states[len(states)-1])
	}
	h.ingest("s1:sig", 1, []float64{5.3, 5.4}, []float64{-1, 1})
	if h.trigger.State() != trigCollecting {
		t.Fatalf("state = %q, want collecting once the pre-window is buffered", h.trigger.State())
	}
}

// drainStates decodes every triggerState frame the hub has queued for
// broadcast. Hub.Run is what normally drains this queue, and it is not running
// in these tests.
func drainStates(t *testing.T, h *Hub) []map[string]any {
	t.Helper()
	var out []map[string]any
	for {
		select {
		case msg := <-h.broadcastCh:
			var m map[string]any
			if err := json.Unmarshal(msg, &m); err != nil {
				continue
			}
			if m["type"] == "triggerState" {
				out = append(out, m)
			}
		default:
			return out
		}
	}
}

// A ring whose coverage saturates below the window (measured source rate that
// over-estimates the true one) can never satisfy the full-window need. The
// coverage clamp must open the gate once the ring is full, delivering a short
// capture rather than staying deaf forever.
func TestFillNeedClampedToCoverage(t *testing.T) {
	te := newTriggerEngine()
	te.SetConfig(trigConfig{signalKey: "s:x", windowSec: 60, prePercent: 20, mode: "normal", holdoffSec: 0.2})
	setFill(te, 50, 0, 100) // ring full at 50 s, no growth
	te.mu.Lock()
	te.bufCoverage = 50 // the ring can never reach further back
	te.mu.Unlock()

	if need := te.fillNeedLocked(); need != 50 {
		t.Errorf("need = %v, want 50 (clamped to coverage, not the 60 s window)", need)
	}
	if f := te.fillLocked(); f < 1 {
		t.Errorf("fillLocked = %v, want >= 1: a full ring below the window must still open the gate", f)
	}

	// Without the clamp the gate would stay shut forever.
	te.mu.Lock()
	te.bufCoverage = 0
	te.mu.Unlock()
	if f := te.fillLocked(); f >= 1 {
		t.Errorf("baseline: fillLocked = %v, want < 1 without a coverage clamp", f)
	}
}

// When the disk archive already spans the window it can back-fill the front of
// a capture, so the gate must only require the ring to have reached the
// pre-window, not the whole window.
func TestFillNeedArchiveLowersToPreWindow(t *testing.T) {
	te := newTriggerEngine()
	te.SetConfig(trigConfig{signalKey: "s:x", windowSec: 60, prePercent: 20, mode: "normal", holdoffSec: 0.2})
	setFill(te, 30, 0, 100) // ring holds only 30 s, no growth → need 60 without archive
	te.mu.Lock()
	te.bufArchived = true
	te.mu.Unlock()

	if want := 12.0; te.fillNeedLocked() != want { // 60 * 0.20
		t.Errorf("need = %v, want %v (archive lowers to the pre-window)", te.fillNeedLocked(), want)
	}

	// A ring holding just the pre-window opens the gate once archived.
	te.mu.Lock()
	te.bufSpan = 12
	te.mu.Unlock()
	if f := te.fillLocked(); f < 1 {
		t.Errorf("fillLocked = %v, want >= 1 with pre-window buffered and the archive available", f)
	}
}