bldr/plugin/res-lookup-rpc-service_test.go
Rama master · réplica de lectura del repositorio original.
package bldr_plugin
import (
"context"
"errors"
"io"
"sync"
"testing"
"github.com/aperturerobotics/controllerbus/directive"
"github.com/aperturerobotics/starpc/srpc"
bifrost_rpc "github.com/s4wave/spacewave/net/rpc"
)
func TestResolveLookupRpcServiceUsesPluginServiceIDPrefix(t *testing.T) {
// Resolve an RPC service addressed to the notes plugin.
serviceID := PluginServiceID("spacewave-notes", "resource.ResourceService")
resolver, err := ResolveLookupRpcService(
t.Context(),
bifrost_rpc.NewLookupRpcService(serviceID, ""),
nil,
)
if err != nil {
t.Fatal(err)
}
// Verify the resolver targets the plugin and strips its service prefix.
got, ok := resolver.(*LookupRpcServiceResolver)
if !ok {
t.Fatalf("expected LookupRpcServiceResolver, got %T", resolver)
}
if got.pluginID != "spacewave-notes" {
t.Fatalf("expected plugin id spacewave-notes, got %q", got.pluginID)
}
if got.stripServiceIDPrefix != "plugin/spacewave-notes/" {
t.Fatalf("expected service prefix strip, got %q", got.stripServiceIDPrefix)
}
}
func TestResolveLookupRpcServiceDoesNotRouteRequesterServerID(t *testing.T) {
resolver, err := ResolveLookupRpcService(
t.Context(),
bifrost_rpc.NewLookupRpcService(
"resource.ResourceService",
PluginServerID("spacewave-notes", ""),
),
nil,
)
if err != nil {
t.Fatal(err)
}
if resolver != nil {
t.Fatalf("expected plain service lookup with plugin requester server ID to be ignored, got %T", resolver)
}
}
func TestResolveLookupRpcServiceHandlesNilReleaseFunc(t *testing.T) {
// Regression: WaitPluginClient/WaitPluginHostClient return a nil release func
// on the loopback and plugin-host client paths. Resolve must not defer a nil
// func call, which previously panicked (nil func value, pc=0x0) when the
// deferred ran and crashed the whole plugin process.
resolver := NewLookupRpcServiceResolver(
&nilReleaseClientHandler{client: &testForwardingClient{}},
"mercury-core",
"plugin/mercury-core/",
)
ctx, cancel := context.WithCancel(t.Context())
cancel()
if err := resolver.Resolve(ctx, stubResolverHandler{}); err != context.Canceled {
t.Fatalf("expected context.Canceled, got %v", err)
}
}
func TestClientForwardingInvokerReturnsWhenServerCompletesBeforeCaller(t *testing.T) {
// Connect a completed server response to a caller that sends after closure.
outgoing := newTestForwardingStream(t.Context(), [][]byte{[]byte("server-response")})
local := &testForwardingLocalStream{
ctx: t.Context(),
recv: func(msg *srpc.RawMessage) error {
<-outgoing.closed
msg.SetData([]byte("late-client-message"))
return nil
},
}
client := &testForwardingClient{stream: outgoing}
// Forward the plugin RPC until the server completes the stream.
ok, err := newClientForwardingInvoker(client, "plugin/test/").
InvokeMethod("plugin/test/resource.ResourceService", "ResourceRpc", local)
// Verify completion, the forwarded target, and the caller's response.
if err != nil {
t.Fatal(err)
}
if !ok {
t.Fatal("expected forwarding invoker to handle the method")
}
if client.serviceID != "resource.ResourceService" || client.methodID != "ResourceRpc" {
t.Fatalf("unexpected forwarded target: %s/%s", client.serviceID, client.methodID)
}
if got := local.sentStrings(); len(got) != 1 || got[0] != "server-response" {
t.Fatalf("expected server response to reach caller, got %q", got)
}
}
func TestClientForwardingInvokerPreservesServerStreamingHalfClose(t *testing.T) {
// Prepare a server that responds only after the caller closes its send side.
outgoing := newTestForwardingStream(t.Context(), [][]byte{[]byte("stream-init")})
outgoing.waitCloseSendBeforeRecv = true
// Send one caller request before reporting the end of the request stream.
var sent bool
local := &testForwardingLocalStream{
ctx: t.Context(),
recv: func(msg *srpc.RawMessage) error {
// End the caller's request stream after its first message.
if sent {
return io.EOF
}
// Deliver the caller's single request to the forwarding invoker.
sent = true
msg.SetData([]byte("stream-request"))
return nil
},
}
// Forward the server-streaming RPC through the caller's half-close.
ok, err := newClientForwardingInvoker(&testForwardingClient{stream: outgoing}, "").
InvokeMethod("resource.ResourceService", "ResourceClient", local)
// Verify the request, outgoing half-close, and returned server response.
if err != nil {
t.Fatal(err)
}
if !ok {
t.Fatal("expected forwarding invoker to handle the method")
}
if got := outgoing.sentStrings(); len(got) != 1 || got[0] != "stream-request" {
t.Fatalf("expected caller request to reach server, got %q", got)
}
if !outgoing.closeSendCalled() {
t.Fatal("expected caller EOF to close the outgoing send side")
}
if got := local.sentStrings(); len(got) != 1 || got[0] != "stream-init" {
t.Fatalf("expected server-streaming response to reach caller, got %q", got)
}
}
type testForwardingClient struct {
stream *testForwardingStream
serviceID string
methodID string
}
func (c *testForwardingClient) ExecCall(context.Context, string, string, srpc.Message, srpc.Message) error {
return errors.New("unexpected ExecCall")
}
func (c *testForwardingClient) NewStream(
ctx context.Context,
serviceID string,
methodID string,
firstMsg srpc.Message,
) (srpc.Stream, error) {
// Reject an initial message that this forwarding test client does not accept.
if firstMsg != nil {
return nil, errors.New("unexpected first message")
}
// Record the forwarded RPC target and bind the stream to its context.
c.serviceID = serviceID
c.methodID = methodID
c.stream.ctx = ctx
return c.stream, nil
}
type testForwardingStream struct {
ctx context.Context
mu sync.Mutex
responses [][]byte
recvIdx int
sent [][]byte
closeSend bool
waitCloseSendBeforeRecv bool
closeSendCh chan struct{}
closed chan struct{}
closeSendDo sync.Once
closeDo sync.Once
}
func newTestForwardingStream(ctx context.Context, responses [][]byte) *testForwardingStream {
return &testForwardingStream{
ctx: ctx,
responses: responses,
closeSendCh: make(chan struct{}),
closed: make(chan struct{}),
}
}
func (s *testForwardingStream) Context() context.Context {
return s.ctx
}
func (s *testForwardingStream) MsgSend(msg srpc.Message) error {
// Reject messages sent after the forwarding stream closes.
select {
case <-s.closed:
return srpc.ErrCompleted
default:
}
// Encode the message sent to the forwarding stream.
data, err := msg.MarshalVT()
if err != nil {
return err
}
// Retain the forwarded message bytes for request assertions.
s.mu.Lock()
s.sent = append(s.sent, append([]byte(nil), data...))
s.mu.Unlock()
return nil
}
func (s *testForwardingStream) MsgRecv(msg srpc.Message) error {
// Wait for the caller's half-close when the first server response requires it.
s.mu.Lock()
wait := s.waitCloseSendBeforeRecv && s.recvIdx == 0
s.mu.Unlock()
if wait {
select {
case <-s.closeSendCh:
case <-s.closed:
return context.Canceled
case <-s.ctx.Done():
return context.Canceled
}
}
// Consume the next configured server response under the stream lock.
s.mu.Lock()
defer s.mu.Unlock()
if s.recvIdx >= len(s.responses) {
return io.EOF
}
data := append([]byte(nil), s.responses[s.recvIdx]...)
s.recvIdx++
// Deliver the configured response through the raw RPC message.
raw, ok := msg.(*srpc.RawMessage)
if !ok {
return errors.New("unexpected message type")
}
raw.SetData(data)
return nil
}
func (s *testForwardingStream) CloseSend() error {
s.closeSendDo.Do(func() {
s.mu.Lock()
s.closeSend = true
s.mu.Unlock()
close(s.closeSendCh)
})
return nil
}
func (s *testForwardingStream) Close() error {
s.closeDo.Do(func() {
close(s.closed)
})
return nil
}
func (s *testForwardingStream) sentStrings() []string {
// Hold the stream lock while reading its recorded messages.
s.mu.Lock()
defer s.mu.Unlock()
// Convert the recorded stream messages to strings for assertions.
out := make([]string, 0, len(s.sent))
for _, data := range s.sent {
out = append(out, string(data))
}
return out
}
func (s *testForwardingStream) closeSendCalled() bool {
s.mu.Lock()
defer s.mu.Unlock()
return s.closeSend
}
type testForwardingLocalStream struct {
ctx context.Context
recv func(*srpc.RawMessage) error
mu sync.Mutex
sent [][]byte
}
func (s *testForwardingLocalStream) Context() context.Context {
return s.ctx
}
func (s *testForwardingLocalStream) MsgSend(msg srpc.Message) error {
// Encode the server response delivered to the local stream.
data, err := msg.MarshalVT()
if err != nil {
return err
}
// Retain the local response bytes for caller assertions.
s.mu.Lock()
s.sent = append(s.sent, append([]byte(nil), data...))
s.mu.Unlock()
return nil
}
func (s *testForwardingLocalStream) MsgRecv(msg srpc.Message) error {
raw, ok := msg.(*srpc.RawMessage)
if !ok {
return errors.New("unexpected message type")
}
return s.recv(raw)
}
func (s *testForwardingLocalStream) CloseSend() error {
return nil
}
func (s *testForwardingLocalStream) Close() error {
return nil
}
func (s *testForwardingLocalStream) sentStrings() []string {
// Hold the local stream lock while reading its recorded responses.
s.mu.Lock()
defer s.mu.Unlock()
// Convert the recorded local responses to strings for assertions.
out := make([]string, 0, len(s.sent))
for _, data := range s.sent {
out = append(out, string(data))
}
return out
}
// nilReleaseClientHandler returns a non-nil client and a nil release func,
// matching the loopback and plugin-host client paths.
type nilReleaseClientHandler struct {
client srpc.Client
}
func (h *nilReleaseClientHandler) WaitPluginHostClient(context.Context, func()) (srpc.Client, func(), error) {
return h.client, nil, nil
}
func (h *nilReleaseClientHandler) WaitPluginClient(context.Context, func(), string) (srpc.Client, func(), error) {
return h.client, nil, nil
}
// stubResolverHandler is a no-op directive.ResolverHandler for resolver tests.
type stubResolverHandler struct{}
func (stubResolverHandler) AddValue(directive.Value) (uint32, bool) { return 1, true }
func (stubResolverHandler) RemoveValue(uint32) (directive.Value, bool) { return nil, true }
func (stubResolverHandler) CountValues(bool) int { return 0 }
func (stubResolverHandler) ClearValues() []uint32 { return nil }
func (stubResolverHandler) MarkIdle(bool) {}
func (stubResolverHandler) AddValueRemovedCallback(uint32, func()) func() { return func() {} }
func (stubResolverHandler) AddResolverRemovedCallback(func()) func() { return func() {} }
func (stubResolverHandler) AddResolver(directive.Resolver, func()) func() { return func() {} }
var (
_ srpc.Client = (*testForwardingClient)(nil)
_ srpc.Stream = (*testForwardingStream)(nil)
_ srpc.Stream = (*testForwardingLocalStream)(nil)
_ LookupRpcClientHandler = (*nilReleaseClientHandler)(nil)
_ directive.ResolverHandler = stubResolverHandler{}
)