uopi/internal/controllogic/engine_reload_test.go

package controllogic

import (
	"context"
	"io"
	"log/slog"
	"sync"
	"testing"
	"time"

	"github.com/uopi/uopi/internal/audit"
	"github.com/uopi/uopi/internal/broker"
	"github.com/uopi/uopi/internal/datasource"
)

// pushSource is a DataSource whose Subscribe captures the broker's delivery
// channel per signal, letting a test push successive values to drive level/edge
// triggers. Writes are recorded so action.write effects can be asserted.
type pushSource struct {
	mu      sync.Mutex
	chans   map[string]chan<- datasource.Value
	written map[string]any
}

func newPushSource() *pushSource {
	return &pushSource{chans: map[string]chan<- datasource.Value{}, written: map[string]any{}}
}

func (s *pushSource) Name() string                  { return "tgt" }
func (s *pushSource) Connect(context.Context) error { return nil }
func (s *pushSource) ListSignals(context.Context) ([]datasource.Metadata, error) {
	return nil, nil
}
func (s *pushSource) GetMetadata(_ context.Context, sig string) (datasource.Metadata, error) {
	return datasource.Metadata{Name: sig, Writable: true}, nil
}
func (s *pushSource) Subscribe(_ context.Context, sig string, ch chan<- datasource.Value) (datasource.CancelFunc, error) {
	s.mu.Lock()
	s.chans[sig] = ch
	s.mu.Unlock()
	return func() {}, nil
}
func (s *pushSource) Write(_ context.Context, signal string, value any) error {
	s.mu.Lock()
	s.written[signal] = value
	s.mu.Unlock()
	return nil
}
func (s *pushSource) History(context.Context, string, time.Time, time.Time, int) ([]datasource.Value, error) {
	return nil, datasource.ErrHistoryUnavailable
}

func (s *pushSource) chanFor(sig string) (chan<- datasource.Value, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	ch, ok := s.chans[sig]
	return ch, ok
}

func (s *pushSource) get(sig string) (any, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	v, ok := s.written[sig]
	return v, ok
}

// TestEngineReloadThresholdTrigger drives a real Reload generation: a
// trigger.threshold on tgt:IN (>5) wired to an action.write of 42 to tgt:OUT.
// Pushing IN below then above the threshold must fire exactly the rising edge.
func TestEngineReloadThresholdTrigger(t *testing.T) {
	log := slog.New(slog.NewTextHandler(io.Discard, nil))
	ctx := t.Context()

	src := newPushSource()
	brk := broker.New(ctx, log)
	brk.Register(src)

	store, err := NewStore(t.TempDir())
	if err != nil {
		t.Fatal("NewStore:", err)
	}
	g := Graph{
		Name:    "watchdog",
		Enabled: true,
		Nodes: []Node{
			{ID: "t1", Kind: "trigger.threshold", Params: map[string]string{
				"signal": "tgt:IN", "op": ">", "value": "5",
			}},
			{ID: "a1", Kind: "action.write", Params: map[string]string{
				"target": "tgt:OUT", "expr": "42",
			}},
		},
		Wires: []Wire{{From: "t1", To: "a1"}},
	}
	if err := store.Save(g); err != nil {
		t.Fatal("Save:", err)
	}

	e := NewEngine(ctx, brk, store, nil, audit.Nop(), log)
	e.Reload()
	// t.Context() is cancelled at test cleanup, tearing the generation down.

	// Wait until the engine has subscribed to tgt:IN.
	var ch chan<- datasource.Value
	waitFor(t, time.Second, func() bool {
		c, ok := src.chanFor("IN")
		if ok {
			ch = c
		}
		return ok
	})

	// Below threshold: no fire (prev state seeds to false).
	ch <- datasource.Value{Data: 0.0, Timestamp: time.Now()}
	// Rising edge above threshold: fires the action.
	ch <- datasource.Value{Data: 10.0, Timestamp: time.Now()}

	waitFor(t, 2*time.Second, func() bool {
		v, ok := src.get("OUT")
		return ok && toNum(v) == 42
	})
	if v, ok := src.get("OUT"); !ok || toNum(v) != 42 {
		t.Fatalf("OUT = %v (ok=%v), want 42 after rising-edge trigger", v, ok)
	}
}

// TestEngineReloadTimerIfWrite covers the timer trigger, startTriggers, and a
// flow.if then-branch driving an action.write.
func TestEngineReloadTimerIfWrite(t *testing.T) {
	log := slog.New(slog.NewTextHandler(io.Discard, nil))
	ctx := t.Context()

	src := newPushSource()
	brk := broker.New(ctx, log)
	brk.Register(src)

	store, err := NewStore(t.TempDir())
	if err != nil {
		t.Fatal("NewStore:", err)
	}
	g := Graph{
		Name:    "ticker",
		Enabled: true,
		Nodes: []Node{
			{ID: "t1", Kind: "trigger.timer", Params: map[string]string{"interval": "50"}},
			{ID: "if1", Kind: "flow.if", Params: map[string]string{"cond": "2 > 1"}},
			{ID: "a1", Kind: "action.write", Params: map[string]string{"target": "tgt:OUT2", "expr": "7"}},
		},
		Wires: []Wire{
			{From: "t1", To: "if1"},
			{From: "if1", FromPort: "then", To: "a1"},
		},
	}
	if err := store.Save(g); err != nil {
		t.Fatal("Save:", err)
	}

	e := NewEngine(ctx, brk, store, nil, audit.Nop(), log)
	e.Reload()

	waitFor(t, 2*time.Second, func() bool {
		v, ok := src.get("OUT2")
		return ok && toNum(v) == 7
	})
}

// waitFor polls cond until it returns true or the deadline elapses.
func waitFor(t *testing.T, d time.Duration, cond func() bool) {
	t.Helper()
	deadline := time.Now().Add(d)
	for time.Now().Before(deadline) {
		if cond() {
			return
		}
		time.Sleep(5 * time.Millisecond)
	}
	if !cond() {
		t.Fatalf("condition not met within %s", d)
	}
}