Add unit tests and code coverage for the message path

README.md

@@ -1,6 +1,7 @@
 [![rxtx data transmission](https://raw.githubusercontent.com/txn2/rtbeat/master/mast-logo.jpg)](https://github.com/txn2/rtbeat)
 [![Release](https://img.shields.io/github/release/txn2/rtbeat.svg)](https://github.com/txn2/rtbeat/releases)
 [![CI](https://github.com/txn2/rtbeat/actions/workflows/ci.yml/badge.svg)](https://github.com/txn2/rtbeat/actions/workflows/ci.yml)
+[![codecov](https://codecov.io/gh/txn2/rtbeat/graph/badge.svg)](https://codecov.io/gh/txn2/rtbeat)
 [![Go Report Card](https://goreportcard.com/badge/github.com/txn2/rtbeat)](https://goreportcard.com/report/github.com/txn2/rtbeat)
 [![OpenSSF Scorecard](https://api.securityscorecards.dev/projects/github.com/txn2/rtbeat/badge)](https://securityscorecards.dev/viewer/?uri=github.com/txn2/rtbeat)
 [![Docker Pulls](https://img.shields.io/docker/pulls/txn2/rtbeat.svg)](https://hub.docker.com/r/txn2/rtbeat/)

beater/rtbeat.go

@@ -106,54 +106,9 @@ func (bt *Rtbeat) Run(b *beat.Beat) error {
 	// get a router
 	r := gin.Default()
 
-	r.POST("/in", func(c *gin.Context) {
-
-		// increment batches
-		batches.Inc()
-		msg := &rtq.MessageBatch{}
-
-		rawData, _ := c.GetRawData()
-
-		err := json.Unmarshal(rawData, &msg)
-		if err != nil {
-			c.JSON(500, gin.H{
-				"status":  "FAIL",
-				"message": fmt.Sprintf("could not unmarshal json: %s", rawData),
-			})
-
-			bt.logger.Error("Run", zap.String("error", "could not unmarshal json"), zap.Error(err))
-			return
-		}
-
-		// respond quickly to avoid getting a re-send from the server
-		c.JSON(200, gin.H{
-			"status": "OK",
-		})
-
-		// fie this in the background
-		go func() {
-			events := make([]beat.Event, 1)
-			var event beat.Event
-
-			for i, message := range msg.Messages {
-				messages.Inc()
-				event = beat.Event{
-					Timestamp: time.Now(),
-					Fields: common.MapStr{
-						"type":     b.Info.Name,
-						"rxtxMsg":  message,
-						"clientIp": c.ClientIP(),
-					},
-					Private: i,
-				}
-				events = append(events, event)
-			}
-
-			// TODO check for published state
-			bt.client.PublishAll(events)
-		}()
-
-	})
+	r.POST("/in", inHandler(b.Info.Name, bt.logger, bt.client, batches.Inc, func(n int) {
+		messages.Add(float64(n))
+	}))
 
 	// Prometheus Metrics
 	r.GET("/metrics", gin.WrapH(promhttp.Handler()))
@@ -200,3 +155,62 @@ func (bt *Rtbeat) Stop() {
 	_ = bt.client.Close()
 	close(bt.done)
 }
+
+// eventPublisher is the subset of beat.Client the /in handler needs. It lets
+// tests substitute a fake for the libbeat publisher pipeline.
+type eventPublisher interface {
+	PublishAll([]beat.Event)
+}
+
+// buildEvents converts an rxtx MessageBatch into the beat.Event slice rtbeat
+// publishes. Each message becomes one event carrying the original message under
+// "rxtxMsg" alongside "type" and "clientIp". The slice is pre-sized with one
+// leading zero-value event; this is long-standing behavior, preserved here
+// intentionally (changing it is tracked separately).
+func buildEvents(beatName, clientIP string, msg *rtq.MessageBatch) []beat.Event {
+	events := make([]beat.Event, 1)
+	for i, message := range msg.Messages {
+		events = append(events, beat.Event{
+			Timestamp: time.Now(),
+			Fields: common.MapStr{
+				"type":     beatName,
+				"rxtxMsg":  message,
+				"clientIp": clientIP,
+			},
+			Private: i,
+		})
+	}
+	return events
+}
+
+// inHandler builds the POST /in gin handler. Metric updates are injected as
+// callbacks (onBatch, onMessages) so the handler can be exercised in tests
+// without registering against the global prometheus registry. The handler
+// responds before publishing so a slow output never blocks the rxtx client.
+func inHandler(beatName string, logger *zap.Logger, pub eventPublisher, onBatch func(), onMessages func(n int)) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		onBatch()
+
+		msg := &rtq.MessageBatch{}
+		rawData, _ := c.GetRawData()
+
+		if err := json.Unmarshal(rawData, &msg); err != nil {
+			c.JSON(http.StatusInternalServerError, gin.H{
+				"status":  "FAIL",
+				"message": fmt.Sprintf("could not unmarshal json: %s", rawData),
+			})
+			logger.Error("Run", zap.String("error", "could not unmarshal json"), zap.Error(err))
+			return
+		}
+
+		// respond quickly to avoid getting a re-send from the server
+		c.JSON(http.StatusOK, gin.H{"status": "OK"})
+
+		onMessages(len(msg.Messages))
+		// Capture the client IP synchronously: the gin context is recycled
+		// once the handler returns, so it must not be read from the goroutine.
+		events := buildEvents(beatName, c.ClientIP(), msg)
+
+		go pub.PublishAll(events)
+	}
+}

beater/rtbeat_test.go

@@ -0,0 +1,148 @@
+package beater
+
+import (
+	"bytes"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/elastic/beats/v7/libbeat/beat"
+	"github.com/gin-gonic/gin"
+	"github.com/txn2/rxtx/rtq"
+	"go.uber.org/zap"
+)
+
+func init() { gin.SetMode(gin.TestMode) }
+
+// fakePublisher records the events handed to PublishAll, satisfying the
+// eventPublisher interface in place of a real libbeat client.
+type fakePublisher struct {
+	got chan []beat.Event
+}
+
+func newFakePublisher() *fakePublisher {
+	return &fakePublisher{got: make(chan []beat.Event, 1)}
+}
+
+func (f *fakePublisher) PublishAll(events []beat.Event) { f.got <- events }
+
+func TestBuildEvents(t *testing.T) {
+	msg := &rtq.MessageBatch{
+		Uuid: "batch-1",
+		Size: 2,
+		Messages: []rtq.Message{
+			{Seq: "1", Producer: "p", Payload: map[string]interface{}{"a": 1.0}},
+			{Seq: "2", Producer: "p", Payload: map[string]interface{}{"b": 2.0}},
+		},
+	}
+
+	events := buildEvents("rtbeat", "10.0.0.1", msg)
+
+	// One leading zero-value placeholder event (preserved legacy behavior),
+	// then one event per message.
+	if got, want := len(events), 1+len(msg.Messages); got != want {
+		t.Fatalf("len(events) = %d, want %d", got, want)
+	}
+	if events[0].Fields != nil {
+		t.Errorf("events[0] should be the zero-value placeholder, got Fields=%v", events[0].Fields)
+	}
+
+	for i, in := range msg.Messages {
+		ev := events[i+1]
+		if ev.Fields["type"] != "rtbeat" {
+			t.Errorf("event %d type = %v, want rtbeat", i, ev.Fields["type"])
+		}
+		if ev.Fields["clientIp"] != "10.0.0.1" {
+			t.Errorf("event %d clientIp = %v, want 10.0.0.1", i, ev.Fields["clientIp"])
+		}
+		got, ok := ev.Fields["rxtxMsg"].(rtq.Message)
+		if !ok {
+			t.Fatalf("event %d rxtxMsg is %T, want rtq.Message", i, ev.Fields["rxtxMsg"])
+		}
+		if got.Seq != in.Seq {
+			t.Errorf("event %d rxtxMsg.Seq = %q, want %q", i, got.Seq, in.Seq)
+		}
+		if ev.Private != i {
+			t.Errorf("event %d Private = %v, want %d", i, ev.Private, i)
+		}
+		if ev.Timestamp.IsZero() {
+			t.Errorf("event %d Timestamp is zero", i)
+		}
+	}
+}
+
+func TestBuildEventsEmptyBatch(t *testing.T) {
+	events := buildEvents("rtbeat", "1.2.3.4", &rtq.MessageBatch{})
+	if len(events) != 1 {
+		t.Fatalf("empty batch: len(events) = %d, want 1 (placeholder only)", len(events))
+	}
+}
+
+func newTestRouter(pub eventPublisher, onBatch func(), onMessages func(int)) *gin.Engine {
+	r := gin.New()
+	r.POST("/in", inHandler("rtbeat", zap.NewNop(), pub, onBatch, onMessages))
+	return r
+}
+
+func TestInHandlerValidBatch(t *testing.T) {
+	pub := newFakePublisher()
+	var batches, messages int
+	r := newTestRouter(pub, func() { batches++ }, func(n int) { messages += n })
+
+	body := `{"uuid":"b1","size":1,"messages":[{"seq":"1","payload":{"hello":"world"}}]}`
+	w := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodPost, "/in", bytes.NewBufferString(body))
+	req.RemoteAddr = "203.0.113.7:54321"
+	r.ServeHTTP(w, req)
+
+	if w.Code != http.StatusOK {
+		t.Fatalf("status = %d, want 200; body=%s", w.Code, w.Body.String())
+	}
+	if batches != 1 {
+		t.Errorf("onBatch called %d times, want 1", batches)
+	}
+	if messages != 1 {
+		t.Errorf("onMessages total = %d, want 1", messages)
+	}
+
+	select {
+	case events := <-pub.got:
+		if len(events) != 2 { // placeholder + one message
+			t.Fatalf("published %d events, want 2", len(events))
+		}
+		// clientIp is read synchronously from the request; assert it
+		// propagates through the handler into the published event.
+		if got := events[1].Fields["clientIp"]; got != "203.0.113.7" {
+			t.Errorf("published clientIp = %v, want 203.0.113.7", got)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("PublishAll was not called")
+	}
+}
+
+func TestInHandlerBadJSON(t *testing.T) {
+	pub := newFakePublisher()
+	var batches, messages int
+	r := newTestRouter(pub, func() { batches++ }, func(n int) { messages += n })
+
+	w := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodPost, "/in", bytes.NewBufferString(`{not valid json`))
+	r.ServeHTTP(w, req)
+
+	if w.Code != http.StatusInternalServerError {
+		t.Fatalf("status = %d, want 500; body=%s", w.Code, w.Body.String())
+	}
+	if batches != 1 {
+		t.Errorf("onBatch called %d times, want 1", batches)
+	}
+	if messages != 0 {
+		t.Errorf("onMessages total = %d, want 0 on parse failure", messages)
+	}
+
+	// On the error path the handler returns before spawning the publish
+	// goroutine, so nothing can ever be sent — assert deterministically.
+	if len(pub.got) != 0 {
+		t.Fatal("PublishAll must not be called on bad JSON")
+	}
+}

config/config_test.go

@@ -1,3 +1,54 @@
-// +build !integration
-
 package config
+
+import (
+	"testing"
+
+	"github.com/elastic/beats/v7/libbeat/common"
+)
+
+func TestDefaultConfig(t *testing.T) {
+	if DefaultConfig.Port != "8081" {
+		t.Errorf("default Port = %q, want %q", DefaultConfig.Port, "8081")
+	}
+	if DefaultConfig.Timeout != 5 {
+		t.Errorf("default Timeout = %d, want 5", DefaultConfig.Timeout)
+	}
+}
+
+func TestUnpackOverrides(t *testing.T) {
+	raw, err := common.NewConfigFrom(map[string]interface{}{
+		"port":    "9090",
+		"timeout": 30,
+	})
+	if err != nil {
+		t.Fatalf("NewConfigFrom: %v", err)
+	}
+
+	c := DefaultConfig
+	if err := raw.Unpack(&c); err != nil {
+		t.Fatalf("Unpack: %v", err)
+	}
+
+	if c.Port != "9090" {
+		t.Errorf("Port = %q, want %q", c.Port, "9090")
+	}
+	if c.Timeout != 30 {
+		t.Errorf("Timeout = %d, want 30", c.Timeout)
+	}
+}
+
+func TestUnpackEmptyKeepsDefaults(t *testing.T) {
+	raw, err := common.NewConfigFrom(map[string]interface{}{})
+	if err != nil {
+		t.Fatalf("NewConfigFrom: %v", err)
+	}
+
+	c := DefaultConfig
+	if err := raw.Unpack(&c); err != nil {
+		t.Fatalf("Unpack: %v", err)
+	}
+
+	if c.Port != "8081" || c.Timeout != 5 {
+		t.Errorf("defaults not preserved after empty unpack: %+v", c)
+	}
+}