MARTe-Integrated-Components/Test/E2E/suite/client/configcheck/main.go

// configcheck exercises the calibration and config-persistence WebSocket frames
// against a StreamHub (either the Go hub or the C++ StreamHub) and exits
// non-zero if the hub's replies do not match the protocol.
//
// Frame-strictness rule
// ─────────────────────
// Frames are divided into two categories:
//
//   Protocol frames — part of the five command→response sequences under test:
//     "calibration", "sources", "configSaved", "configReloaded"
//   Ambient frames  — unsolicited live traffic the hubs push independently:
//     "data", "stats", "triggerState", "monotonicState" (and any unknown type)
//
// The checker is strict about protocol frames: if one arrives when a different
// protocol frame is expected, that is an error (out-of-order or unexpected).
// Ambient frames are logged and skipped without failing the check.
//
// Known documented exception: C++ HandleReloadConfig() emits a "sources" frame
// after "calibration", while the Go hub does not (because Go propagates source
// changes per-add, already broadcasting "sources" when sources are added).
// Both behaviours are correct; the extra "sources" frame from C++ is accepted
// and logged as a deliberate known difference.
package main

import (
	"encoding/json"
	"flag"
	"fmt"
	"os"
	"sort"
	"time"

	"github.com/gorilla/websocket"
)

type calEntry struct {
	Source string  `json:"source"`
	Signal string  `json:"signal"`
	Scale  float64 `json:"scale"`
	Offset float64 `json:"offset"`
	Unit   string  `json:"unit"`
}

type frame struct {
	Type  string     `json:"type"`
	Cal   []calEntry `json:"cal"`
	OK    bool       `json:"ok"`
	Path  string     `json:"path"`
	Error string     `json:"error"`
}

// protocolFrameTypes is the set of frame types that are part of the protocol
// sequences under test.  Any frame whose type is in this set but is not the
// one currently expected causes an immediate failure.  Frames with types NOT
// in this set are ambient traffic and are silently skipped.
var protocolFrameTypes = map[string]bool{
	"calibration":    true,
	"sources":        true,
	"configSaved":    true,
	"configReloaded": true,
}

type conn struct {
	ws      *websocket.Conn
	timeout time.Duration
	// readCh is driven by a long-lived background goroutine; nil until startReader.
	readCh chan readResult
}

type readResult struct {
	f   frame
	err error
}

// startReader launches a background goroutine that reads all text frames from
// the WebSocket and forwards them on readCh.  This avoids setting a read
// deadline on the underlying connection, which permanently poisons gorilla
// websocket v1.5.1 after a timeout fires.
func (c *conn) startReader() {
	c.readCh = make(chan readResult, 32)
	go func() {
		for {
			mt, data, err := c.ws.ReadMessage()
			if err != nil {
				c.readCh <- readResult{err: fmt.Errorf("ws read: %w", err)}
				return
			}
			if mt != websocket.TextMessage {
				continue
			}
			var f frame
			if jsonErr := json.Unmarshal(data, &f); jsonErr != nil {
				continue
			}
			c.readCh <- readResult{f: f}
		}
	}()
}

// next reads from the background reader, skipping ambient frames, until a
// protocol frame arrives or the deadline passes.
//
// If a protocol frame arrives that is NOT the expected one, next returns an
// error immediately — that is the out-of-order/unexpected signal.
func (c *conn) next(want string) (frame, error) {
	return c.nextSkipping(want, nil)
}

// nextSkipping is like next but also skips (logs and discards) protocol frames
// whose type is in also.  This is used for the initial-connect step where C++
// emits "sources" before "calibration" as part of its state-push sequence,
// whereas the Go hub emits "calibration" first.  Passing also=[]string{"sources"}
// lets the checker accept either ordering without letting the connect-time
// sources frame go completely unnoticed.
func (c *conn) nextSkipping(want string, also []string) (frame, error) {
	isSkip := make(map[string]bool, len(also))
	for _, t := range also {
		isSkip[t] = true
	}
	deadline := time.NewTimer(c.timeout)
	defer deadline.Stop()
	for {
		select {
		case <-deadline.C:
			return frame{}, fmt.Errorf("timeout waiting for %q", want)
		case r, ok := <-c.readCh:
			if !ok {
				return frame{}, fmt.Errorf("reader closed while waiting for %q", want)
			}
			if r.err != nil {
				return frame{}, fmt.Errorf("read while waiting for %q: %w", want, r.err)
			}
			if r.f.Type == want {
				return r.f, nil
			}
			// Explicitly-skipped protocol frames (known ordering differences).
			if isSkip[r.f.Type] {
				fmt.Printf("[skip allowed protocol frame %q while waiting for %q]\n", r.f.Type, want)
				continue
			}
			// Is this an unexpected protocol frame (out-of-order)?
			if protocolFrameTypes[r.f.Type] {
				return frame{}, fmt.Errorf(
					"unexpected protocol frame %q while waiting for %q (out-of-order or spurious emission)",
					r.f.Type, want)
			}
			// Ambient frame — log and skip.
			fmt.Printf("[skip ambient %q]\n", r.f.Type)
		}
	}
}

// nextOrOptional reads from the background reader, skipping ambient frames,
// and returns (frame, frametype) where frametype is the type of the first
// protocol frame that arrives, regardless of whether it matches want.
// If the optional type arrives instead, that is returned too.
// This is used for the reload sequence where C++ may emit an extra "sources"
// frame after "calibration".
func (c *conn) nextOneOf(want, optional string) (frame, string, error) {
	deadline := time.NewTimer(c.timeout)
	defer deadline.Stop()
	for {
		select {
		case <-deadline.C:
			return frame{}, "", fmt.Errorf("timeout waiting for %q (or %q)", want, optional)
		case r, ok := <-c.readCh:
			if !ok {
				return frame{}, "", fmt.Errorf("reader closed while waiting for %q", want)
			}
			if r.err != nil {
				return frame{}, "", fmt.Errorf("read while waiting for %q: %w", want, r.err)
			}
			if r.f.Type == want || r.f.Type == optional {
				return r.f, r.f.Type, nil
			}
			// Any other protocol frame is unexpected.
			if protocolFrameTypes[r.f.Type] {
				return frame{}, "", fmt.Errorf(
					"unexpected protocol frame %q while waiting for %q or %q",
					r.f.Type, want, optional)
			}
			fmt.Printf("[skip ambient %q]\n", r.f.Type)
		}
	}
}

// nextWithin reads from the background reader until a frame with the wanted
// type arrives within d, returning (frame, true) or (frame{}, false).  Unlike
// next() it does NOT return an error on timeout, making it suitable for the
// "must NOT arrive" assertion.  Protocol frames with wrong type still skip
// (they will be picked up by the next next() call from the buffer).
func (c *conn) nextWithin(want string, d time.Duration) (frame, bool) {
	deadline := time.NewTimer(d)
	defer deadline.Stop()
	for {
		select {
		case <-deadline.C:
			return frame{}, false
		case r, ok := <-c.readCh:
			if !ok || r.err != nil {
				return frame{}, false
			}
			if r.f.Type == want {
				return r.f, true
			}
			// For the "must not arrive" check we skip everything else
			// (ambient and other protocol frames alike) — we're only
			// interested in whether the specific type appears.
		}
	}
}

func (c *conn) send(v interface{}) error {
	data, err := json.Marshal(v)
	if err != nil {
		return err
	}
	return c.ws.WriteMessage(websocket.TextMessage, data)
}

func findCal(list []calEntry, source, signal string) (calEntry, bool) {
	for _, e := range list {
		if e.Source == source && e.Signal == signal {
			return e, true
		}
	}
	return calEntry{}, false
}

// isSorted returns true iff the calibration list is sorted by source then signal.
func isSorted(list []calEntry) bool {
	for i := 1; i < len(list); i++ {
		prev, cur := list[i-1], list[i]
		if prev.Source > cur.Source {
			return false
		}
		if prev.Source == cur.Source && prev.Signal > cur.Signal {
			return false
		}
	}
	return true
}

func run(url string, timeout time.Duration) error {
	ws, _, err := websocket.DefaultDialer.Dial(url, nil)
	if err != nil {
		return fmt.Errorf("dial %s: %w", url, err)
	}
	defer ws.Close()
	c := &conn{ws: ws, timeout: timeout}
	c.startReader()

	// ── Step 1: hub sends a calibration frame on connect, even when empty ────
	// C++ emits "sources" before "calibration" as part of its initial state
	// push; Go emits "calibration" first (Go's sources broadcast is triggered
	// per-add when sources are added, not on client-connect in this test where
	// no sources exist yet).  We explicitly skip "sources" here so the checker
	// is not confused by the ordering difference at connect time.
	fmt.Println("Step 1: expect calibration on connect")
	if _, err := c.nextSkipping("calibration", []string{"sources"}); err != nil {
		return fmt.Errorf("on connect: %w", err)
	}

	// ── Step 2a: set two calibration entries in REVERSE sort order ───────────
	// We deliberately set signal "Zeta" before "Alpha" under source "src1",
	// and set source "src2" before "src1".  The hubs must sort them and the
	// checker asserts the received frame and the saved config file are both
	// in sorted (source asc, signal asc) order.
	wantEntries := []calEntry{
		{Source: "src1", Signal: "Alpha", Scale: 2.0, Offset: 0.5, Unit: "m"},
		{Source: "src1", Signal: "Zeta", Scale: 0.5, Offset: -1.25, Unit: "V"},
		{Source: "src2", Signal: "Beta", Scale: 1.5, Offset: 0.0, Unit: "A"},
	}
	// Submit in reverse-sort order: src2/Beta, then src1/Zeta, then src1/Alpha.
	submitOrder := []calEntry{
		wantEntries[2], // src2/Beta
		wantEntries[1], // src1/Zeta
		wantEntries[0], // src1/Alpha
	}
	fmt.Println("Step 2: set calibration entries in reverse sort order")
	var lastCalFrame frame
	for i, e := range submitOrder {
		if err := c.send(map[string]interface{}{
			"type":   "setCalibration",
			"source": e.Source,
			"signal": e.Signal,
			"scale":  e.Scale,
			"offset": e.Offset,
			"unit":   e.Unit,
		}); err != nil {
			return err
		}
		cf, cerr := c.next("calibration")
		if cerr != nil {
			return fmt.Errorf("after setCalibration[%d]: %w", i, cerr)
		}
		if !isSorted(cf.Cal) {
			return fmt.Errorf("setCalibration[%d]: calibration frame not sorted by source/signal; got %v", i, cf.Cal)
		}
		got, ok := findCal(cf.Cal, e.Source, e.Signal)
		if !ok {
			return fmt.Errorf("setCalibration[%d]: entry %s/%s missing from broadcast", i, e.Source, e.Signal)
		}
		if got != e {
			return fmt.Errorf("setCalibration[%d]: got %+v, want %+v", i, got, e)
		}
		lastCalFrame = cf
	}
	// The last broadcast should contain all three entries in sorted order.
	if len(lastCalFrame.Cal) != len(wantEntries) {
		return fmt.Errorf("after all setCalibration: got %d entries, want %d", len(lastCalFrame.Cal), len(wantEntries))
	}
	// Confirm sorted order in the final broadcast.
	if !isSorted(lastCalFrame.Cal) {
		return fmt.Errorf("final calibration broadcast not sorted; got %v", lastCalFrame.Cal)
	}

	// ── Step 3: invalid entry (scale = 0) must be rejected ───────────────────
	fmt.Println("Step 3: setCalibration with scale=0 must be rejected (no broadcast)")
	if err := c.send(map[string]interface{}{
		"type": "setCalibration", "source": "src1", "signal": "Alpha",
		"scale": 0.0, "offset": 0.0, "unit": "",
	}); err != nil {
		return err
	}
	if _, accepted := c.nextWithin("calibration", 500*time.Millisecond); accepted {
		return fmt.Errorf("setCalibration with scale=0 was accepted, must be rejected")
	}

	// ── Step 4: saveSources → configSaved ────────────────────────────────────
	fmt.Println("Step 4: saveSources → configSaved")
	if err := c.send(map[string]string{"type": "saveSources"}); err != nil {
		return err
	}
	savedF, err := c.next("configSaved")
	if err != nil {
		return err
	}
	if !savedF.OK {
		return fmt.Errorf("configSaved: ok=false, error=%q", savedF.Error)
	}
	if savedF.Path == "" {
		return fmt.Errorf("configSaved: ok=true but path is empty")
	}
	savedPath := savedF.Path

	// ── Step 5: reloadConfig → configReloaded → calibration ─────────────────
	// Documented exception: C++ emits an additional "sources" frame after
	// "calibration" in HandleReloadConfig().  The Go hub does not emit it
	// at that point (it broadcast per-source additions earlier).  We accept
	// the "sources" frame from C++ but log it as a deliberate known difference.
	fmt.Println("Step 5: reloadConfig → configReloaded → calibration")
	if err := c.send(map[string]string{"type": "reloadConfig"}); err != nil {
		return err
	}
	reloadedF, err := c.next("configReloaded")
	if err != nil {
		return err
	}
	if !reloadedF.OK {
		return fmt.Errorf("configReloaded: ok=false, error=%q", reloadedF.Error)
	}
	calAfterReload, err := c.next("calibration")
	if err != nil {
		return fmt.Errorf("after reloadConfig, expected calibration: %w", err)
	}
	if !isSorted(calAfterReload.Cal) {
		return fmt.Errorf("reloadConfig calibration not sorted; got %v", calAfterReload.Cal)
	}
	if len(calAfterReload.Cal) != len(wantEntries) {
		return fmt.Errorf("reloadConfig: got %d calibration entries, want %d", len(calAfterReload.Cal), len(wantEntries))
	}
	for _, want := range wantEntries {
		got, ok := findCal(calAfterReload.Cal, want.Source, want.Signal)
		if !ok {
			return fmt.Errorf("reloadConfig: entry %s/%s did not survive round-trip", want.Source, want.Signal)
		}
		if got != want {
			return fmt.Errorf("reloadConfig: entry %s/%s: got %+v, want %+v", want.Source, want.Signal, got, want)
		}
	}

	// Check for the optional extra "sources" frame from C++ (documented exception).
	// We peek with a short timeout; if it arrives we log the known difference.
	// If another unexpected protocol frame arrives instead, that is still a failure.
	if extraF, arrived := c.nextWithin("sources", 500*time.Millisecond); arrived {
		fmt.Printf("[KNOWN DIFFERENCE] C++ hub emitted extra \"sources\" frame after reload "+
			"(Go hub does not). This is expected — C++ BroadcastSources() in "+
			"HandleReloadConfig() notifies clients of source-list changes after "+
			"LoadSourcesFile(skipActive=true); Go hub propagates per-add via commandCh. "+
			"Extra frame sources count: %d\n", len(extraF.Cal))
	}

	// ── Step 6: verify the saved config file is sorted ───────────────────────
	// We re-read the saved file and parse it to check sort order.
	// This is a file-system check, not a WebSocket check.
	fmt.Printf("Step 6: verify saved config file is sorted: %s\n", savedPath)
	raw, fileErr := os.ReadFile(savedPath)
	if fileErr != nil {
		// The config file may not be accessible from this process (e.g. different
		// temp dir).  Log and skip — the frame sort assertion above already
		// provides coverage.
		fmt.Printf("[note: cannot read config file %s: %v — skipping file sort check]\n", savedPath, fileErr)
	} else {
		var fileEntries []calEntry
		if jsonErr := json.Unmarshal(raw, &fileEntries); jsonErr != nil {
			return fmt.Errorf("config file %s: invalid JSON: %w", savedPath, jsonErr)
		}
		if !isSorted(fileEntries) {
			// Build the expected sorted order for the error message.
			sorted := make([]calEntry, len(fileEntries))
			copy(sorted, fileEntries)
			sort.Slice(sorted, func(i, j int) bool {
				if sorted[i].Source != sorted[j].Source {
					return sorted[i].Source < sorted[j].Source
				}
				return sorted[i].Signal < sorted[j].Signal
			})
			return fmt.Errorf("config file not sorted by source/signal:\n  got:  %v\n  want: %v", fileEntries, sorted)
		}
		fmt.Printf("Config file has %d entries in sorted order.\n", len(fileEntries))
	}

	return nil
}

func main() {
	url := flag.String("url", "ws://127.0.0.1:8090/ws", "hub WebSocket URL")
	timeout := flag.Duration("timeout", 5*time.Second, "per-frame timeout")
	flag.Parse()

	if err := run(*url, *timeout); err != nil {
		fmt.Fprintf(os.Stderr, "configcheck FAIL: %v\n", err)
		os.Exit(1)
	}
	fmt.Println("configcheck OK")
}