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

   1 package m31
   2
   3 import (
   4         "context"
   5         "errors"
   6         "testing"
   7         "testing/synctest"
   8         "time"
   9
  10         "github.com/stretchr/testify/require"
  11
  12         "onehouse/protocol"
  13 )
  14
  15 func TestNewRejectsInvalidAssemblySettings(t *testing.T) {
  16         testCases := []struct {
  17                 name   string
  18                 change func(*Settings)
  19         }{
  20                 {name: "rejects noncanonical ID", change: func(s *Settings) {
  21                         s.DeviceID = "bad"
  22                 }},
  23                 {name: "requires host and port", change: func(s *Settings) {
  24                         s.Address = "example.com"
  25                 }},
  26                 {name: "requires a host", change: func(s *Settings) {
  27                         s.Address = ":502"
  28                 }},
  29                 {name: "requires numeric port", change: func(s *Settings) {
  30                         s.Address = "example.com:bad"
  31                 }},
  32                 {name: "rejects port zero", change: func(s *Settings) {
  33                         s.Address = "example.com:0"
  34                 }},
  35                 {name: "rejects excess port", change: func(s *Settings) {
  36                         s.Address = "example.com:65536"
  37                 }},
  38                 {name: "rejects broadcast unit", change: func(s *Settings) {
  39                         s.Unit = 0
  40                 }},
  41                 {name: "rejects reserved unit", change: func(s *Settings) {
  42                         s.Unit = 248
  43                 }},
  44                 {name: "rejects fast polling", change: func(s *Settings) {
  45                         s.PollInterval = time.Millisecond
  46                 }},
  47                 {name: "rejects slow polling", change: func(s *Settings) {
  48                         s.PollInterval = time.Minute + time.Nanosecond
  49                 }},
  50                 {name: "requires valid modules", change: func(s *Settings) {
  51                         s.Modules = nil
  52                 }},
  53         }
  54
  55         for _, tc := range testCases {
  56                 t.Run(tc.name, func(t *testing.T) {
  57                         d, client, _ := testAssembly(t)
  58                         settings := Settings{
  59                                 DeviceID:     testDeviceID,
  60                                 Address:      "example.com:502",
  61                                 Unit:         1,
  62                                 PollInterval: 50 * time.Millisecond,
  63                                 Modules:      []Module{{Model: HostModel}},
  64                         }
  65                         tc.change(&settings)
  66
  67                         _, err := New(settings, d.publisher, client)
  68                         require.Error(t, err)
  69                 })
  70         }
  71 }
  72
  73 func TestDriverReconnectsWithConfigurationAndFreshBaseline(t *testing.T) {
  74         synctest.Test(t, func(t *testing.T) {
  75                 d, client, transport := testAssembly(t)
  76                 client.connectErrors = []error{errors.New("unavailable")}
  77                 client.inputs[1] = true
  78                 ctx, cancel := context.WithCancel(t.Context())
  79                 defer cancel()
  80
  81                 done := make(chan error, 1)
  82
  83                 go func() {
  84                         done <- d.Run(ctx)
  85                 }()
  86
  87                 synctest.Wait()
  88                 client.withLock(func(client *fakeClient) {
  89                         require.Len(t, client.attempts, 1)
  90                         require.Empty(t, client.registerWrites)
  91                 })
  92                 require.Empty(t, transport.ofKind(protocol.Event))
  93
  94                 time.Sleep(time.Second)
  95                 synctest.Wait()
  96                 client.withLock(func(client *fakeClient) {
  97                         require.Len(t, client.attempts, 2)
  98                         require.Len(t, client.registerWrites, 33)
  99                         require.EqualValues(t, 200, client.registers[pulseWidthBase+2])
 100                         require.EqualValues(t, 1, client.registers[outputModeBase+2])
 101                 })
 102                 require.Empty(t, transport.ofKind(protocol.Event))
 103
 104                 client.withLock(func(client *fakeClient) {
 105                         client.inputs[1] = false
 106                 })
 107
 108                 time.Sleep(50 * time.Millisecond)
 109                 synctest.Wait()
 110                 require.Len(t, transport.ofKind(protocol.Event), 1)
 111
 112                 client.withLock(func(client *fakeClient) {
 113                         client.inputErr = errors.New("connection lost")
 114                 })
 115
 116                 time.Sleep(50 * time.Millisecond)
 117                 synctest.Wait()
 118
 119                 availability, valid := transport.ofKind(protocol.Availability)[2].Payload.(protocol.AvailabilityPayload)
 120                 require.True(t, valid)
 121                 require.Equal(t, "offline", availability.Status)
 122
 123                 client.withLock(func(client *fakeClient) {
 124                         client.inputErr = nil
 125                         client.inputs[1] = true
 126                 })
 127
 128                 time.Sleep(time.Second)
 129                 synctest.Wait()
 130                 client.withLock(func(client *fakeClient) {
 131                         require.Len(t, client.attempts, 3)
 132                         require.Len(t, client.registerWrites, 66)
 133                 })
 134                 require.Len(t, transport.ofKind(protocol.Event), 1)
 135                 cancel()
 136                 require.ErrorIs(t, <-done, context.Canceled)
 137                 client.withLock(func(client *fakeClient) {
 138                         require.Equal(t, 3, client.closes)
 139                 })
 140         })
 141 }
 142
 143 func TestDriverBoundsReconnectBackoff(t *testing.T) {
 144         synctest.Test(t, func(t *testing.T) {
 145                 d, client, _ := testAssembly(t)
 146                 for range 8 {
 147                         client.withLock(func(client *fakeClient) {
 148                                 client.connectErrors = append(client.connectErrors, errors.New("offline"))
 149                         })
 150                 }
 151
 152                 ctx, cancel := context.WithCancel(t.Context())
 153                 defer cancel()
 154
 155                 done := make(chan error, 1)
 156
 157                 go func() {
 158                         done <- d.Run(ctx)
 159                 }()
 160
 161                 synctest.Wait()
 162
 163                 for _, delay := range []time.Duration{1, 2, 4, 8, 16, 30, 30} {
 164                         time.Sleep(delay * time.Second)
 165                         synctest.Wait()
 166                 }
 167
 168                 client.withLock(func(client *fakeClient) {
 169                         require.Len(t, client.attempts, 8)
 170                         require.Equal(t, 30*time.Second, client.attempts[7].Sub(client.attempts[6]))
 171                 })
 172                 cancel()
 173                 require.ErrorIs(t, <-done, context.Canceled)
 174         })
 175 }
 176
 177 func TestDriverReturnsPublicationAndCleanupFailures(t *testing.T) {
 178         testCases := []struct {
 179                 name         string
 180                 kind         protocol.Kind
 181                 closeFailure bool
 182         }{
 183                 {name: "returns descriptor failure", kind: protocol.Descriptor},
 184                 {name: "returns online failure", kind: protocol.Availability},
 185                 {name: "returns baseline failure", kind: protocol.Reported},
 186                 {name: "returns cleanup failure", closeFailure: true},
 187         }
 188
 189         for _, tc := range testCases {
 190                 t.Run(tc.name, func(t *testing.T) {
 191                         d, client, transport := testAssembly(t)
 192                         failure := errors.New("independent failure")
 193                         if tc.closeFailure {
 194                                 client.closeErr = failure
 195                                 client.connectErrors = []error{errors.New("device unavailable")}
 196                         } else {
 197                                 transport.failKind, transport.err = tc.kind, failure
 198                         }
 199
 200                         require.ErrorIs(t, d.Run(t.Context()), failure)
 201                         require.LessOrEqual(t, len(client.attempts), 1)
 202                 })
 203         }
 204 }
 205
 206 func TestDriverRefreshDoesNotRepeatCommandsOrEvents(t *testing.T) {
 207         synctest.Test(t, func(t *testing.T) {
 208                 d, client, transport := testAssembly(t)
 209                 ctx, cancel := context.WithCancel(t.Context())
 210                 defer cancel()
 211
 212                 done := make(chan error, 2)
 213
 214                 go func() {
 215                         done <- d.Run(ctx)
 216                 }()
 217                 go func() {
 218                         done <- d.publisher.Run(ctx)
 219                 }()
 220
 221                 synctest.Wait()
 222
 223                 count := len(transport.all())
 224
 225                 d.Refresh()
 226                 synctest.Wait()
 227                 require.Greater(t, len(transport.all()), count)
 228                 require.Empty(t, client.writes)
 229                 require.Empty(t, transport.ofKind(protocol.Event))
 230                 cancel()
 231                 require.ErrorIs(t, <-done, context.Canceled)
 232                 require.ErrorIs(t, <-done, context.Canceled)
 233         })
 234 }