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{stopped: make(chan struct{}), refreshed: make(chan struct{}, 1)}
  24                 second := &fakeDriver{stopped: make(chan struct{}), refreshed: make(chan struct{}, 1)}
  25                 connected := make(chan struct{}, 1)
  26                 finished := make(chan error, 1)
  27                 go func() {
  28                         finished <- Run(ctx, []Driver{first, second}, connected)
  29                 }()
  30
  31                 synctest.Wait()
  32
  33                 connected <- struct{}{}
  34
  35                 synctest.Wait()
  36                 require.Len(t, first.refreshed, 1)
  37                 require.Len(t, second.refreshed, 1)
  38                 close(connected)
  39                 synctest.Wait()
  40                 require.Empty(t, finished)
  41                 cancel()
  42                 require.NoError(t, <-finished)
  43                 <-first.stopped
  44                 <-second.stopped
  45         })
  46 }
  47
  48 func TestRunClassifiesStopsAndJoinedFailures(t *testing.T) {
  49         failure := errors.New("readback failed")
  50         testCases := []struct {
  51                 name     string
  52                 err      error
  53                 canceled bool
  54                 failure  bool
  55         }{
  56                 {name: "normal completion", err: nil},
  57                 {name: "cancellation is a stop", err: context.Canceled, canceled: true},
  58                 {name: "closed transport is a stop", err: net.ErrClosed, canceled: true},
  59                 {name: "closed pipe is a stop", err: io.ErrClosedPipe, canceled: true},
  60                 {name: "broken pipe is a stop", err: syscall.EPIPE, canceled: true},
  61                 {name: "reset connection is a stop", err: syscall.ECONNRESET, canceled: true},
  62                 {name: "wrapped stop stays a stop", err: fmt.Errorf("read: %w", net.ErrClosed), canceled: true},
  63                 {name: "joined stops stay stops", err: errors.Join(context.Canceled, net.ErrClosed), canceled: true},
  64                 {name: "joined failure stays a failure", err: errors.Join(context.Canceled, failure), canceled: true, failure: true},
  65                 {name: "request timeout stays a failure", err: context.DeadlineExceeded, failure: true},
  66                 {name: "active transport close is a failure", err: net.ErrClosed, failure: true},
  67                 {name: "independent failure stays a failure", err: failure, canceled: true, failure: true},
  68         }
  69
  70         for _, tc := range testCases {
  71                 t.Run(tc.name, func(t *testing.T) {
  72                         var output bytes.Buffer
  73
  74                         previous := slog.Default()
  75
  76                         slog.SetDefault(slog.New(slog.NewTextHandler(&output, nil)))
  77                         t.Cleanup(func() {
  78                                 slog.SetDefault(previous)
  79                         })
  80
  81                         ctx, cancel := context.WithCancel(t.Context())
  82                         defer cancel()
  83
  84                         if tc.canceled {
  85                                 cancel()
  86                         }
  87
  88                         integration := &resultDriver{err: tc.err}
  89                         err := Run(ctx, []Driver{integration}, nil)
  90                         if tc.failure {
  91                                 require.ErrorIs(t, err, tc.err)
  92                                 require.Contains(t, output.String(), "level=ERROR")
  93                         } else {
  94                                 require.NoError(t, err)
  95                                 require.Empty(t, output.String())
  96                         }
  97                 })
  98         }
  99 }
 100
 101 func TestRunCancelsPeersAndKeepsTheirIndependentFailure(t *testing.T) {
 102         firstFailure := errors.New("first failed")
 103         cleanupFailure := errors.New("cleanup failed")
 104         first := &resultDriver{err: firstFailure}
 105         second := &fakeDriver{
 106                 stopped:   make(chan struct{}),
 107                 refreshed: make(chan struct{}, 1),
 108                 err:       cleanupFailure,
 109         }
 110         err := Run(t.Context(), []Driver{first, second}, nil)
 111         require.ErrorIs(t, err, firstFailure)
 112         require.ErrorIs(t, err, cleanupFailure)
 113         <-second.stopped
 114 }
 115
 116 func TestRunEmptyListWaitsForCancellation(t *testing.T) {
 117         synctest.Test(t, func(t *testing.T) {
 118                 ctx, cancel := context.WithCancel(t.Context())
 119                 defer cancel()
 120
 121                 finished := make(chan error, 1)
 122                 go func() {
 123                         finished <- Run(ctx, nil, nil)
 124                 }()
 125
 126                 synctest.Wait()
 127                 require.Empty(t, finished)
 128                 cancel()
 129                 require.NoError(t, <-finished)
 130         })
 131 }
 132
 133 type fakeDriver struct {
 134         stopped   chan struct{}
 135         refreshed chan struct{}
 136         err       error
 137 }
 138
 139 func (d *fakeDriver) Run(ctx context.Context) error {
 140         <-ctx.Done()
 141
 142         defer close(d.stopped)
 143
 144         return errors.Join(ctx.Err(), d.err)
 145 }
 146
 147 func (d *fakeDriver) Refresh() {
 148         d.refreshed <- struct{}{}
 149 }
 150
 151 type resultDriver struct {
 152         err error
 153 }
 154
 155 func (d *resultDriver) Run(_ context.Context) error {
 156         return d.err
 157 }
 158
 159 func (d *resultDriver) Refresh() {
 160 }