X-Git-Url: https://git.giacomolli.uk/onehouse/blobdiff_plain/af9b6f942b1296e915ff49abae3e0e3ee4068e0f..b9a4f3bdd2609a8b178dd42f313c383ccb59049b:/driver/m31/tcp_client_test.go?ds=inline

diff --git a/driver/m31/tcp_client_test.go b/driver/m31/tcp_client_test.go
index befb688..0067566 100644
--- a/driver/m31/tcp_client_test.go
+++ b/driver/m31/tcp_client_test.go
@@ -4,13 +4,17 @@ import (
 	"context"
 	"encoding/binary"
 	"errors"
+	"fmt"
 	"io"
 	"net"
+	"os"
 	"testing"
 	"testing/synctest"
 	"time"
 
 	"github.com/stretchr/testify/require"
+
+	"onehouse/driver"
 )
 
 func TestTCPClientEncodesAndChecksModbusOperations(t *testing.T) {
@@ -234,6 +238,206 @@ func TestTCPClientCancellationAndTimeouts(t *testing.T) {
 	}
 }
 
+func TestTCPClientCancellationIsAnExpectedRunnerStop(t *testing.T) {
+	testCases := []struct {
+		name    string
+		reading bool
+	}{
+		{name: "stops a blocked write"},
+		{name: "stops a blocked read", reading: true},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				conn, peer := net.Pipe()
+				c := &tcpClient{conn: conn, unit: 1}
+
+				t.Cleanup(func() {
+					require.NoError(t, peer.Close())
+				})
+
+				ctx, cancel := context.WithCancel(t.Context())
+				defer cancel()
+
+				ready := make(chan error, 1)
+				if tc.reading {
+					go func() {
+						var request [12]byte
+
+						_, err := io.ReadFull(peer, request[:])
+						ready <- err
+					}()
+				}
+
+				done := make(chan error, 1)
+				go func() {
+					done <- driver.Run(ctx, []driver.Driver{&requestDriver{client: c}}, nil)
+				}()
+
+				synctest.Wait()
+
+				if tc.reading {
+					require.NoError(t, <-ready)
+				}
+
+				cancel()
+				require.NoError(t, <-done)
+			})
+		})
+	}
+}
+
+func TestTCPClientUsesCallerDeadlineAndStartsClosed(t *testing.T) {
+	client := NewTCPClient()
+	_, err := client.ReadCoils(t.Context(), 1)
+	require.Error(t, err)
+	require.NoError(t, client.Close())
+	synctest.Test(t, func(t *testing.T) {
+		conn, peer := net.Pipe()
+		c := &tcpClient{conn: conn, unit: 1}
+		t.Cleanup(func() {
+			require.NoError(t, c.Close())
+			require.NoError(t, peer.Close())
+		})
+
+		ctx, cancel := context.WithTimeout(t.Context(), time.Millisecond)
+		defer cancel()
+
+		started := time.Now()
+		_, err := c.ReadCoils(ctx, 1)
+		require.Error(t, err)
+		require.Equal(t, time.Millisecond, time.Since(started))
+	})
+}
+
+func TestTCPClientReadTimeoutStartsAfterRequestWrite(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		conn, peer := net.Pipe()
+		c := &tcpClient{conn: conn, unit: 1}
+		t.Cleanup(func() {
+			require.NoError(t, c.Close())
+			require.NoError(t, peer.Close())
+		})
+
+		done := make(chan error, 1)
+		go func() {
+			time.Sleep(500 * time.Millisecond)
+
+			var request [12]byte
+
+			_, err := io.ReadFull(peer, request[:])
+			done <- err
+		}()
+
+		started := time.Now()
+		_, err := c.ReadCoils(t.Context(), 1)
+		require.Error(t, err)
+		require.NoError(t, <-done)
+		require.Equal(t, 2500*time.Millisecond, time.Since(started))
+	})
+}
+
+func TestTCPClientStopsAfterCancelledRequestWrite(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		conn, peer := net.Pipe()
+		t.Cleanup(func() {
+			require.NoError(t, peer.Close())
+		})
+
+		ctx, cancel := context.WithCancel(t.Context())
+		defer cancel()
+
+		transport := &interruptConn{
+			Conn:        conn,
+			cancel:      cancel,
+			interrupted: make(chan struct{}),
+		}
+		c := &tcpClient{conn: transport, unit: 1}
+		t.Cleanup(func() {
+			require.NoError(t, c.Close())
+		})
+
+		_, err := c.ReadCoils(ctx, 1)
+		require.ErrorIs(t, err, context.Canceled)
+		require.NotErrorIs(t, err, os.ErrDeadlineExceeded)
+	})
+}
+
+func TestTCPClientCancellationPreservesIndependentFailure(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		conn, peer := net.Pipe()
+		t.Cleanup(func() {
+			require.NoError(t, peer.Close())
+		})
+
+		ctx, cancel := context.WithCancel(t.Context())
+		defer cancel()
+
+		failure := errors.New("independent write failure")
+		transport := &interruptConn{
+			Conn:        conn,
+			cancel:      cancel,
+			failure:     errors.Join(os.ErrDeadlineExceeded, failure),
+			interrupted: make(chan struct{}),
+		}
+		c := &tcpClient{conn: transport, unit: 1}
+		t.Cleanup(func() {
+			require.NoError(t, c.Close())
+		})
+
+		_, err := c.ReadCoils(ctx, 1)
+		require.ErrorIs(t, err, failure)
+		require.ErrorIs(t, err, context.Canceled)
+	})
+}
+
+type requestDriver struct {
+	client *tcpClient
+}
+
+func (d *requestDriver) Run(ctx context.Context) (err error) {
+	defer func() {
+		err = errors.Join(err, d.client.Close())
+	}()
+
+	_, err = d.client.ReadCoils(ctx, 1)
+
+	return err
+}
+
+func (d *requestDriver) Refresh() {
+}
+
+type interruptConn struct {
+	net.Conn
+	cancel      context.CancelFunc
+	failure     error
+	interrupted chan struct{}
+}
+
+func (c *interruptConn) Write(data []byte) (int, error) {
+	c.cancel()
+	<-c.interrupted
+
+	if c.failure != nil {
+		return 0, c.failure
+	}
+
+	return len(data), nil
+}
+
+func (c *interruptConn) SetDeadline(deadline time.Time) error {
+	defer close(c.interrupted)
+
+	err := c.Conn.SetDeadline(deadline)
+	if err != nil {
+		return fmt.Errorf("interrupt fake connection: %w", err)
+	}
+
+	return nil
+}
+
 func wireCall(t *testing.T, call func(*tcpClient) error, response func([]byte) []byte) ([]byte, error) {
 	t.Helper()