Martino Ferrari
105 lines
3.0 KiB
Go
105 lines
3.0 KiB
Go
package ca_test
|
|
|
|
import (
|
|
"context"
|
|
"math"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/uopi/goca/proto"
|
|
"github.com/uopi/goca/testca"
|
|
)
|
|
|
|
// TestGetDisconnectMidRequest covers the "circuit disconnected during GET" path
|
|
// in conn.go: the server drops the connection while a READ_NOTIFY is in flight,
|
|
// so the in-flight reply channel is closed by the reconnect loop and the waiter
|
|
// unblocks with ok=false.
|
|
func TestGetDisconnectMidRequest(t *testing.T) {
|
|
srv := newTestServer(t)
|
|
cli := newTestClient(t, srv)
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
|
|
// First GET establishes the channel/circuit successfully.
|
|
if _, err := cli.Get(ctx, "TEST:DOUBLE"); err != nil {
|
|
t.Fatalf("initial Get: %v", err)
|
|
}
|
|
|
|
// Arm a mid-request disconnect for the next READ_NOTIFY.
|
|
srv.SetGetFault(testca.GetFaultDisconnect)
|
|
|
|
if _, err := cli.Get(ctx, "TEST:DOUBLE"); err == nil {
|
|
t.Fatal("Get with mid-request disconnect: want error")
|
|
}
|
|
}
|
|
|
|
// TestGetCorruptReply covers the "failed to decode GET reply" path: the server
|
|
// answers READ_NOTIFY with a payload too short for DecodeTimeValue.
|
|
func TestGetCorruptReply(t *testing.T) {
|
|
srv := newTestServer(t)
|
|
cli := newTestClient(t, srv)
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
|
|
// Establish the channel first so resolution succeeds.
|
|
if _, err := cli.Get(ctx, "TEST:DOUBLE"); err != nil {
|
|
t.Fatalf("initial Get: %v", err)
|
|
}
|
|
|
|
srv.SetGetFault(testca.GetFaultCorrupt)
|
|
|
|
if _, err := cli.Get(ctx, "TEST:DOUBLE"); err == nil {
|
|
t.Fatal("Get with corrupt reply: want error")
|
|
}
|
|
}
|
|
|
|
// TestServerDisconnectReconnects covers the CmdServerDisc dispatch branch and the
|
|
// reconnect loop: after an orderly server disconnect the client transparently
|
|
// reconnects, re-creates its channel, re-subscribes the monitor, and continues to
|
|
// receive updates.
|
|
func TestServerDisconnectReconnects(t *testing.T) {
|
|
srv := newTestServer(t)
|
|
cli := newTestClient(t, srv)
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
|
|
ch := make(chan proto.TimeValue, 16)
|
|
unsub, err := cli.Subscribe(ctx, "TEST:DOUBLE", ch)
|
|
if err != nil {
|
|
t.Fatalf("Subscribe: %v", err)
|
|
}
|
|
defer unsub()
|
|
|
|
// Drain the initial value.
|
|
select {
|
|
case <-ch:
|
|
case <-ctx.Done():
|
|
t.Fatal("timeout waiting for initial value")
|
|
}
|
|
|
|
// Force an orderly disconnect; the client must reconnect on its own.
|
|
srv.Disconnect()
|
|
|
|
// After the reconnect settles, push a fresh value and expect to see it.
|
|
// The reconnect back-off is ~1s, so poll generously and re-arm the value
|
|
// until it is observed (the re-subscribe also delivers an initial value).
|
|
deadline := time.After(8 * time.Second)
|
|
tick := time.NewTicker(300 * time.Millisecond)
|
|
defer tick.Stop()
|
|
for {
|
|
select {
|
|
case tv := <-ch:
|
|
if math.Abs(tv.Double-88.8) < 1e-6 {
|
|
return // reconnect succeeded and the update flowed through
|
|
}
|
|
case <-tick.C:
|
|
_ = srv.SetValue("TEST:DOUBLE", 88.8)
|
|
case <-deadline:
|
|
t.Fatal("timeout waiting for post-reconnect update")
|
|
}
|
|
}
|
|
}
|