x/internal/net/pipe_test.go

package net

import (
	"context"
	"errors"
	"io"
	"net"
	"testing"
	"time"
)

// pipeConn is a fully controllable in-memory connection for testing Pipe.
// It supports proper half-close semantics: CloseRead only stops reads,
// CloseWrite only stops writes, and Close stops both.
type pipeConn struct {
	readBuf     []byte
	readPos     int
	writeBuf    []byte
	readClosed  bool
	writeClosed bool
	closedCh    chan struct{}
	readDead    time.Time
	writeDead   time.Time
	readErr     error // error to return on next Read
	eofAfter    int   // return io.EOF after this many bytes total read (-1 = never)
	bytesRead   int
}

func newPipeConn(data []byte) *pipeConn {
	return &pipeConn{
		readBuf:  data,
		closedCh: make(chan struct{}),
	}
}

func (pc *pipeConn) Read(p []byte) (int, error) {
	select {
	case <-pc.closedCh:
		return 0, net.ErrClosed
	default:
	}
	if pc.readClosed {
		return 0, net.ErrClosed
	}
	if pc.readErr != nil {
		err := pc.readErr
		pc.readErr = nil
		return 0, err
	}
	if pc.readPos >= len(pc.readBuf) {
		// eofAfter=-1 means sleep briefly then return (0, nil) to simulate
		// a connection that's still open but has no data yet.
		// This avoids busy-looping while still preventing EOF.
		if pc.eofAfter < 0 {
			time.Sleep(10 * time.Millisecond)
			return 0, nil // no data, no error, no EOF
		}
		return 0, io.EOF
	}
	n := copy(p, pc.readBuf[pc.readPos:])
	pc.readPos += n
	pc.bytesRead += n
	return n, nil
}

func (pc *pipeConn) Write(p []byte) (int, error) {
	select {
	case <-pc.closedCh:
		return 0, net.ErrClosed
	default:
	}
	if pc.writeClosed {
		return 0, net.ErrClosed
	}
	pc.writeBuf = append(pc.writeBuf, p...)
	return len(p), nil
}

func (pc *pipeConn) Close() error {
	select {
	case <-pc.closedCh:
	default:
		close(pc.closedCh)
	}
	return nil
}

func (pc *pipeConn) CloseRead() error {
	pc.readClosed = true
	return nil
}

func (pc *pipeConn) CloseWrite() error {
	pc.writeClosed = true
	return nil
}

func (pc *pipeConn) SetReadDeadline(t time.Time) error {
	pc.readDead = t
	return nil
}

func (pc *pipeConn) SetWriteDeadline(t time.Time) error {
	pc.writeDead = t
	return nil
}

func TestPipe_NormalTransfer(t *testing.T) {
	ctx := context.Background()
	c1 := newPipeConn([]byte("hello from c1"))
	c2 := newPipeConn([]byte("hello from c2"))

	err := Pipe(ctx, c1, c2)
	if err != nil {
		t.Errorf("expected nil error, got %v", err)
	}

	// c2's writeBuf has data from c1
	if string(c2.writeBuf) != "hello from c1" {
		t.Errorf("c2.writeBuf = %q, want %q", string(c2.writeBuf), "hello from c1")
	}
	// c1's writeBuf has data from c2
	if string(c1.writeBuf) != "hello from c2" {
		t.Errorf("c1.writeBuf = %q, want %q", string(c1.writeBuf), "hello from c2")
	}
}

func TestPipe_ImmediateEOF(t *testing.T) {
	ctx := context.Background()
	// Empty read buffers produce immediate EOF, simulating closed connections
	c1 := newPipeConn(nil)
	c2 := newPipeConn(nil)

	err := Pipe(ctx, c1, c2)
	if err != nil {
		t.Errorf("expected nil error (EOF is not an error from Pipe), got %v", err)
	}
}

func TestPipe_ContextCancel(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())

	// These never produce EOF and never error, so Pipe will run until canceled
	c1 := newPipeConn(nil)
	c2 := newPipeConn(nil)
	c1.eofAfter = -1 // disable EOF
	c2.eofAfter = -1

	// Start cancellation in a goroutine
	go func() {
		time.Sleep(50 * time.Millisecond)
		cancel()
	}()

	err := Pipe(ctx, c1, c2)
	if err == nil {
		t.Error("expected context error, got nil")
	}
}

func TestPipe_ContextAlreadyCanceled(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	c1 := newPipeConn(nil)
	c1.eofAfter = -1
	c2 := newPipeConn(nil)
	c2.eofAfter = -1

	err := Pipe(ctx, c1, c2)
	if err == nil {
		t.Error("expected context error, got nil")
	}
}

func TestPipe_ReadError(t *testing.T) {
	ctx := context.Background()
	readErr := errors.New("read failure")
	c1 := newPipeConn(nil)
	c1.readErr = readErr
	// c2 also has a read error so both goroutines terminate
	otherErr := errors.New("other error")
	c2 := newPipeConn(nil)
	c2.readErr = otherErr

	err := Pipe(ctx, c1, c2)
	if err == nil {
		t.Error("expected error, got nil")
	}
}

func TestPipe_WriteError(t *testing.T) {
	ctx := context.Background()
	c1 := newPipeConn([]byte("data"))
	// c2 has its write side pre-closed
	c2 := newPipeConn(nil)
	c2.writeClosed = true

	err := Pipe(ctx, c1, c2)
	if err == nil {
		t.Error("expected write error, got nil")
	}
}

func Test_readDeadliner(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())

	r := &mockReadWriteCloser{data: []byte("testdata")}
	rd := &readDeadliner{Reader: r, ctx: ctx}

	// Normal read
	buf := make([]byte, 8)
	n, err := rd.Read(buf)
	if err != nil && err != io.EOF {
		t.Fatal(err)
	}
	if n != 8 || string(buf[:n]) != "testdata" {
		t.Errorf("expected 'testdata', got %q", string(buf[:n]))
	}

	// Context canceled
	cancel()
	_, err = rd.Read(buf)
	if err == nil {
		t.Error("expected context error, got nil")
	}
}

func Test_halfClose(t *testing.T) {
	// pipeConn supports CloseRead and CloseWrite (success).
	// CloseWrite returns nil (successful half-close), which sets a read deadline on dst.
	src := newPipeConn(nil)
	dst := newPipeConn(nil)
	halfClose(src, dst)
	if !src.readClosed {
		t.Error("src should have CloseRead called")
	}
	if !dst.writeClosed {
		t.Error("dst should have CloseWrite called (success path)")
	}
	if dst.readDead.IsZero() {
		t.Error("dst should have read deadline set after successful CloseWrite")
	}
}

func Test_halfClose_MockReadWriteCloser(t *testing.T) {
	// mockReadWriteCloser has CloseRead, but CloseWrite returns ErrUnsupported -> dst.Close()
	src2 := &mockReadWriteCloser{}
	dst2 := &mockReadWriteCloser{}
	halfClose(src2, dst2)
	if !src2.readClosed {
		t.Error("src should have CloseRead called")
	}
	if !dst2.closed {
		t.Error("dst should be fully closed when CloseWrite is ErrUnsupported")
	}
}

func Test_forceClose(t *testing.T) {
	c1 := newPipeConn(nil)
	c2 := newPipeConn(nil)
	forceClose(c1, c2, nil)
	// Check that closedCh is closed
	select {
	case <-c1.closedCh:
	default:
		t.Error("c1 should be closed")
	}
	select {
	case <-c2.closedCh:
	default:
		t.Error("c2 should be closed")
	}
}

func Test_pipeHalf(t *testing.T) {
	ctx := context.Background()
	src := newPipeConn([]byte("hello world"))
	dst := newPipeConn(nil)

	err := pipeHalf(ctx, src, dst, 0)
	if err != nil {
		t.Fatal(err)
	}
	if string(dst.writeBuf) != "hello world" {
		t.Errorf("dst.writeBuf = %q, want %q", string(dst.writeBuf), "hello world")
	}
}

func Test_pipeHalf_ReadError(t *testing.T) {
	ctx := context.Background()
	readErr := errors.New("read failed")
	src := newPipeConn(nil)
	src.readErr = readErr
	dst := newPipeConn(nil)

	err := pipeHalf(ctx, src, dst, 0)
	if err == nil {
		t.Error("expected error, got nil")
	}
}

func Test_pipeHalf_WriteError(t *testing.T) {
	ctx := context.Background()
	src := newPipeConn([]byte("data"))
	dst := newPipeConn(nil)
	dst.writeClosed = true

	err := pipeHalf(ctx, src, dst, 0)
	if err == nil {
		t.Error("expected error, got nil")
	}
}

func Test_pipeHalf_ContextCanceled(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	src := newPipeConn([]byte("data"))
	src.eofAfter = -1
	dst := newPipeConn(nil)

	err := pipeHalf(ctx, src, dst, 0)
	if err == nil {
		t.Error("expected context error, got nil")
	}
}

func TestPipe_CtxDoneWithError(t *testing.T) {
	// Test the ctx.Done() path where firstErr is non-nil
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	c1 := newPipeConn(nil)
	c1.readErr = errors.New("error before cancel")
	c1.eofAfter = -1
	c2 := newPipeConn(nil)
	c2.eofAfter = -1

	done := make(chan error, 1)
	go func() {
		done <- Pipe(ctx, c1, c2)
	}()

	time.Sleep(20 * time.Millisecond)
	cancel()

	err := <-done
	if err == nil {
		t.Error("expected error, got nil")
	}
}

func TestPipe_CtxDoneWithoutError(t *testing.T) {
	// Standard ctx.Done path with no prior error → returns ctx.Err()
	ctx, cancel := context.WithCancel(context.Background())

	c1 := newPipeConn(nil)
	c1.eofAfter = -1
	c2 := newPipeConn(nil)
	c2.eofAfter = -1

	go func() {
		time.Sleep(30 * time.Millisecond)
		cancel()
	}()

	err := Pipe(ctx, c1, c2)
	if err == nil {
		t.Error("expected context error, got nil")
	}
}

func TestTransport_ReturnsError(t *testing.T) {
	// Both directions must return errors since Transport reads one from errc
	readErr := errors.New("not EOF error")
	rw1 := &mockReadWriter{readErr: readErr}
	rw2 := &mockReadWriter{readErr: errors.New("other error")}

	err := Transport(rw1, rw2)
	if err == nil {
		t.Error("expected error, got nil")
	}
}

func Test_halfClose_NoCloseWriteInterface(t *testing.T) {
	// Test the path where dst doesn't implement CloseWrite
	type noCloseWrite struct {
		*mockReadWriteCloser
	}

	src := &mockReadWriteCloser{}
	// Create a type that doesn't have CloseWrite (use a type without the method)
	// Actually, we need a type that implements io.ReadWriteCloser but NOT CloseWrite
	dst := &noCloseWriteCloser{}
	halfClose(src, dst)
	if !dst.closed {
		// halfClose checks cw, ok := dst.(xio.CloseWrite) → false
		// falls through to dst.Close()
		t.Error("dst should be closed when CloseWrite is not available")
	}
}

type noCloseWriteCloser struct {
	closed bool
}

func (c *noCloseWriteCloser) Read(p []byte) (int, error)  { return 0, nil }
func (c *noCloseWriteCloser) Write(p []byte) (int, error) { return len(p), nil }
func (c *noCloseWriteCloser) Close() error                { c.closed = true; return nil }