mirror of
https://github.com/zap-proto/zip.git
synced 2026-10-19 15:38:57 +00:00
zeekay
8cd2f7ac34
A socket op's handshake is decided before its handler runs, so a handler with effects runs only for a connection it will get: 426 for another protocol version, 400 without a key, and 403 for a browser on an origin the op does not admit — its own, plus zip.Origins(...). The origin policy is the op's, not the answer's, so Socket carries only Serve, Subprotocols and Compression, and wsx's own refusals are back to the library's. The in-process CLI refuses a socket op before invoking it. An op that answers JSON or bytes keeps JSON's media for its JSON when Produces names the bytes'. 204, 304 and a redirecting op's 3xx carry no body in the document; any other declared status, 3xx included, carries the Out and the doc comment's example, as before. An MCP read of a Reader is bounded in time as a stream is, and the reader closed. TestConnMemory counts the goroutines holding connections added, as it counts the heap they added, so goroutines other tests leave winding down are not charged to the server.
86 lines
2.5 KiB
Go
86 lines
2.5 KiB
Go
// SPDX-License-Identifier: BSD-3-Clause-Eco
|
|
// Copyright (C) 2026, Lux Industries Inc. All rights reserved.
|
|
// See the file LICENSE for licensing terms.
|
|
|
|
package zip
|
|
|
|
import (
|
|
"context"
|
|
"io"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
type tick struct {
|
|
N int `json:"n"`
|
|
}
|
|
|
|
// A tool call collecting a stream that goes quiet answers with what it has
|
|
// when its time is up, and the producer sees its ctx end, so a producer that
|
|
// waits on it stops instead of outliving the call.
|
|
func TestCollect_AQuietStreamEnds(t *testing.T) {
|
|
stopped := make(chan struct{})
|
|
s := &Sse[tick]{Send: func(ctx context.Context, emit func(Event[tick]) error) error {
|
|
if err := emit(Event[tick]{Data: tick{N: 1}}); err != nil {
|
|
return err
|
|
}
|
|
<-ctx.Done()
|
|
close(stopped)
|
|
return ctx.Err()
|
|
}}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
|
|
defer cancel()
|
|
start := time.Now()
|
|
data, media, truncated, err := collect(ctx, s)
|
|
if err != nil || !truncated || media != "application/jsonl" || string(data) != "{\"n\":1}\n" {
|
|
t.Fatalf("collect = %q %q %v %v", data, media, truncated, err)
|
|
}
|
|
if time.Since(start) > 5*time.Second {
|
|
t.Errorf("collect took %v", time.Since(start))
|
|
}
|
|
select {
|
|
case <-stopped:
|
|
case <-time.After(5 * time.Second):
|
|
t.Error("the producer never saw its ctx end")
|
|
}
|
|
}
|
|
|
|
// A producer that ignores its ctx cannot hold the call: collect still answers.
|
|
func TestCollect_AStuckProducerCannotHoldTheCall(t *testing.T) {
|
|
block := make(chan struct{})
|
|
defer close(block)
|
|
s := &Sse[tick]{Send: func(_ context.Context, emit func(Event[tick]) error) error {
|
|
<-block
|
|
return nil
|
|
}}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
|
|
defer cancel()
|
|
done := make(chan struct{})
|
|
go func() {
|
|
_, _, _, _ = collect(ctx, s)
|
|
close(done)
|
|
}()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("collect waited on a producer that never emits")
|
|
}
|
|
}
|
|
|
|
// A reader that never yields cannot hold a tool call: the read is bounded in
|
|
// time as a stream is, and the reader is closed.
|
|
func TestReadBounded_AQuietReaderEnds(t *testing.T) {
|
|
r, w := io.Pipe()
|
|
defer func() { _ = w.Close() }()
|
|
go func() { _, _ = w.Write([]byte("first")) }()
|
|
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
|
|
defer cancel()
|
|
data, truncated, err := readBounded(ctx, r)
|
|
if err != nil || !truncated || string(data) != "first" {
|
|
t.Errorf("readBounded = %q %v %v", data, truncated, err)
|
|
}
|
|
if _, err := w.Write([]byte("x")); err == nil {
|
|
t.Error("the reader was left open")
|
|
}
|
|
}
|