111 lines
3.2 KiB
Go
111 lines
3.2 KiB
Go
package stream
|
|
|
|
import (
|
|
"encoding/json"
|
|
"errors"
|
|
"io"
|
|
"reflect"
|
|
"strings"
|
|
"testing"
|
|
)
|
|
|
|
const evJSON = `{"event_id":"evt_1","timestamp":1,"type":"node_command","node_id":"n1","payload":{"action":"drain"}}`
|
|
|
|
var evWant = Event{
|
|
ID: "evt_1",
|
|
Timestamp: 1,
|
|
Type: EventNodeCommand,
|
|
NodeID: "n1",
|
|
Payload: json.RawMessage(`{"action":"drain"}`),
|
|
}
|
|
|
|
func decodeAll(t *testing.T, input string) ([]Event, error) {
|
|
t.Helper()
|
|
dec := NewDecoder(strings.NewReader(input))
|
|
var out []Event
|
|
for {
|
|
ev, err := dec.Next()
|
|
if err != nil {
|
|
return out, err
|
|
}
|
|
out = append(out, ev)
|
|
}
|
|
}
|
|
|
|
func TestDecoder(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
input string
|
|
want []Event
|
|
}{
|
|
{"einzelnes Event", "id: evt_1\nevent: node_command\ndata: " + evJSON + "\n\n", []Event{evWant}},
|
|
{"Keepalives werden übersprungen", ":\n\n:\n\ndata: " + evJSON + "\n\n:\n\n", []Event{evWant}},
|
|
{"CRLF-Zeilenenden", "id: evt_1\r\ndata: " + evJSON + "\r\n\r\n", []Event{evWant}},
|
|
{"data ohne Leerzeichen", "data:" + evJSON + "\n\n", []Event{evWant}},
|
|
{
|
|
"mehrere data-Zeilen",
|
|
"data: {\"event_id\":\"evt_1\",\"timestamp\":1,\ndata: \"type\":\"node_command\",\"node_id\":\"n1\",\ndata: \"payload\":{\"action\":\"drain\"}}\n\n",
|
|
[]Event{evWant},
|
|
},
|
|
{"unbekannte Felder ignoriert", "retry: 1000\nfoo: bar\ndata: " + evJSON + "\n\n", []Event{evWant}},
|
|
{"Leerzeilen ohne Daten", "\n\n\nevent: x\n\n", nil},
|
|
{"zwei Events", "data: " + evJSON + "\n\ndata: " + evJSON + "\n\n", []Event{evWant, evWant}},
|
|
{"leerer Stream", "", nil},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
got, err := decodeAll(t, tt.input)
|
|
if !errors.Is(err, io.EOF) {
|
|
t.Fatalf("erwartet io.EOF am Ende, bekommen %v", err)
|
|
}
|
|
if !reflect.DeepEqual(got, tt.want) {
|
|
t.Errorf("Events weichen ab\n got: %+v\nwant: %+v", got, tt.want)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestDecoder_Errors(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
input string
|
|
want error
|
|
}{
|
|
{"ungültiges JSON", "data: {kaputt\n\n", ErrInvalidEvent},
|
|
{"Event ohne Typ", "data: {\"event_id\":\"evt_1\"}\n\n", ErrInvalidEvent},
|
|
{"abgeschnittenes Event", "data: " + evJSON + "\n", io.ErrUnexpectedEOF},
|
|
{"Zeile zu lang", "data: " + strings.Repeat("x", MaxEventSize+1) + "\n\n", ErrEventTooLarge},
|
|
{"Summe der data-Zeilen zu lang", strings.Repeat("data: "+strings.Repeat("x", MaxEventSize/4)+"\n", 5) + "\n", ErrEventTooLarge},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
_, err := decodeAll(t, tt.input)
|
|
if !errors.Is(err, tt.want) {
|
|
t.Errorf("erwartet %v, bekommen %v", tt.want, err)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestEncodeDecodeRoundTrip(t *testing.T) {
|
|
ev, err := NewEvent(EventServiceState, map[string]any{"labels": map[string]string{"traefik.enable": "true"}})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ev.NodeID = "n1"
|
|
frame, err := encode(ev)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !strings.HasPrefix(string(frame), "id: "+ev.ID+"\nevent: service_state\ndata: ") || !strings.HasSuffix(string(frame), "\n\n") {
|
|
t.Errorf("unerwarteter SSE-Rahmen: %q", frame)
|
|
}
|
|
got, err := NewDecoder(strings.NewReader(string(frame))).Next()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !reflect.DeepEqual(got, ev) {
|
|
t.Errorf("Round-Trip weicht ab\n got: %+v\nwant: %+v", got, ev)
|
|
}
|
|
}
|