239 lines
7.1 KiB
Go
239 lines
7.1 KiB
Go
package mesh
|
|
|
|
import (
|
|
"context"
|
|
"net/http/httptest"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/example/remote-access-platform/agents/rap-node-agent/internal/fabricproto"
|
|
)
|
|
|
|
func TestClientFabricSessionFrameRoundTrip(t *testing.T) {
|
|
server := httptest.NewServer(Server{
|
|
Local: PeerIdentity{ClusterID: "cluster-1", NodeID: "node-a"},
|
|
FabricSessionEnabled: true,
|
|
}.Handler())
|
|
defer server.Close()
|
|
|
|
client := NewClient(server.URL)
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
|
defer cancel()
|
|
response, err := client.SendFabricSessionFrame(ctx, FabricSessionDialOptions{
|
|
Token: "rap_fsn_clienttest",
|
|
Timeout: time.Second,
|
|
}, fabricproto.Frame{
|
|
Type: fabricproto.FramePing,
|
|
Sequence: 12,
|
|
Payload: []byte("probe"),
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("send fabric session frame: %v", err)
|
|
}
|
|
if response.Type != fabricproto.FramePong || response.Sequence != 12 || string(response.Payload) != "probe" {
|
|
t.Fatalf("response = %+v, want pong seq 12", response)
|
|
}
|
|
}
|
|
|
|
func TestClientFabricSessionPersistentRoundTrips(t *testing.T) {
|
|
server := httptest.NewServer(Server{
|
|
Local: PeerIdentity{ClusterID: "cluster-1", NodeID: "node-a"},
|
|
FabricSessionEnabled: true,
|
|
}.Handler())
|
|
defer server.Close()
|
|
|
|
client := NewClient(server.URL)
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
|
defer cancel()
|
|
session, _, err := client.OpenFabricSession(ctx, FabricSessionDialOptions{
|
|
Token: "rap_fsn_persistent",
|
|
Timeout: time.Second,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("open fabric session: %v", err)
|
|
}
|
|
defer session.Close()
|
|
|
|
first, err := session.RoundTrip(ctx, fabricproto.Frame{
|
|
Type: fabricproto.FramePing,
|
|
Sequence: 1,
|
|
Payload: []byte("first"),
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("first round trip: %v", err)
|
|
}
|
|
second, err := session.RoundTrip(ctx, fabricproto.Frame{
|
|
Type: fabricproto.FramePing,
|
|
Sequence: 2,
|
|
Payload: []byte("second"),
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("second round trip: %v", err)
|
|
}
|
|
if first.Type != fabricproto.FramePong || first.Sequence != 1 || string(first.Payload) != "first" {
|
|
t.Fatalf("first response = %+v, want pong seq 1", first)
|
|
}
|
|
if second.Type != fabricproto.FramePong || second.Sequence != 2 || string(second.Payload) != "second" {
|
|
t.Fatalf("second response = %+v, want pong seq 2", second)
|
|
}
|
|
}
|
|
|
|
func TestClientFabricSessionPersistentDataAcks(t *testing.T) {
|
|
server := httptest.NewServer(Server{
|
|
Local: PeerIdentity{ClusterID: "cluster-1", NodeID: "node-a"},
|
|
FabricSessionEnabled: true,
|
|
}.Handler())
|
|
defer server.Close()
|
|
|
|
client := NewClient(server.URL)
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
|
defer cancel()
|
|
session, _, err := client.OpenFabricSession(ctx, FabricSessionDialOptions{
|
|
Token: "rap_fsn_dataacks",
|
|
Timeout: time.Second,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("open fabric session: %v", err)
|
|
}
|
|
defer session.Close()
|
|
|
|
if err := session.WriteFrame(ctx, fabricproto.Frame{
|
|
Type: fabricproto.FrameOpenStream,
|
|
StreamID: 77,
|
|
TrafficClass: fabricproto.TrafficClassInteractive,
|
|
}); err != nil {
|
|
t.Fatalf("open stream frame: %v", err)
|
|
}
|
|
|
|
first, err := session.RoundTrip(ctx, fabricproto.Frame{
|
|
Type: fabricproto.FrameData,
|
|
StreamID: 77,
|
|
Sequence: 10,
|
|
TrafficClass: fabricproto.TrafficClassInteractive,
|
|
Payload: []byte("first payload"),
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("first data round trip: %v", err)
|
|
}
|
|
second, err := session.RoundTrip(ctx, fabricproto.Frame{
|
|
Type: fabricproto.FrameData,
|
|
StreamID: 77,
|
|
Sequence: 11,
|
|
TrafficClass: fabricproto.TrafficClassInteractive,
|
|
Payload: []byte("second payload"),
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("second data round trip: %v", err)
|
|
}
|
|
if first.Type != fabricproto.FrameAck || first.StreamID != 77 || first.Sequence != 10 {
|
|
t.Fatalf("first ack = %+v, want stream 77 seq 10", first)
|
|
}
|
|
if second.Type != fabricproto.FrameAck || second.StreamID != 77 || second.Sequence != 11 {
|
|
t.Fatalf("second ack = %+v, want stream 77 seq 11", second)
|
|
}
|
|
}
|
|
|
|
func TestClientFabricSessionPumpMovesIndependentFrames(t *testing.T) {
|
|
server := httptest.NewServer(Server{
|
|
Local: PeerIdentity{ClusterID: "cluster-1", NodeID: "node-a"},
|
|
FabricSessionEnabled: true,
|
|
}.Handler())
|
|
defer server.Close()
|
|
|
|
client := NewClient(server.URL)
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
|
defer cancel()
|
|
session, _, err := client.OpenFabricSession(ctx, FabricSessionDialOptions{
|
|
Token: "rap_fsn_pump",
|
|
Timeout: time.Second,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("open fabric session: %v", err)
|
|
}
|
|
pump := session.StartPump(ctx, FabricSessionPumpOptions{
|
|
OutboundBuffer: 4,
|
|
InboundBuffer: 4,
|
|
ErrorBuffer: 4,
|
|
})
|
|
defer pump.Close()
|
|
|
|
if err := pump.Send(ctx, fabricproto.Frame{
|
|
Type: fabricproto.FrameOpenStream,
|
|
StreamID: 900,
|
|
TrafficClass: fabricproto.TrafficClassBulk,
|
|
}); err != nil {
|
|
t.Fatalf("send open bulk stream: %v", err)
|
|
}
|
|
if err := pump.Send(ctx, fabricproto.Frame{
|
|
Type: fabricproto.FrameData,
|
|
StreamID: 900,
|
|
Sequence: 31,
|
|
TrafficClass: fabricproto.TrafficClassBulk,
|
|
Payload: []byte("bulk payload"),
|
|
}); err != nil {
|
|
t.Fatalf("send bulk data: %v", err)
|
|
}
|
|
if err := pump.Send(ctx, fabricproto.Frame{
|
|
Type: fabricproto.FramePing,
|
|
Sequence: 32,
|
|
Payload: []byte("control ping"),
|
|
}); err != nil {
|
|
t.Fatalf("send ping: %v", err)
|
|
}
|
|
|
|
gotAck := false
|
|
gotPong := false
|
|
for !gotAck || !gotPong {
|
|
select {
|
|
case frame := <-pump.Frames():
|
|
switch {
|
|
case frame.Type == fabricproto.FrameAck && frame.StreamID == 900 && frame.Sequence == 31:
|
|
gotAck = true
|
|
case frame.Type == fabricproto.FramePong && frame.Sequence == 32 && string(frame.Payload) == "control ping":
|
|
gotPong = true
|
|
}
|
|
case err := <-pump.Errors():
|
|
t.Fatalf("pump error: %v", err)
|
|
case <-ctx.Done():
|
|
t.Fatalf("timed out waiting for pump frames: ack=%v pong=%v", gotAck, gotPong)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestClientFabricSessionReportsRejectedStatus(t *testing.T) {
|
|
server := httptest.NewServer(Server{
|
|
Local: PeerIdentity{ClusterID: "cluster-1", NodeID: "node-a"},
|
|
FabricSessionEnabled: true,
|
|
}.Handler())
|
|
defer server.Close()
|
|
|
|
client := NewClient(server.URL)
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
|
defer cancel()
|
|
_, err := client.SendFabricSessionFrame(ctx, FabricSessionDialOptions{}, fabricproto.Frame{Type: fabricproto.FramePing})
|
|
if err == nil {
|
|
t.Fatal("send fabric session without token unexpectedly succeeded")
|
|
}
|
|
}
|
|
|
|
func TestClientFabricSessionWebSocketURL(t *testing.T) {
|
|
cases := []struct {
|
|
base string
|
|
want string
|
|
}{
|
|
{base: "http://node.example", want: "ws://node.example/mesh/v1/fabric/session/ws"},
|
|
{base: "https://node.example/base/", want: "wss://node.example/base/mesh/v1/fabric/session/ws"},
|
|
{base: "ws://node.example", want: "ws://node.example/mesh/v1/fabric/session/ws"},
|
|
}
|
|
for _, tc := range cases {
|
|
client := NewClient(tc.base)
|
|
got, err := client.fabricSessionWebSocketURL()
|
|
if err != nil {
|
|
t.Fatalf("fabricSessionWebSocketURL(%q): %v", tc.base, err)
|
|
}
|
|
if got != tc.want {
|
|
t.Fatalf("fabricSessionWebSocketURL(%q) = %q, want %q", tc.base, got, tc.want)
|
|
}
|
|
}
|
|
}
|