git.giacomolli.uk - onehouse/blob - driver/m31/tcp_client_test.go

   1 package m31
   2
   3 import (
   4         "context"
   5         "encoding/binary"
   6         "errors"
   7         "fmt"
   8         "io"
   9         "net"
  10         "os"
  11         "testing"
  12         "testing/synctest"
  13         "time"
  14
  15         "github.com/stretchr/testify/require"
  16
  17         "onehouse/driver"
  18 )
  19
  20 func TestTCPClientEncodesAndChecksModbusOperations(t *testing.T) {
  21         testCases := []struct {
  22                 name    string
  23                 call    func(*tcpClient) error
  24                 request []byte
  25                 reply   []byte
  26         }{
  27                 {
  28                         name: "reads low bit first across byte boundaries",
  29                         call: func(c *tcpClient) error {
  30                                 value, err := c.ReadInputs(t.Context(), 9)
  31                                 require.Equal(t, []bool{true, false, true, false, false, false, false, false, true}, value)
  32
  33                                 return err
  34                         },
  35                         request: []byte{2, 0, 0, 0, 9},
  36                         reply:   []byte{2, 2, 5, 1},
  37                 },
  38                 {
  39                         name: "reads coil state",
  40                         call: func(c *tcpClient) error {
  41                                 value, err := c.ReadCoils(t.Context(), 1)
  42                                 require.Equal(t, []bool{true}, value)
  43
  44                                 return err
  45                         },
  46                         request: []byte{1, 0, 0, 0, 1},
  47                         reply:   []byte{1, 1, 1},
  48                 },
  49                 {
  50                         name: "reads full sixteen bit registers",
  51                         call: func(c *tcpClient) error {
  52                                 value, err := c.ReadRegisters(t.Context(), 8, 2)
  53                                 require.Equal(t, []uint16{10000, 65535}, value)
  54
  55                                 return err
  56                         },
  57                         request: []byte{3, 0, 8, 0, 2},
  58                         reply:   []byte{3, 4, 0x27, 0x10, 0xff, 0xff},
  59                 },
  60                 {
  61                         name: "writes true with function five",
  62                         call: func(c *tcpClient) error {
  63                                 return c.WriteCoil(t.Context(), 7, true)
  64                         },
  65                         request: []byte{5, 0, 7, 0xff, 0},
  66                         reply:   []byte{5, 0, 7, 0xff, 0},
  67                 },
  68                 {
  69                         name: "writes false with function five",
  70                         call: func(c *tcpClient) error {
  71                                 return c.WriteCoil(t.Context(), 7, false)
  72                         },
  73                         request: []byte{5, 0, 7, 0, 0},
  74                         reply:   []byte{5, 0, 7, 0, 0},
  75                 },
  76                 {
  77                         name: "writes a packed contiguous coil range",
  78                         call: func(c *tcpClient) error {
  79                                 return c.WriteCoils(t.Context(), 6, []bool{true, false, true, false, false, false, false, false, true})
  80                         },
  81                         request: []byte{15, 0, 6, 0, 9, 2, 5, 1},
  82                         reply:   []byte{15, 0, 6, 0, 9},
  83                 },
  84                 {
  85                         name: "writes a holding register with function sixteen",
  86                         call: func(c *tcpClient) error {
  87                                 return c.WriteRegister(t.Context(), 0x30d4, 200)
  88                         },
  89                         request: []byte{16, 0x30, 0xd4, 0, 1, 2, 0, 200},
  90                         reply:   []byte{16, 0x30, 0xd4, 0, 1},
  91                 },
  92         }
  93
  94         for _, tc := range testCases {
  95                 t.Run(tc.name, func(t *testing.T) {
  96                         request, err := wireCall(t, tc.call, func(header []byte) []byte {
  97                                 return append(append(header[:4:4], words(uint16(len(tc.reply)+1))...), append([]byte{7}, tc.reply...)...)
  98                         })
  99                         require.NoError(t, err)
 100                         require.Equal(t, append([]byte{0, 1, 0, 0, 0, byte(len(tc.request) + 1), 7}, tc.request...), request)
 101                 })
 102         }
 103 }
 104
 105 func TestTCPClientRejectsInvalidAndTruncatedResponses(t *testing.T) {
 106         testCases := []struct {
 107                 name  string
 108                 reply []byte
 109         }{
 110                 {name: "rejects another transaction", reply: []byte{0, 2, 0, 0, 0, 4, 7, 1, 1, 0}},
 111                 {name: "rejects another protocol", reply: []byte{0, 1, 0, 1, 0, 4, 7, 1, 1, 0}},
 112                 {name: "rejects another unit", reply: []byte{0, 1, 0, 0, 0, 4, 8, 1, 1, 0}},
 113                 {name: "rejects empty PDU", reply: []byte{0, 1, 0, 0, 0, 1, 7}},
 114                 {name: "rejects excessive PDU", reply: []byte{0, 1, 0, 0, 0, 255, 7}},
 115                 {name: "rejects truncated header", reply: []byte{0, 1}},
 116                 {name: "rejects truncated body", reply: []byte{0, 1, 0, 0, 0, 4, 7, 1}},
 117                 {name: "rejects wrong function", reply: []byte{0, 1, 0, 0, 0, 4, 7, 2, 1, 0}},
 118                 {name: "rejects exception", reply: []byte{0, 1, 0, 0, 0, 3, 7, 0x81, 2}},
 119                 {name: "rejects wrong byte count", reply: []byte{0, 1, 0, 0, 0, 4, 7, 1, 2, 0}},
 120         }
 121
 122         for _, tc := range testCases {
 123                 t.Run(tc.name, func(t *testing.T) {
 124                         _, err := wireCall(t, func(c *tcpClient) error {
 125                                 _, err := c.ReadCoils(t.Context(), 1)
 126                                 return err
 127                         }, func([]byte) []byte {
 128                                 return tc.reply
 129                         })
 130                         require.Error(t, err)
 131                 })
 132         }
 133 }
 134
 135 func TestTCPClientRejectsInvalidRangesAndAcknowledgements(t *testing.T) {
 136         c := &tcpClient{}
 137         _, err := c.ReadInputs(t.Context(), 0)
 138         require.Error(t, err)
 139
 140         _, err = c.ReadCoils(t.Context(), 2001)
 141         require.Error(t, err)
 142
 143         _, err = c.ReadRegisters(t.Context(), 0, 0)
 144         require.Error(t, err)
 145
 146         _, err = c.ReadRegisters(t.Context(), 0, 126)
 147         require.Error(t, err)
 148
 149         _, err = c.ReadRegisters(t.Context(), 65535, 2)
 150         require.Error(t, err)
 151         require.Error(t, c.WriteCoils(t.Context(), 0, nil))
 152         require.Error(t, c.WriteCoils(t.Context(), 65535, []bool{true, true}))
 153         require.Error(t, c.WriteCoils(t.Context(), 0, make([]bool, 1969)))
 154         require.Error(t, c.WriteCoil(t.Context(), 0, true))
 155         require.NoError(t, c.Close())
 156         require.Error(t, c.Connect(t.Context(), "", 0))
 157         require.Error(t, c.Connect(t.Context(), "", 248))
 158         require.Error(t, c.Connect(t.Context(), "", 1))
 159
 160         _, err = wireCall(t, func(c *tcpClient) error {
 161                 return c.WriteRegister(t.Context(), 0, 20)
 162         }, func(header []byte) []byte {
 163                 return append(header[:7:7], 16, 0, 0, 0, 2)
 164         })
 165         require.Error(t, err)
 166
 167         _, err = wireCall(t, func(c *tcpClient) error {
 168                 _, readErr := c.ReadRegisters(t.Context(), 0, 1)
 169                 return readErr
 170         }, func(header []byte) []byte {
 171                 return append(header[:7:7], 3, 3, 0, 0, 0)
 172         })
 173         require.Error(t, err)
 174 }
 175
 176 func TestTCPClientCancellationAndTimeouts(t *testing.T) {
 177         testCases := []struct {
 178                 name   string
 179                 cancel bool
 180                 write  bool
 181         }{
 182                 {name: "cancels a blocked write", cancel: true, write: true},
 183                 {name: "cancels a blocked read", cancel: true},
 184                 {name: "limits a blocked write", write: true},
 185                 {name: "limits a blocked read"},
 186         }
 187
 188         for _, tc := range testCases {
 189                 t.Run(tc.name, func(t *testing.T) {
 190                         synctest.Test(t, func(t *testing.T) {
 191                                 conn, peer := net.Pipe()
 192                                 c := &tcpClient{conn: conn, unit: 1}
 193                                 t.Cleanup(func() {
 194                                         require.NoError(t, c.Close())
 195                                         require.NoError(t, peer.Close())
 196                                 })
 197
 198                                 ctx, cancel := context.WithCancel(t.Context())
 199                                 defer cancel()
 200
 201                                 result := make(chan error, 1)
 202
 203                                 if !tc.write {
 204                                         go func() {
 205                                                 var request [12]byte
 206
 207                                                 _, err := io.ReadFull(peer, request[:])
 208                                                 result <- err
 209                                         }()
 210                                 }
 211
 212                                 done := make(chan error, 1)
 213                                 go func() {
 214                                         _, err := c.ReadCoils(ctx, 1)
 215                                         done <- err
 216                                 }()
 217
 218                                 synctest.Wait()
 219
 220                                 if !tc.write {
 221                                         require.NoError(t, <-result)
 222                                 }
 223
 224                                 if tc.cancel {
 225                                         cancel()
 226                                 } else {
 227                                         time.Sleep(2 * time.Second)
 228                                 }
 229
 230                                 err := <-done
 231                                 require.Error(t, err)
 232
 233                                 if tc.cancel {
 234                                         require.ErrorIs(t, err, context.Canceled)
 235                                 }
 236                         })
 237                 })
 238         }
 239 }
 240
 241 func TestTCPClientCancellationIsAnExpectedRunnerStop(t *testing.T) {
 242         testCases := []struct {
 243                 name    string
 244                 reading bool
 245         }{
 246                 {name: "stops a blocked write"},
 247                 {name: "stops a blocked read", reading: true},
 248         }
 249
 250         for _, tc := range testCases {
 251                 t.Run(tc.name, func(t *testing.T) {
 252                         synctest.Test(t, func(t *testing.T) {
 253                                 conn, peer := net.Pipe()
 254                                 c := &tcpClient{conn: conn, unit: 1}
 255
 256                                 t.Cleanup(func() {
 257                                         require.NoError(t, peer.Close())
 258                                 })
 259
 260                                 ctx, cancel := context.WithCancel(t.Context())
 261                                 defer cancel()
 262
 263                                 ready := make(chan error, 1)
 264                                 if tc.reading {
 265                                         go func() {
 266                                                 var request [12]byte
 267
 268                                                 _, err := io.ReadFull(peer, request[:])
 269                                                 ready <- err
 270                                         }()
 271                                 }
 272
 273                                 done := make(chan error, 1)
 274                                 go func() {
 275                                         done <- driver.Run(ctx, []driver.Driver{&requestDriver{client: c}}, nil)
 276                                 }()
 277
 278                                 synctest.Wait()
 279
 280                                 if tc.reading {
 281                                         require.NoError(t, <-ready)
 282                                 }
 283
 284                                 cancel()
 285                                 require.NoError(t, <-done)
 286                         })
 287                 })
 288         }
 289 }
 290
 291 func TestTCPClientUsesCallerDeadlineAndStartsClosed(t *testing.T) {
 292         client := NewTCPClient()
 293         _, err := client.ReadCoils(t.Context(), 1)
 294         require.Error(t, err)
 295         require.NoError(t, client.Close())
 296         synctest.Test(t, func(t *testing.T) {
 297                 conn, peer := net.Pipe()
 298                 c := &tcpClient{conn: conn, unit: 1}
 299                 t.Cleanup(func() {
 300                         require.NoError(t, c.Close())
 301                         require.NoError(t, peer.Close())
 302                 })
 303
 304                 ctx, cancel := context.WithTimeout(t.Context(), time.Millisecond)
 305                 defer cancel()
 306
 307                 started := time.Now()
 308                 _, err := c.ReadCoils(ctx, 1)
 309                 require.Error(t, err)
 310                 require.Equal(t, time.Millisecond, time.Since(started))
 311         })
 312 }
 313
 314 func TestTCPClientReadTimeoutStartsAfterRequestWrite(t *testing.T) {
 315         synctest.Test(t, func(t *testing.T) {
 316                 conn, peer := net.Pipe()
 317                 c := &tcpClient{conn: conn, unit: 1}
 318                 t.Cleanup(func() {
 319                         require.NoError(t, c.Close())
 320                         require.NoError(t, peer.Close())
 321                 })
 322
 323                 done := make(chan error, 1)
 324                 go func() {
 325                         time.Sleep(500 * time.Millisecond)
 326
 327                         var request [12]byte
 328
 329                         _, err := io.ReadFull(peer, request[:])
 330                         done <- err
 331                 }()
 332
 333                 started := time.Now()
 334                 _, err := c.ReadCoils(t.Context(), 1)
 335                 require.Error(t, err)
 336                 require.NoError(t, <-done)
 337                 require.Equal(t, 2500*time.Millisecond, time.Since(started))
 338         })
 339 }
 340
 341 func TestTCPClientStopsAfterCancelledRequestWrite(t *testing.T) {
 342         synctest.Test(t, func(t *testing.T) {
 343                 conn, peer := net.Pipe()
 344                 t.Cleanup(func() {
 345                         require.NoError(t, peer.Close())
 346                 })
 347
 348                 ctx, cancel := context.WithCancel(t.Context())
 349                 defer cancel()
 350
 351                 transport := &interruptConn{
 352                         Conn:        conn,
 353                         cancel:      cancel,
 354                         interrupted: make(chan struct{}),
 355                 }
 356                 c := &tcpClient{conn: transport, unit: 1}
 357                 t.Cleanup(func() {
 358                         require.NoError(t, c.Close())
 359                 })
 360
 361                 _, err := c.ReadCoils(ctx, 1)
 362                 require.ErrorIs(t, err, context.Canceled)
 363                 require.NotErrorIs(t, err, os.ErrDeadlineExceeded)
 364         })
 365 }
 366
 367 func TestTCPClientCancellationPreservesIndependentFailure(t *testing.T) {
 368         synctest.Test(t, func(t *testing.T) {
 369                 conn, peer := net.Pipe()
 370                 t.Cleanup(func() {
 371                         require.NoError(t, peer.Close())
 372                 })
 373
 374                 ctx, cancel := context.WithCancel(t.Context())
 375                 defer cancel()
 376
 377                 failure := errors.New("independent write failure")
 378                 transport := &interruptConn{
 379                         Conn:        conn,
 380                         cancel:      cancel,
 381                         failure:     errors.Join(os.ErrDeadlineExceeded, failure),
 382                         interrupted: make(chan struct{}),
 383                 }
 384                 c := &tcpClient{conn: transport, unit: 1}
 385                 t.Cleanup(func() {
 386                         require.NoError(t, c.Close())
 387                 })
 388
 389                 _, err := c.ReadCoils(ctx, 1)
 390                 require.ErrorIs(t, err, failure)
 391                 require.ErrorIs(t, err, context.Canceled)
 392         })
 393 }
 394
 395 type requestDriver struct {
 396         client *tcpClient
 397 }
 398
 399 func (d *requestDriver) Run(ctx context.Context) (err error) {
 400         defer func() {
 401                 err = errors.Join(err, d.client.Close())
 402         }()
 403
 404         _, err = d.client.ReadCoils(ctx, 1)
 405
 406         return err
 407 }
 408
 409 func (d *requestDriver) Refresh() {
 410 }
 411
 412 type interruptConn struct {
 413         net.Conn
 414         cancel      context.CancelFunc
 415         failure     error
 416         interrupted chan struct{}
 417 }
 418
 419 func (c *interruptConn) Write(data []byte) (int, error) {
 420         c.cancel()
 421         <-c.interrupted
 422
 423         if c.failure != nil {
 424                 return 0, c.failure
 425         }
 426
 427         return len(data), nil
 428 }
 429
 430 func (c *interruptConn) SetDeadline(deadline time.Time) error {
 431         defer close(c.interrupted)
 432
 433         err := c.Conn.SetDeadline(deadline)
 434         if err != nil {
 435                 return fmt.Errorf("interrupt fake connection: %w", err)
 436         }
 437
 438         return nil
 439 }
 440
 441 func wireCall(t *testing.T, call func(*tcpClient) error, response func([]byte) []byte) ([]byte, error) {
 442         t.Helper()
 443
 444         conn, peer := net.Pipe()
 445         c := &tcpClient{conn: conn, unit: 7}
 446         t.Cleanup(func() {
 447                 require.NoError(t, c.Close())
 448                 require.NoError(t, peer.Close())
 449         })
 450
 451         requests := make(chan []byte, 1)
 452         done := make(chan error, 1)
 453
 454         go func() {
 455                 var header [7]byte
 456
 457                 _, err := io.ReadFull(peer, header[:])
 458                 if err != nil {
 459                         done <- err
 460                         return
 461                 }
 462
 463                 body := make([]byte, int(binary.BigEndian.Uint16(header[4:6]))-1)
 464                 _, err = io.ReadFull(peer, body)
 465                 requests <- append(header[:7:7], body...)
 466                 if err == nil {
 467                         _, err = peer.Write(response(header[:]))
 468                 }
 469
 470                 if errors.Is(err, io.ErrClosedPipe) {
 471                         err = nil
 472                 }
 473
 474                 err = errors.Join(err, peer.Close())
 475                 done <- err
 476         }()
 477
 478         err := call(c)
 479         require.NoError(t, c.Close())
 480         require.NoError(t, <-done)
 481
 482         return <-requests, err
 483 }