75 lines
2.0 KiB
Go
75 lines
2.0 KiB
Go
package controller
|
|
|
|
import (
|
|
"bufio"
|
|
"context"
|
|
"errors"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/gin-gonic/gin"
|
|
"github.com/stretchr/testify/require"
|
|
)
|
|
|
|
func TestGroupStatusStreamPushAndReconnect(t *testing.T) {
|
|
var reads atomic.Int32
|
|
router := gin.New()
|
|
router.GET("/events", func(c *gin.Context) {
|
|
streamGroupStatus(c, func(context.Context) (any, error) {
|
|
return gin.H{"success": true, "revision": reads.Add(1)}, nil
|
|
})
|
|
})
|
|
server := httptest.NewServer(router)
|
|
defer server.Close()
|
|
client := &http.Client{Timeout: 3 * time.Second}
|
|
response, err := client.Get(server.URL + "/events")
|
|
require.NoError(t, err)
|
|
defer response.Body.Close()
|
|
require.Equal(t, "text/event-stream", response.Header.Get("Content-Type"))
|
|
require.Equal(t, "no", response.Header.Get("X-Accel-Buffering"))
|
|
reader := bufio.NewReader(response.Body)
|
|
frame := func() string {
|
|
var value string
|
|
for {
|
|
line, err := reader.ReadString('\n')
|
|
require.NoError(t, err)
|
|
value += line
|
|
if line == "\n" {
|
|
return value
|
|
}
|
|
}
|
|
}
|
|
require.Contains(t, frame(), `"revision":1`)
|
|
require.EqualValues(t, 1, reads.Load())
|
|
publishGroupStatusChange()
|
|
require.Contains(t, frame(), `"revision":2`)
|
|
response.Body.Close()
|
|
response, err = client.Get(server.URL + "/events")
|
|
require.NoError(t, err)
|
|
defer response.Body.Close()
|
|
reader = bufio.NewReader(response.Body)
|
|
require.Contains(t, frame(), `"revision":3`)
|
|
}
|
|
|
|
func TestGroupStatusStreamReadFailure(t *testing.T) {
|
|
c, _ := gin.CreateTestContext(httptest.NewRecorder())
|
|
c.Request = httptest.NewRequest("GET", "/events", nil)
|
|
streamGroupStatus(c, func(context.Context) (any, error) { return nil, errors.New("unavailable") })
|
|
require.Equal(t, http.StatusServiceUnavailable, c.Writer.Status())
|
|
}
|
|
|
|
func TestGroupStatusEventsCoalesceAndRelease(t *testing.T) {
|
|
ch, unsubscribe, err := subscribeGroupStatus()
|
|
require.NoError(t, err)
|
|
publishGroupStatusChange()
|
|
publishGroupStatusChange()
|
|
require.Len(t, ch, 1)
|
|
unsubscribe()
|
|
<-ch
|
|
publishGroupStatusChange()
|
|
require.Empty(t, ch)
|
|
}
|