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