返回 DeepSeek-Reasonix
stream_helpers_test.go
根目录 / internal / extension / providerext / stream_helpers_test.go
1 package providerext
2
3 import (
4 "context"
5 "reasonix/internal/extension/protocol"
6 "reasonix/internal/provider"
7 "testing"
8 "time"
9 )
10
11 func TestDefaultStreamIdleTimeoutIsFiveMinutes(t *testing.T) {
12 if defaultStreamIdleTimeout != 300*time.Second {
13 t.Fatalf("default stream idle timeout = %s, want 5m", defaultStreamIdleTimeout)
14 }
15 }
16
17 // openTestStream resolves the demo ref and opens a stream, returning the
18 // chunk channel and the stream ID the sidecar would address.
19 func openTestStream(t *testing.T, r *Resolver, fc *fakeClient, effort *string) (<-chan provider.Chunk, string) {
20 t.Helper()
21 p, err := r.Resolve(provider.Selection{Ref: "plugin/demo/fake/x", Effort: effort})
22 if err != nil {
23 t.Fatalf("Resolve: %v", err)
24 }
25 out, err := p.Stream(context.Background(), provider.Request{
26 Messages: []provider.Message{{Role: provider.RoleUser, Content: "hi"}},
27 MaxTokens: 16,
28 })
29 if err != nil {
30 t.Fatalf("Stream: %v", err)
31 }
32 return out, fc.openedParams(t).StreamID
33 }
34
35 func textChunk(text string) protocol.ProviderChunk {
36 return protocol.ProviderChunk{Type: protocol.ChunkText, Text: text}
37 }
38
39 // collectChunks drains the channel until it closes, failing on a wedge.
40 func collectChunks(t *testing.T, out <-chan provider.Chunk) []provider.Chunk {
41 t.Helper()
42 var chunks []provider.Chunk
43 for {
44 select {
45 case chunk, ok := <-out:
46 if !ok {
47 return chunks
48 }
49 chunks = append(chunks, chunk)
50 case <-time.After(testBudget):
51 t.Fatal("stream channel did not close")
52 }
53 }
54 }
55
56 func texts(chunks []provider.Chunk) []string {
57 var out []string
58 for _, c := range chunks {
59 out = append(out, c.Text)
60 }
61 return out
62 }
63
63 lines GO