git.giacomolli.uk - onehouse/blob - driver/runner_test.go

   1 package driver
   2
   3 import (
   4         "bytes"
   5         "context"
   6         "errors"
   7         "fmt"
   8         "io"
   9         "log/slog"
  10         "net"
  11         "syscall"
  12         "testing"
  13         "testing/synctest"
  14
  15         "github.com/stretchr/testify/require"
  16 )
  17
  18 func TestRunRefreshesDriversAndWaitsForCleanup(t *testing.T) {
  19         synctest.Test(t, func(t *testing.T) {
  20                 ctx, cancel := context.WithCancel(t.Context())
  21                 defer cancel()
  22
  23                 first := &fakeDriver{
  24                         stopped:   make(chan struct{}),
  25                         refreshed: make(chan struct{}, 1),
  26                 }
  27                 second := &fakeDriver{
  28                         stopped:   make(chan struct{}),
  29                         refreshed: make(chan struct{}, 1),
  30                 }
  31                 connected := make(chan struct{}, 1)
  32                 finished := make(chan error, 1)
  33                 go func() {
  34                         finished <- Run(ctx, []Driver{first, second}, connected)
  35                 }()
  36
  37                 synctest.Wait()
  38
  39                 connected <- struct{}{}
  40
  41                 synctest.Wait()
  42                 require.Len(t, first.refreshed, 1)
  43                 require.Len(t, second.refreshed, 1)
  44                 close(connected)
  45                 synctest.Wait()
  46                 require.Empty(t, finished)
  47                 cancel()
  48                 require.NoError(t, <-finished)
  49                 <-first.stopped
  50                 <-second.stopped
  51         })
  52 }
  53
  54 func TestRunClassifiesStopsAndJoinedFailures(t *testing.T) {
  55         failure := errors.New("readback failed")
  56         testCases := []struct {
  57                 name     string
  58                 err      error
  59                 canceled bool
  60                 failure  bool
  61         }{
  62                 {
  63                         name: "normal completion",
  64                         err:  nil,
  65                 },
  66                 {
  67                         name:     "cancellation is a stop",
  68                         err:      context.Canceled,
  69                         canceled: true,
  70                 },
  71                 {
  72                         name:     "closed transport is a stop",
  73                         err:      net.ErrClosed,
  74                         canceled: true,
  75                 },
  76                 {
  77                         name:     "closed pipe is a stop",
  78                         err:      io.ErrClosedPipe,
  79                         canceled: true,
  80                 },
  81                 {
  82                         name:     "broken pipe is a stop",
  83                         err:      syscall.EPIPE,
  84                         canceled: true,
  85                 },
  86                 {
  87                         name:     "reset connection is a stop",
  88                         err:      syscall.ECONNRESET,
  89                         canceled: true,
  90                 },
  91                 {
  92                         name:     "wrapped stop stays a stop",
  93                         err:      fmt.Errorf("read: %w", net.ErrClosed),
  94                         canceled: true,
  95                 },
  96                 {
  97                         name:     "joined stops stay stops",
  98                         err:      errors.Join(context.Canceled, net.ErrClosed),
  99                         canceled: true,
 100                 },
 101                 {
 102                         name:     "joined failure stays a failure",
 103                         err:      errors.Join(context.Canceled, failure),
 104                         canceled: true,
 105                         failure:  true,
 106                 },
 107                 {
 108                         name:    "request timeout stays a failure",
 109                         err:     context.DeadlineExceeded,
 110                         failure: true,
 111                 },
 112                 {
 113                         name:    "active transport close is a failure",
 114                         err:     net.ErrClosed,
 115                         failure: true,
 116                 },
 117                 {
 118                         name:     "independent failure stays a failure",
 119                         err:      failure,
 120                         canceled: true,
 121                         failure:  true,
 122                 },
 123         }
 124
 125         for _, tc := range testCases {
 126                 t.Run(tc.name, func(t *testing.T) {
 127                         var output bytes.Buffer
 128
 129                         previous := slog.Default()
 130
 131                         slog.SetDefault(slog.New(slog.NewTextHandler(&output, nil)))
 132                         t.Cleanup(func() {
 133                                 slog.SetDefault(previous)
 134                         })
 135
 136                         ctx, cancel := context.WithCancel(t.Context())
 137                         defer cancel()
 138
 139                         if tc.canceled {
 140                                 cancel()
 141                         }
 142
 143                         integration := &resultDriver{err: tc.err}
 144                         err := Run(ctx, []Driver{integration}, nil)
 145                         if tc.failure {
 146                                 require.ErrorIs(t, err, tc.err)
 147                                 require.Contains(t, output.String(), "level=ERROR")
 148                         } else {
 149                                 require.NoError(t, err)
 150                                 require.Empty(t, output.String())
 151                         }
 152                 })
 153         }
 154 }
 155
 156 func TestRunCancelsPeersAndKeepsTheirIndependentFailure(t *testing.T) {
 157         firstFailure := errors.New("first failed")
 158         cleanupFailure := errors.New("cleanup failed")
 159         first := &resultDriver{err: firstFailure}
 160         second := &fakeDriver{
 161                 stopped:   make(chan struct{}),
 162                 refreshed: make(chan struct{}, 1),
 163                 err:       cleanupFailure,
 164         }
 165         err := Run(t.Context(), []Driver{first, second}, nil)
 166         require.ErrorIs(t, err, firstFailure)
 167         require.ErrorIs(t, err, cleanupFailure)
 168         <-second.stopped
 169 }
 170
 171 func TestRunEmptyListWaitsForCancellation(t *testing.T) {
 172         synctest.Test(t, func(t *testing.T) {
 173                 ctx, cancel := context.WithCancel(t.Context())
 174                 defer cancel()
 175
 176                 finished := make(chan error, 1)
 177                 go func() {
 178                         finished <- Run(ctx, nil, nil)
 179                 }()
 180
 181                 synctest.Wait()
 182                 require.Empty(t, finished)
 183                 cancel()
 184                 require.NoError(t, <-finished)
 185         })
 186 }
 187
 188 type fakeDriver struct {
 189         stopped   chan struct{}
 190         refreshed chan struct{}
 191         err       error
 192 }
 193
 194 func (d *fakeDriver) Run(ctx context.Context) error {
 195         <-ctx.Done()
 196
 197         defer close(d.stopped)
 198
 199         return errors.Join(ctx.Err(), d.err)
 200 }
 201
 202 func (d *fakeDriver) Refresh() {
 203         d.refreshed <- struct{}{}
 204 }
 205
 206 type resultDriver struct {
 207         err error
 208 }
 209
 210 func (d *resultDriver) Run(_ context.Context) error {
 211         return d.err
 212 }
 213
 214 func (d *resultDriver) Refresh() {
 215 }