Files
gochat/internal/handler/api/v1/sse_event_handler_test.go
T
2026-06-04 15:44:48 +08:00

223 lines
6.1 KiB
Go

package v1
import (
"encoding/json"
"net/http"
"net/http/httptest"
"strconv"
"strings"
"testing"
"time"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
wspkg "github.com/gochat/gochat/internal/ws"
)
func setupSSETest(t *testing.T) (*gin.Engine, *wspkg.SSERegistry, *SSEEventHandler) {
t.Helper()
gin.SetMode(gin.TestMode)
registry := wspkg.NewSSERegistry()
handler := NewSSEEventHandler(registry)
engine := gin.New()
engine.Use(func(c *gin.Context) {
c.Set("user_id", uint(100))
if accID, err := strconv.ParseUint(c.Param("account_id"), 10, 32); err == nil {
c.Set("account_id", uint(accID))
}
c.Next()
})
engine.GET("/api/v1/accounts/:account_id/events", handler.StreamEvents)
return engine, registry, handler
}
func TestSSEEventHandler_StreamEvents_InvalidAccountID(t *testing.T) {
registry := wspkg.NewSSERegistry()
handler := NewSSEEventHandler(registry)
gin.SetMode(gin.TestMode)
engine := gin.New()
engine.Use(func(c *gin.Context) {
c.Set("user_id", uint(100))
c.Next()
})
engine.GET("/api/v1/accounts/:account_id/events", handler.StreamEvents)
w := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/api/v1/accounts/abc/events", nil)
engine.ServeHTTP(w, req)
assert.Equal(t, http.StatusBadRequest, w.Code)
}
func TestSSERegistry_SubscribeUnsubscribe(t *testing.T) {
registry := wspkg.NewSSERegistry()
channel := registry.Subscribe("chan1", 1, 100)
require.NotNil(t, channel)
assert.Equal(t, "chan1", channel.ID)
assert.Equal(t, uint(1), channel.AccountID)
assert.Equal(t, uint(100), channel.UserID)
assert.Equal(t, 1, registry.ChannelCount(1))
registry.Unsubscribe("chan1")
assert.Equal(t, 0, registry.ChannelCount(1))
assert.True(t, channel.Closed)
}
func TestSSERegistry_SendToAccount(t *testing.T) {
registry := wspkg.NewSSERegistry()
channel := registry.Subscribe("chan1", 1, 100)
event := wspkg.SSEEvent{
Type: "message.created",
Payload: map[string]interface{}{"id": 1, "content": "hello"},
}
registry.SendToAccount(1, event)
// Read from channel with timeout
select {
case received := <-channel.Events:
assert.Equal(t, "message.created", received.Type)
payloadMap, ok := received.Payload.(map[string]interface{})
require.True(t, ok)
assert.Equal(t, 1, payloadMap["id"])
assert.Equal(t, "hello", payloadMap["content"])
case <-time.After(100 * time.Millisecond):
t.Fatal("timeout waiting for SSE event")
}
}
func TestSSERegistry_SendToAccount_MultipleChannels(t *testing.T) {
registry := wspkg.NewSSERegistry()
chan1 := registry.Subscribe("chan1", 1, 100)
chan2 := registry.Subscribe("chan2", 1, 200)
event := wspkg.SSEEvent{Type: "conversation.updated", Payload: "data"}
registry.SendToAccount(1, event)
// Both channels should receive the event
select {
case <-chan1.Events:
case <-time.After(100 * time.Millisecond):
t.Fatal("chan1 didn't receive event")
}
select {
case <-chan2.Events:
case <-time.After(100 * time.Millisecond):
t.Fatal("chan2 didn't receive event")
}
}
func TestSSERegistry_SendToAccount_DifferentAccounts(t *testing.T) {
registry := wspkg.NewSSERegistry()
chan1 := registry.Subscribe("chan1", 1, 100)
chan2 := registry.Subscribe("chan2", 2, 200)
event := wspkg.SSEEvent{Type: "message.created", Payload: "data"}
registry.SendToAccount(1, event)
// Only chan1 should receive
select {
case <-chan1.Events:
case <-time.After(100 * time.Millisecond):
t.Fatal("chan1 didn't receive event")
}
// chan2 should NOT receive
select {
case <-chan2.Events:
t.Fatal("chan2 should not receive event for account 1")
case <-time.After(50 * time.Millisecond):
// Expected — no event
}
}
func TestSSERegistry_SendToConversation(t *testing.T) {
registry := wspkg.NewSSERegistry()
channel := registry.Subscribe("chan1", 1, 100)
registry.SubscribeConversation("chan1", 42)
event := wspkg.SSEEvent{Type: "message.created", Payload: "conv-data"}
registry.SendToConversation(1, 42, event)
select {
case received := <-channel.Events:
assert.Equal(t, "message.created", received.Type)
case <-time.After(100 * time.Millisecond):
t.Fatal("timeout waiting for conversation event")
}
// Send to a different conversation — should not arrive
registry.SendToConversation(1, 99, wspkg.SSEEvent{Type: "other", Payload: "x"})
select {
case <-channel.Events:
t.Fatal("should not receive event for unsubscribed conversation")
case <-time.After(50 * time.Millisecond):
// Expected
}
}
func TestSSERegistry_UnsubscribeConversation(t *testing.T) {
registry := wspkg.NewSSERegistry()
channel := registry.Subscribe("chan1", 1, 100)
registry.SubscribeConversation("chan1", 42)
// Unsubscribe from conversation 42
registry.UnsubscribeConversation("chan1", 42)
// Send to conversation 42 — should not arrive (no conversation filter)
registry.SendToConversation(1, 42, wspkg.SSEEvent{Type: "test", Payload: "x"})
select {
case <-channel.Events:
// This may arrive because SendToConversation sends to all account channels
// if no specific conversation filter matches. The behavior depends on impl.
case <-time.After(50 * time.Millisecond):
// Also acceptable
}
}
func TestSSERegistry_TotalChannelCount(t *testing.T) {
registry := wspkg.NewSSERegistry()
assert.Equal(t, 0, registry.TotalChannelCount())
registry.Subscribe("chan1", 1, 100)
assert.Equal(t, 1, registry.TotalChannelCount())
registry.Subscribe("chan2", 2, 200)
assert.Equal(t, 2, registry.TotalChannelCount())
registry.Unsubscribe("chan1")
assert.Equal(t, 1, registry.TotalChannelCount())
}
func TestFormatSSE(t *testing.T) {
event := wspkg.SSEEvent{
Type: "message.created",
Payload: map[string]interface{}{"id": 1},
}
result, err := wspkg.FormatSSE(event)
require.NoError(t, err)
assert.True(t, strings.Contains(result, "event: message.created"))
assert.True(t, strings.Contains(result, "data: "))
assert.True(t, strings.Contains(result, "\n\n"))
}
func TestFormatSSE_InvalidPayload(t *testing.T) {
event := wspkg.SSEEvent{
Type: "test",
Payload: json.Marshal, // function can't be marshaled
}
_, err := wspkg.FormatSSE(event)
assert.Error(t, err)
}