test(rest): add black-box WebSocket monitor tests (#621)

Cover the REST WebSocket monitor endpoints (v1 + v2), previously at 0%
for all 18 functions in socketv1_controller.go and socketv2_controller.go;
pkg/rest coverage goes from 20.9% to 73.2%.

The harness dials a real gorilla/websocket client into an httptest server
wired to a running msghub.Hub, and relies on the hub's history replay for
deterministic ordering: dispatching before dialing seeds the replay, and
events are relayed in dispatch order, so receiving event N next proves
filtered events were skipped rather than delayed.

Covered for both API versions:

- monitor-all and per-mailbox endpoints, including history replay and live
  delivery, and mailbox filtering of both message-stored and (v2)
  message-deleted events
- address-form mailbox names via Manager.MailboxForAddress (user%40example.com)
- invalid mailbox names rejected with 500 before the upgrade, and plain GET
  requests failing the handshake with 400
- client data messages discarded by the read pump
- clean client shutdown: listener deregistration, empty close frame from
  the writer racing the read pump's close-code echo, and the
  WebSocketConnectsCurrent expvar gauge increment/decrement

Tests only; no production changes.  Notes for follow-up commits:

- v1 monitors ignore deletions while v2 reports them (pinned as current
  behavior in TestMonitorMailboxMessagesV1)
- httptest.Server.Close does not wait for hijacked connections, so
  dialMonitor's cleanup awaits the handler goroutine via the connections
  gauge before the hub is stopped; the read/write pumps calling
  msgListener.Close concurrently can still double-close its channel
  (check-then-act race in Close's select) and deserves a fix
- deadline-set error paths, the 54s ping/pong cycle, and write-failure
  paths remain uncovered, consistent with the POP3 send-error paths

pkg/rest/apiv1_controller_test.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/inbucket/inbucket/v3/pkg/extension/event"
 	"github.com/inbucket/inbucket/v3/pkg/message"
+	"github.com/inbucket/inbucket/v3/pkg/msghub"
 	"github.com/inbucket/inbucket/v3/pkg/test"
 	"github.com/jhillyerd/enmime/v2"
 )
@@ -18,7 +19,7 @@ import (
 func TestRestMailboxList(t *testing.T) {
 	// Setup
 	mm := test.NewManager()
-	logbuf := setupWebServer(mm)
+	logbuf := setupWebServer(mm, &msghub.Hub{})
 
 	// Test invalid mailbox name
 	w, err := testRestGet("http://localhost/api/v1/mailbox/foo%20bar")
@@ -122,7 +123,7 @@ func TestRestMailboxList(t *testing.T) {
 func TestRestMessage(t *testing.T) {
 	// Setup
 	mm := test.NewManager()
-	logbuf := setupWebServer(mm)
+	logbuf := setupWebServer(mm, &msghub.Hub{})
 
 	// Test invalid mailbox name
 	w, err := testRestGet("http://localhost/api/v1/mailbox/foo%20bar/0001")
@@ -244,7 +245,7 @@ func TestRestMessage(t *testing.T) {
 
 func TestRestMarkSeen(t *testing.T) {
 	mm := test.NewManager()
-	logbuf := setupWebServer(mm)
+	logbuf := setupWebServer(mm, &msghub.Hub{})
 	// Create some messages.
 	tzPDT := time.FixedZone("PDT", -7*3600)
 	tzPST := time.FixedZone("PST", -8*3600)

pkg/rest/socket_controller_test.go

@@ -0,0 +1,339 @@
+package rest
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net/http/httptest"
+	"net/mail"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/gorilla/websocket"
+	"github.com/inbucket/inbucket/v3/pkg/extension"
+	"github.com/inbucket/inbucket/v3/pkg/extension/event"
+	"github.com/inbucket/inbucket/v3/pkg/msghub"
+	"github.com/inbucket/inbucket/v3/pkg/server/web"
+	"github.com/inbucket/inbucket/v3/pkg/test"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// monitorTestTimeout bounds WebSocket reads so a broken server fails the test instead of hanging.
+const monitorTestTimeout = 5 * time.Second
+
+// monitorDialer dials test servers directly; the default dialer would honor proxy env vars.
+var monitorDialer = &websocket.Dialer{}
+
+// startMsgHub returns a running *msghub.Hub with room for every message a test dispatches
+// (replayed to listeners that connect later).  The actor goroutine is stopped via t.Cleanup.
+func startMsgHub(t *testing.T) *msghub.Hub {
+	t.Helper()
+	hub := msghub.New(10, extension.NewHost())
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	go hub.Start(ctx)
+	return hub
+}
+
+// setupMonitorServer wires a real (running) msg hub into the web server globals and serves the
+// global router over HTTP; upgrading WebSocket endpoints needs a real connection to hijack.
+// The returned gauge baseline is the value of web.ExpWebSocketConnectsCurrent before any
+// connection is made; pass it to dialMonitor so sessions can be awaited on teardown.
+func setupMonitorServer(t *testing.T, hub *msghub.Hub) (*httptest.Server, int64) {
+	t.Helper()
+	mm := test.NewManager()
+	setupWebServer(mm, hub)
+	server := httptest.NewServer(web.Router)
+	// Registered before any connection cleanup so the listener socket is closed after the
+	// sessions below have been awaited.
+	t.Cleanup(server.Close)
+	return server, web.ExpWebSocketConnectsCurrent.Value()
+}
+
+// dialMonitor connects a WebSocket client to path on server.  The cleanup closes the connection
+// and waits for the server-side handler goroutine to exit, which is required before the msg hub
+// may be stopped: httptest.Server.Close does not wait for hijacked (WebSocket) connections, and
+// a handler deregistering its hub listener after the hub was stopped races with it.
+func dialMonitor(t *testing.T, server *httptest.Server, path string, gaugeBase int64) *websocket.Conn {
+	t.Helper()
+	url := "ws" + strings.TrimPrefix(server.URL, "http") + path
+	conn, resp, err := monitorDialer.Dial(url, nil)
+	require.NoError(t, err)
+	// resp.Body is an empty no-op closer once the connection is upgraded.
+	_ = resp.Body.Close()
+	t.Cleanup(func() {
+		_ = conn.Close()
+		require.Eventually(t, func() bool {
+			return web.ExpWebSocketConnectsCurrent.Value() == gaugeBase
+		}, monitorTestTimeout, time.Millisecond, "server handler did not shut down")
+	})
+	return conn
+}
+
+// readMonitorJSON reads the next data message from conn and decodes it into a generic map.
+func readMonitorJSON(t *testing.T, conn *websocket.Conn) map[string]interface{} {
+	t.Helper()
+	require.NoError(t, conn.SetReadDeadline(time.Now().Add(monitorTestTimeout)))
+	_, data, err := conn.ReadMessage()
+	require.NoError(t, err)
+	var decoded interface{}
+	require.NoError(t, json.Unmarshal(data, &decoded), "payload: %s", data)
+	obj, ok := decoded.(map[string]interface{})
+	require.True(t, ok, "expected JSON object, got: %s", data)
+	return obj
+}
+
+// testMetadata builds a message for dispatch into the hub.
+func testMetadata(mailbox, id string) event.MessageMetadata {
+	return event.MessageMetadata{
+		Mailbox: mailbox,
+		ID:      id,
+		From:    &mail.Address{Address: "from-" + id + "@host"},
+		To:      []*mail.Address{{Address: "to-" + id + "@host"}},
+		Subject: "subject " + id,
+		Date:    time.Date(2012, 2, 1, 10, 11, 12, 0, time.UTC),
+		Size:    1234,
+	}
+}
+
+// assertMonitorHeader checks the JSONMessageHeaderV1 fields of a decoded monitor payload.
+func assertMonitorHeader(t *testing.T, decoded interface{}, want event.MessageMetadata) {
+	t.Helper()
+	decodedStringEquals(t, decoded, "mailbox", want.Mailbox)
+	decodedStringEquals(t, decoded, "id", want.ID)
+	decodedStringEquals(t, decoded, "from", "<"+want.From.Address+">")
+	decodedStringEquals(t, decoded, "to/[0]", "<"+want.To[0].Address+">")
+	decodedStringEquals(t, decoded, "subject", want.Subject)
+	decodedStringEquals(t, decoded, "date", "2012-02-01T10:11:12Z")
+	decodedNumberEquals(t, decoded, "posix-millis", float64(want.Date.UnixNano()/1000000))
+	decodedNumberEquals(t, decoded, "size", float64(want.Size))
+}
+
+// monitorEventHeader returns the nested header object of a decoded v2 monitor event.
+func monitorEventHeader(t *testing.T, ev map[string]interface{}) map[string]interface{} {
+	t.Helper()
+	header, ok := ev["header"].(map[string]interface{})
+	require.True(t, ok, "event has no header object: %v", ev)
+	return header
+}
+
+func TestMonitorAllMessagesV1(t *testing.T) {
+	hub := startMsgHub(t)
+	server, gaugeBase := setupMonitorServer(t, hub)
+
+	// Messages dispatched before connecting are replayed from hub history to new listeners.
+	replayed := testMetadata("replay", "0001")
+	hub.Dispatch(replayed)
+
+	conn := dialMonitor(t, server, "/api/v1/monitor/messages", gaugeBase)
+	assertMonitorHeader(t, readMonitorJSON(t, conn), replayed)
+	assert.Equal(t, gaugeBase+1, web.ExpWebSocketConnectsCurrent.Value(),
+		"open WebSocket connections gauge")
+
+	// Messages dispatched after registration are delivered live.  Receiving the replayed
+	// message above proves the listener was registered before this dispatch.
+	live := testMetadata("live", "0002")
+	hub.Dispatch(live)
+	assertMonitorHeader(t, readMonitorJSON(t, conn), live)
+
+	// A clean client close tears the session down server-side: the read pump sees the close
+	// frame, the listener deregisters from the hub, and the connections gauge drops back.
+	require.NoError(t, conn.WriteControl(websocket.CloseMessage,
+		websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""),
+		time.Now().Add(monitorTestTimeout)))
+	require.NoError(t, conn.Close())
+	require.Eventually(t, func() bool {
+		return web.ExpWebSocketConnectsCurrent.Value() == gaugeBase
+	}, monitorTestTimeout, time.Millisecond, "open WebSocket connections gauge")
+}
+
+func TestMonitorMailboxMessagesV1(t *testing.T) {
+	hub := startMsgHub(t)
+	server, gaugeBase := setupMonitorServer(t, hub)
+
+	hub.Dispatch(testMetadata("other", "0001"))
+	watched := testMetadata("watched", "0002")
+	hub.Dispatch(watched)
+	hub.Dispatch(testMetadata("other", "0003"))
+
+	conn := dialMonitor(t, server, "/api/v1/monitor/messages/watched", gaugeBase)
+	assertMonitorHeader(t, readMonitorJSON(t, conn), watched)
+
+	// Events are relayed in dispatch order, so receiving 0005 next proves both the replayed
+	// and live deliveries for other mailboxes were filtered, not merely delayed.
+	hub.Dispatch(testMetadata("other", "0004"))
+	live := testMetadata("watched", "0005")
+	hub.Dispatch(live)
+	assertMonitorHeader(t, readMonitorJSON(t, conn), live)
+
+	// Documents current behavior: v1 monitors ignore message deletions and emit no event;
+	// v2 emits message-deleted.  The subsequent dispatch arriving next proves the delete
+	// produced nothing.  Flagged for a follow-up fix.
+	hub.Delete("watched", "0005")
+	after := testMetadata("watched", "0006")
+	hub.Dispatch(after)
+	assertMonitorHeader(t, readMonitorJSON(t, conn), after)
+}
+
+func TestMonitorMailboxAddressV1(t *testing.T) {
+	hub := startMsgHub(t)
+	server, gaugeBase := setupMonitorServer(t, hub)
+
+	// The {name} path segment goes through Manager.MailboxForAddress; with the test stub's
+	// full-address naming policy, user@example.com maps to a mailbox of the same name.
+	stored := testMetadata("user@example.com", "0001")
+	hub.Dispatch(stored)
+
+	conn := dialMonitor(t, server, "/api/v1/monitor/messages/user%40example.com", gaugeBase)
+	assertMonitorHeader(t, readMonitorJSON(t, conn), stored)
+}
+
+func TestMonitorMailboxInvalidNameV1(t *testing.T) {
+	hub := startMsgHub(t)
+	setupWebServer(test.NewManager(), hub)
+
+	// Invalid mailbox names are rejected before the connection is upgraded.
+	w, err := testRestGet("http://localhost/api/v1/monitor/messages/foo%20bar")
+	require.NoError(t, err)
+	assert.Equal(t, 500, w.Code)
+}
+
+func TestMonitorUpgradeFailedV1(t *testing.T) {
+	hub := startMsgHub(t)
+	setupWebServer(test.NewManager(), hub)
+
+	// A plain GET without WebSocket upgrade headers fails the handshake; the upgrader writes
+	// the 400 itself before the handler error is reported.
+	w, err := testRestGet("http://localhost/api/v1/monitor/messages")
+	require.NoError(t, err)
+	assert.Equal(t, 400, w.Code)
+}
+
+func TestMonitorClientCloseV1(t *testing.T) {
+	hub := startMsgHub(t)
+	server, gaugeBase := setupMonitorServer(t, hub)
+
+	stored := testMetadata("close", "0001")
+	hub.Dispatch(stored)
+
+	conn := dialMonitor(t, server, "/api/v1/monitor/messages", gaugeBase)
+
+	// Data messages sent by the client are discarded; the monitor keeps relaying events.
+	require.NoError(t, conn.WriteMessage(websocket.TextMessage, []byte(`{"junk":true}`)))
+	assertMonitorHeader(t, readMonitorJSON(t, conn), stored)
+
+	// A clean client close ends the session.  The peer may observe any of: the close-code
+	// echo sent by the server read pump's default close handler, the writer's empty close
+	// frame, or a dropped TCP connection from the handler teardown; all indicate shutdown.
+	require.NoError(t, conn.WriteControl(websocket.CloseMessage,
+		websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""),
+		time.Now().Add(monitorTestTimeout)))
+	require.NoError(t, conn.SetReadDeadline(time.Now().Add(monitorTestTimeout)))
+	_, _, err := conn.ReadMessage()
+	require.Error(t, err)
+	var closeErr *websocket.CloseError
+	if errors.As(err, &closeErr) {
+		assert.Contains(t, []int{
+			websocket.CloseNormalClosure,    // echoed by the server read pump
+			websocket.CloseNoStatusReceived, // writer's empty close frame
+		}, closeErr.Code)
+	}
+}
+
+func TestMonitorAllMessagesV2(t *testing.T) {
+	hub := startMsgHub(t)
+	server, gaugeBase := setupMonitorServer(t, hub)
+
+	// Stored messages dispatched before connecting are replayed from hub history.
+	stored := testMetadata("replay", "0001")
+	hub.Dispatch(stored)
+
+	conn := dialMonitor(t, server, "/api/v2/monitor/messages", gaugeBase)
+	ev := readMonitorJSON(t, conn)
+	decodedStringEquals(t, ev, "variant", "message-stored")
+	assertMonitorHeader(t, monitorEventHeader(t, ev), stored)
+
+	// Data messages sent by the client are discarded; the monitor keeps relaying events.
+	require.NoError(t, conn.WriteMessage(websocket.TextMessage, []byte(`{"junk":true}`)))
+
+	live := testMetadata("live", "0002")
+	hub.Dispatch(live)
+	ev = readMonitorJSON(t, conn)
+	decodedStringEquals(t, ev, "variant", "message-stored")
+	assertMonitorHeader(t, monitorEventHeader(t, ev), live)
+
+	// Deletions are reported as message-deleted events with the message identifier.
+	hub.Delete("live", "0002")
+	ev = readMonitorJSON(t, conn)
+	decodedStringEquals(t, ev, "variant", "message-deleted")
+	decodedStringEquals(t, ev, "identifier/mailbox", "live")
+	decodedStringEquals(t, ev, "identifier/id", "0002")
+}
+
+func TestMonitorMailboxMessagesV2(t *testing.T) {
+	hub := startMsgHub(t)
+	server, gaugeBase := setupMonitorServer(t, hub)
+
+	hub.Dispatch(testMetadata("other", "0001"))
+	watched := testMetadata("watched", "0002")
+	hub.Dispatch(watched)
+
+	conn := dialMonitor(t, server, "/api/v2/monitor/messages/watched", gaugeBase)
+	ev := readMonitorJSON(t, conn)
+	decodedStringEquals(t, ev, "variant", "message-stored")
+	assertMonitorHeader(t, monitorEventHeader(t, ev), watched)
+
+	// Events are relayed in dispatch order, so receiving message-stored for 0003 next proves
+	// the other-mailbox delete was filtered, not merely delayed.
+	hub.Delete("other", "0001")
+	live := testMetadata("watched", "0003")
+	hub.Dispatch(live)
+	ev = readMonitorJSON(t, conn)
+	decodedStringEquals(t, ev, "variant", "message-stored")
+	assertMonitorHeader(t, monitorEventHeader(t, ev), live)
+
+	hub.Delete("other", "9999")
+	hub.Delete("watched", "0003")
+	ev = readMonitorJSON(t, conn)
+	decodedStringEquals(t, ev, "variant", "message-deleted")
+	decodedStringEquals(t, ev, "identifier/mailbox", "watched")
+	decodedStringEquals(t, ev, "identifier/id", "0003")
+}
+
+func TestMonitorMailboxAddressV2(t *testing.T) {
+	hub := startMsgHub(t)
+	server, gaugeBase := setupMonitorServer(t, hub)
+
+	// The {name} path segment goes through Manager.MailboxForAddress; with the test stub's
+	// full-address naming policy, user@example.com maps to a mailbox of the same name.
+	stored := testMetadata("user@example.com", "0001")
+	hub.Dispatch(stored)
+
+	conn := dialMonitor(t, server, "/api/v2/monitor/messages/user%40example.com", gaugeBase)
+	ev := readMonitorJSON(t, conn)
+	decodedStringEquals(t, ev, "variant", "message-stored")
+	assertMonitorHeader(t, monitorEventHeader(t, ev), stored)
+}
+
+func TestMonitorMailboxInvalidNameV2(t *testing.T) {
+	hub := startMsgHub(t)
+	setupWebServer(test.NewManager(), hub)
+
+	// Invalid mailbox names are rejected before the connection is upgraded.
+	w, err := testRestGet("http://localhost/api/v2/monitor/messages/foo%20bar")
+	require.NoError(t, err)
+	assert.Equal(t, 500, w.Code)
+}
+
+func TestMonitorUpgradeFailedV2(t *testing.T) {
+	hub := startMsgHub(t)
+	setupWebServer(test.NewManager(), hub)
+
+	// A plain GET without WebSocket upgrade headers fails the handshake; the upgrader writes
+	// the 400 itself before the handler error is reported.
+	w, err := testRestGet("http://localhost/api/v2/monitor/messages")
+	require.NoError(t, err)
+	assert.Equal(t, 400, w.Code)
+}

pkg/rest/testutils_test.go

@@ -51,7 +51,7 @@ func testRestPatch(url string, body string) (*httptest.ResponseRecorder, error)
 	return w, nil
 }
 
-func setupWebServer(mm message.Manager) *bytes.Buffer {
+func setupWebServer(mm message.Manager, hub *msghub.Hub) *bytes.Buffer {
 	// Capture log output
 	buf := new(bytes.Buffer)
 	log.SetOutput(buf)
@@ -63,7 +63,7 @@ func setupWebServer(mm message.Manager) *bytes.Buffer {
 		},
 	}
 	SetupRoutes(web.Router.PathPrefix("/api/").Subrouter())
-	web.NewServer(cfg, mm, &msghub.Hub{})
+	web.NewServer(cfg, mm, hub)
 
 	return buf
 }