Skip to content

Commit ecc6740

Browse files
committed
test(runed): mock-server unit tests + NewBufconnClient injection
12 test functions / 17 subtest cases against an in-process bufconn-backed RunedServiceServer: - EmbedSingle: happy path, retry-then-success on Unavailable, non-retryable error returns immediately, retries-exhausted (3 attempts), respects cancelled context - EmbedBatch: no-split when len ≤ MaxBatchSize, splits 9-into-(4+4+1) when above MaxBatchSize=4, resp count mismatch errors out, empty input no-op - Info: sync.Once caches across 5 calls (1 RPC), cached error path (1 RPC for two failed calls) - Health: 5-case enum→string matrix (OK/LOADING/DEGRADED/SHUTTING_DOWN/ UNSPECIFIED), uptime/total_requests pass-through Adds NewBufconnClient(*grpc.ClientConn) Client constructor (factored through newWithConn) so tests can inject a bufconn-backed conn without a unix socket. Useful in production too for callers that pool conns. Notes: - Two retry tests (RetryThenSuccess: 0.5s, RetriesExhaust: 2.5s) wait on the real D7 backoff schedule [0, 500ms, 2s]. Total test suite ~3.5s. - Tests use status.FromError + codes.* for assertions; doesn't depend on any embedder/errors.go (which doesn't exist yet — see runed audit P2.4). go test ./internal/adapters/embedder/... ok
1 parent 309e76d commit ecc6740

2 files changed

Lines changed: 357 additions & 1 deletion

File tree

‎internal/adapters/embedder/client.go‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -75,13 +75,24 @@ func New(sockPath string) (Client, error) {
7575
if err != nil {
7676
return nil, fmt.Errorf("embedder: grpc dial %s: %w", sockPath, err)
7777
}
78+
return newWithConn(sockPath, conn), nil
79+
}
80+
81+
// NewBufconnClient wraps an existing *grpc.ClientConn (e.g., from
82+
// google.golang.org/grpc/test/bufconn) so tests can exercise the same RPC
83+
// path without needing a real unix socket / runed daemon.
84+
func NewBufconnClient(conn *grpc.ClientConn) Client {
85+
return newWithConn("bufconn", conn)
86+
}
87+
88+
func newWithConn(sockPath string, conn *grpc.ClientConn) *client {
7889
pb := runedv1.NewRunedServiceClient(conn)
7990
return &client{
8091
sockPath: sockPath,
8192
conn: conn,
8293
pb: pb,
8394
info: &infoCache{svc: pb},
84-
}, nil
95+
}
8596
}
8697

8798
func (c *client) EmbedSingle(ctx context.Context, text string) ([]float32, error) {
Lines changed: 345 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,345 @@
1+
package embedder_test
2+
3+
import (
4+
"context"
5+
"errors"
6+
"net"
7+
"strings"
8+
"sync/atomic"
9+
"testing"
10+
"time"
11+
12+
runedv1 "github.com/CryptoLabInc/runed/gen/runed/v1"
13+
"google.golang.org/grpc"
14+
"google.golang.org/grpc/codes"
15+
"google.golang.org/grpc/credentials/insecure"
16+
"google.golang.org/grpc/status"
17+
"google.golang.org/grpc/test/bufconn"
18+
19+
"github.com/envector/rune-go/internal/adapters/embedder"
20+
)
21+
22+
// fakeRuned implements RunedServiceServer in-process. Each RPC delegates to
23+
// a per-test func; tests configure those before calling the client.
24+
type fakeRuned struct {
25+
runedv1.UnimplementedRunedServiceServer
26+
27+
embedFn func(*runedv1.EmbedRequest) (*runedv1.EmbedResponse, error)
28+
embedBatchFn func(*runedv1.EmbedBatchRequest) (*runedv1.EmbedBatchResponse, error)
29+
infoFn func(*runedv1.InfoRequest) (*runedv1.InfoResponse, error)
30+
healthFn func(*runedv1.HealthRequest) (*runedv1.HealthResponse, error)
31+
32+
infoCalls int32 // atomic — Info should be invoked exactly once across the lifetime of an infoCache
33+
embedCalls int32 // atomic — used by retry test to count attempts
34+
embedBatchCalls int32 // atomic — used by batch-split test
35+
}
36+
37+
func (f *fakeRuned) Embed(_ context.Context, req *runedv1.EmbedRequest) (*runedv1.EmbedResponse, error) {
38+
atomic.AddInt32(&f.embedCalls, 1)
39+
if f.embedFn != nil {
40+
return f.embedFn(req)
41+
}
42+
return nil, status.Error(codes.Unimplemented, "Embed not stubbed")
43+
}
44+
45+
func (f *fakeRuned) EmbedBatch(_ context.Context, req *runedv1.EmbedBatchRequest) (*runedv1.EmbedBatchResponse, error) {
46+
atomic.AddInt32(&f.embedBatchCalls, 1)
47+
if f.embedBatchFn != nil {
48+
return f.embedBatchFn(req)
49+
}
50+
return nil, status.Error(codes.Unimplemented, "EmbedBatch not stubbed")
51+
}
52+
53+
func (f *fakeRuned) Info(_ context.Context, req *runedv1.InfoRequest) (*runedv1.InfoResponse, error) {
54+
atomic.AddInt32(&f.infoCalls, 1)
55+
if f.infoFn != nil {
56+
return f.infoFn(req)
57+
}
58+
return &runedv1.InfoResponse{
59+
DaemonVersion: "0.1.0-test",
60+
ModelIdentity: "qwen3-test",
61+
VectorDim: 1024,
62+
MaxTextLength: 8192,
63+
MaxBatchSize: 4,
64+
}, nil
65+
}
66+
67+
func (f *fakeRuned) Health(_ context.Context, req *runedv1.HealthRequest) (*runedv1.HealthResponse, error) {
68+
if f.healthFn != nil {
69+
return f.healthFn(req)
70+
}
71+
return &runedv1.HealthResponse{Status: runedv1.HealthResponse_STATUS_OK}, nil
72+
}
73+
74+
func startFakeRuned(t *testing.T) (*fakeRuned, embedder.Client) {
75+
t.Helper()
76+
lis := bufconn.Listen(1 << 20)
77+
srv := grpc.NewServer()
78+
fake := &fakeRuned{}
79+
runedv1.RegisterRunedServiceServer(srv, fake)
80+
go func() { _ = srv.Serve(lis) }()
81+
t.Cleanup(srv.Stop)
82+
83+
conn, err := grpc.NewClient(
84+
"passthrough://bufconn",
85+
grpc.WithContextDialer(func(_ context.Context, _ string) (net.Conn, error) {
86+
return lis.DialContext(context.Background())
87+
}),
88+
grpc.WithTransportCredentials(insecure.NewCredentials()),
89+
)
90+
if err != nil {
91+
t.Fatalf("grpc dial: %v", err)
92+
}
93+
t.Cleanup(func() { _ = conn.Close() })
94+
95+
return fake, embedder.NewBufconnClient(conn)
96+
}
97+
98+
func TestEmbedSingle_HappyPath(t *testing.T) {
99+
fake, c := startFakeRuned(t)
100+
fake.embedFn = func(req *runedv1.EmbedRequest) (*runedv1.EmbedResponse, error) {
101+
if req.GetText() != "hello" {
102+
t.Errorf("text: got %q, want hello", req.GetText())
103+
}
104+
return &runedv1.EmbedResponse{Vector: []float32{0.1, 0.2, 0.3}}, nil
105+
}
106+
107+
v, err := c.EmbedSingle(context.Background(), "hello")
108+
if err != nil {
109+
t.Fatalf("EmbedSingle: %v", err)
110+
}
111+
if len(v) != 3 || v[0] != 0.1 {
112+
t.Errorf("vector: got %v", v)
113+
}
114+
}
115+
116+
func TestEmbedSingle_RetryThenSuccess(t *testing.T) {
117+
fake, c := startFakeRuned(t)
118+
var attempts int32
119+
fake.embedFn = func(*runedv1.EmbedRequest) (*runedv1.EmbedResponse, error) {
120+
n := atomic.AddInt32(&attempts, 1)
121+
if n < 2 {
122+
return nil, status.Error(codes.Unavailable, "transient")
123+
}
124+
return &runedv1.EmbedResponse{Vector: []float32{1}}, nil
125+
}
126+
127+
v, err := c.EmbedSingle(context.Background(), "hi")
128+
if err != nil {
129+
t.Fatalf("retry should succeed: %v", err)
130+
}
131+
if len(v) != 1 || v[0] != 1 {
132+
t.Errorf("vector: got %v", v)
133+
}
134+
if atomic.LoadInt32(&attempts) != 2 {
135+
t.Errorf("attempts: got %d, want 2", atomic.LoadInt32(&attempts))
136+
}
137+
}
138+
139+
func TestEmbedSingle_NonRetryableErrorReturnsImmediately(t *testing.T) {
140+
fake, c := startFakeRuned(t)
141+
fake.embedFn = func(*runedv1.EmbedRequest) (*runedv1.EmbedResponse, error) {
142+
return nil, status.Error(codes.InvalidArgument, "text too long")
143+
}
144+
145+
_, err := c.EmbedSingle(context.Background(), "x")
146+
if err == nil {
147+
t.Fatal("expected error")
148+
}
149+
st, ok := status.FromError(err)
150+
if !ok {
151+
t.Fatalf("expected gRPC status, got %v", err)
152+
}
153+
if st.Code() != codes.InvalidArgument {
154+
t.Errorf("code: got %v, want InvalidArgument", st.Code())
155+
}
156+
if atomic.LoadInt32(&fake.embedCalls) != 1 {
157+
t.Errorf("attempts: got %d, want 1 (no retry)", atomic.LoadInt32(&fake.embedCalls))
158+
}
159+
}
160+
161+
func TestEmbedBatch_NoSplitWhenLEMax(t *testing.T) {
162+
fake, c := startFakeRuned(t)
163+
// Info default: MaxBatchSize=4. Pass 3 texts → single chunk.
164+
fake.embedBatchFn = func(req *runedv1.EmbedBatchRequest) (*runedv1.EmbedBatchResponse, error) {
165+
if len(req.GetTexts()) != 3 {
166+
t.Errorf("texts in single chunk: got %d, want 3", len(req.GetTexts()))
167+
}
168+
return &runedv1.EmbedBatchResponse{Embeddings: []*runedv1.EmbedResponse{
169+
{Vector: []float32{1}}, {Vector: []float32{2}}, {Vector: []float32{3}},
170+
}}, nil
171+
}
172+
173+
out, err := c.EmbedBatch(context.Background(), []string{"a", "b", "c"})
174+
if err != nil {
175+
t.Fatalf("EmbedBatch: %v", err)
176+
}
177+
if len(out) != 3 {
178+
t.Errorf("output: got %d, want 3", len(out))
179+
}
180+
if atomic.LoadInt32(&fake.embedBatchCalls) != 1 {
181+
t.Errorf("EmbedBatch RPC calls: got %d, want 1", atomic.LoadInt32(&fake.embedBatchCalls))
182+
}
183+
}
184+
185+
func TestEmbedBatch_SplitsWhenAboveMax(t *testing.T) {
186+
fake, c := startFakeRuned(t)
187+
fake.embedBatchFn = func(req *runedv1.EmbedBatchRequest) (*runedv1.EmbedBatchResponse, error) {
188+
// Echo a vector per input
189+
emb := make([]*runedv1.EmbedResponse, len(req.GetTexts()))
190+
for i, txt := range req.GetTexts() {
191+
emb[i] = &runedv1.EmbedResponse{Vector: []float32{float32(len(txt))}}
192+
}
193+
return &runedv1.EmbedBatchResponse{Embeddings: emb}, nil
194+
}
195+
196+
// MaxBatchSize=4 (default Info), 9 texts → 3 chunks (4+4+1)
197+
texts := []string{"a", "b", "c", "d", "e", "f", "g", "h", "i"}
198+
out, err := c.EmbedBatch(context.Background(), texts)
199+
if err != nil {
200+
t.Fatalf("EmbedBatch: %v", err)
201+
}
202+
if len(out) != 9 {
203+
t.Errorf("output: got %d, want 9", len(out))
204+
}
205+
if atomic.LoadInt32(&fake.embedBatchCalls) != 3 {
206+
t.Errorf("EmbedBatch RPC calls: got %d, want 3 (4+4+1)", atomic.LoadInt32(&fake.embedBatchCalls))
207+
}
208+
}
209+
210+
func TestEmbedBatch_RespCountMismatchErrors(t *testing.T) {
211+
fake, c := startFakeRuned(t)
212+
fake.embedBatchFn = func(req *runedv1.EmbedBatchRequest) (*runedv1.EmbedBatchResponse, error) {
213+
// Return ONE less embedding than requested — contract violation
214+
return &runedv1.EmbedBatchResponse{Embeddings: []*runedv1.EmbedResponse{
215+
{Vector: []float32{1}},
216+
}}, nil
217+
}
218+
219+
_, err := c.EmbedBatch(context.Background(), []string{"a", "b"})
220+
if err == nil {
221+
t.Fatal("expected error on resp count mismatch")
222+
}
223+
if !strings.Contains(err.Error(), "expected 2 embeddings") {
224+
t.Errorf("error: got %q", err.Error())
225+
}
226+
}
227+
228+
func TestEmbedBatch_EmptyInput(t *testing.T) {
229+
_, c := startFakeRuned(t)
230+
out, err := c.EmbedBatch(context.Background(), nil)
231+
if err != nil {
232+
t.Errorf("empty input: got error %v, want nil", err)
233+
}
234+
if out != nil {
235+
t.Errorf("empty input: got %v, want nil", out)
236+
}
237+
}
238+
239+
func TestInfo_CachesAcrossCalls(t *testing.T) {
240+
fake, c := startFakeRuned(t)
241+
for i := 0; i < 5; i++ {
242+
_, err := c.Info(context.Background())
243+
if err != nil {
244+
t.Fatalf("Info: %v", err)
245+
}
246+
}
247+
if got := atomic.LoadInt32(&fake.infoCalls); got != 1 {
248+
t.Errorf("Info RPC calls: got %d, want 1 (sync.Once cache)", got)
249+
}
250+
}
251+
252+
func TestInfo_CachesError(t *testing.T) {
253+
fake, c := startFakeRuned(t)
254+
fake.infoFn = func(*runedv1.InfoRequest) (*runedv1.InfoResponse, error) {
255+
return nil, status.Error(codes.Unavailable, "down")
256+
}
257+
_, err1 := c.Info(context.Background())
258+
if err1 == nil {
259+
t.Fatal("first Info: expected error")
260+
}
261+
_, err2 := c.Info(context.Background())
262+
if err2 == nil {
263+
t.Fatal("second Info: expected cached error")
264+
}
265+
// Both calls return the same cached error; only one RPC fires.
266+
if got := atomic.LoadInt32(&fake.infoCalls); got != 1 {
267+
t.Errorf("Info RPC calls: got %d, want 1 (error caches too)", got)
268+
}
269+
}
270+
271+
func TestHealth_StatusEnumMapping(t *testing.T) {
272+
cases := []struct {
273+
grpcStatus runedv1.HealthResponse_Status
274+
want string
275+
}{
276+
{runedv1.HealthResponse_STATUS_OK, "OK"},
277+
{runedv1.HealthResponse_STATUS_LOADING, "LOADING"},
278+
{runedv1.HealthResponse_STATUS_DEGRADED, "DEGRADED"},
279+
{runedv1.HealthResponse_STATUS_SHUTTING_DOWN, "SHUTTING_DOWN"},
280+
{runedv1.HealthResponse_STATUS_UNSPECIFIED, "UNSPECIFIED"},
281+
}
282+
for _, tc := range cases {
283+
t.Run(tc.want, func(t *testing.T) {
284+
fake, c := startFakeRuned(t)
285+
fake.healthFn = func(*runedv1.HealthRequest) (*runedv1.HealthResponse, error) {
286+
return &runedv1.HealthResponse{
287+
Status: tc.grpcStatus,
288+
UptimeSeconds: 42,
289+
TotalRequests: 7,
290+
}, nil
291+
}
292+
snap, err := c.Health(context.Background())
293+
if err != nil {
294+
t.Fatalf("Health: %v", err)
295+
}
296+
if snap.Status != tc.want {
297+
t.Errorf("Status: got %q, want %q", snap.Status, tc.want)
298+
}
299+
if snap.UptimeSeconds != 42 || snap.TotalRequests != 7 {
300+
t.Errorf("counters: got %+v", snap)
301+
}
302+
})
303+
}
304+
}
305+
306+
func TestEmbedSingle_RetriesExhaust(t *testing.T) {
307+
fake, c := startFakeRuned(t)
308+
fake.embedFn = func(*runedv1.EmbedRequest) (*runedv1.EmbedResponse, error) {
309+
return nil, status.Error(codes.Unavailable, "permanently down")
310+
}
311+
312+
// Use a context with deadline so the third backoff doesn't drag the test.
313+
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
314+
defer cancel()
315+
316+
_, err := c.EmbedSingle(ctx, "x")
317+
if err == nil {
318+
t.Fatal("expected error after retries exhausted")
319+
}
320+
if !strings.Contains(err.Error(), "all retries exhausted") {
321+
t.Errorf("error: got %q, want 'all retries exhausted'", err.Error())
322+
}
323+
// 3 attempts: backoff [0, 500ms, 2s] × 3 → 3 calls
324+
if got := atomic.LoadInt32(&fake.embedCalls); got != 3 {
325+
t.Errorf("Embed RPC calls: got %d, want 3", got)
326+
}
327+
}
328+
329+
func TestEmbedSingle_RespectsCancelledContext(t *testing.T) {
330+
fake, c := startFakeRuned(t)
331+
fake.embedFn = func(*runedv1.EmbedRequest) (*runedv1.EmbedResponse, error) {
332+
return nil, status.Error(codes.Unavailable, "transient")
333+
}
334+
335+
ctx, cancel := context.WithCancel(context.Background())
336+
cancel() // already cancelled before call
337+
338+
_, err := c.EmbedSingle(ctx, "x")
339+
if err == nil {
340+
t.Fatal("expected error from cancelled ctx")
341+
}
342+
if !errors.Is(err, context.Canceled) && !strings.Contains(err.Error(), "context canceled") {
343+
t.Errorf("error: got %v, want context.Canceled", err)
344+
}
345+
}

0 commit comments

Comments
 (0)