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

   1 package alamo
   2
   3 import (
   4         "context"
   5         "errors"
   6         "strings"
   7         "testing"
   8         "testing/synctest"
   9         "time"
  10
  11         "github.com/stretchr/testify/require"
  12
  13         "onehouse/driver"
  14         "onehouse/protocol"
  15 )
  16
  17 func TestDriverPollSharesOneSnapshotAcrossThermostats(t *testing.T) {
  18         synctest.Test(t, func(t *testing.T) {
  19                 first := testState()
  20                 second := first
  21                 second.MAC = "fc610e7f98ab"
  22                 client := &fakeGateway{states: []State{first, second}}
  23                 transport := &fakeTransport{}
  24                 integration, _ := testDriver(t, client, transport, Thermostat{
  25                         DeviceID: testDeviceID,
  26                         MAC:      testMAC,
  27                 }, Thermostat{
  28                         DeviceID: testSecondID,
  29                         MAC:      second.MAC,
  30                 })
  31
  32                 err := integration.poll(t.Context())
  33                 require.NoError(t, err)
  34                 require.Len(t, client.starts, 1)
  35                 require.Len(t, transport.messages(t, protocol.Availability), 2)
  36                 require.Len(t, transport.messages(t, protocol.Reported), 2)
  37         })
  38 }
  39
  40 func TestDriverMarksOnlyMissingThermostatOfflineAfterThreePollsAndRecovers(t *testing.T) {
  41         synctest.Test(t, func(t *testing.T) {
  42                 second := testState()
  43                 second.MAC = "fc610e7f98ab"
  44                 client := &fakeGateway{states: []State{testState(), second}}
  45                 transport := &fakeTransport{}
  46                 integration, _ := testDriver(t, client, transport, Thermostat{
  47                         DeviceID: testDeviceID,
  48                         MAC:      testMAC,
  49                 }, Thermostat{
  50                         DeviceID: testSecondID,
  51                         MAC:      second.MAC,
  52                 })
  53
  54                 require.NoError(t, integration.poll(t.Context()))
  55
  56                 client.states = []State{second}
  57
  58                 for range 2 {
  59                         require.NoError(t, integration.poll(t.Context()))
  60                         require.Len(t, transport.messages(t, protocol.Availability), 2)
  61                 }
  62
  63                 require.NoError(t, integration.poll(t.Context()))
  64
  65                 messages := transport.messages(t, protocol.Availability)
  66                 require.Len(t, messages, 3)
  67                 require.Equal(t, testDeviceID, messages[2].Address.ID)
  68                 require.Equal(t, protocol.AvailabilityPayload{Status: "offline"}, messages[2].Payload)
  69                 require.NoError(t, integration.poll(t.Context()))
  70                 require.Len(t, transport.messages(t, protocol.Availability), 3)
  71
  72                 client.states = []State{testState(), second}
  73
  74                 require.NoError(t, integration.poll(t.Context()))
  75
  76                 messages = transport.messages(t, protocol.Availability)
  77                 require.Len(t, messages, 4)
  78                 require.Equal(t, protocol.AvailabilityPayload{Status: "online"}, messages[3].Payload)
  79         })
  80 }
  81
  82 func TestDriverFailedPollsHaveNoPrematureStartupAvailability(t *testing.T) {
  83         synctest.Test(t, func(t *testing.T) {
  84                 client := &fakeGateway{err: errors.New("gateway unavailable")}
  85                 transport := &fakeTransport{}
  86                 integration, _ := testDriver(t, client, transport)
  87
  88                 for range 2 {
  89                         require.NoError(t, integration.poll(t.Context()))
  90                         require.Empty(t, transport.messages(t, protocol.Availability))
  91                 }
  92
  93                 require.NoError(t, integration.poll(t.Context()))
  94
  95                 messages := transport.messages(t, protocol.Availability)
  96                 require.Len(t, messages, 1)
  97                 require.Equal(t, protocol.AvailabilityPayload{Status: "offline"}, messages[0].Payload)
  98
  99                 client.err = nil
 100                 client.states = []State{testState()}
 101
 102                 require.NoError(t, integration.poll(t.Context()))
 103
 104                 client.err = errors.New("gateway unavailable")
 105
 106                 require.NoError(t, integration.poll(t.Context()))
 107                 require.Len(t, transport.messages(t, protocol.Availability), 2)
 108         })
 109 }
 110
 111 func TestDriverRunForwardsCommandsRefreshesAndStopsWithoutOutage(t *testing.T) {
 112         synctest.Test(t, func(t *testing.T) {
 113                 client := &fakeGateway{states: []State{testState()}}
 114                 transport := &fakeTransport{}
 115                 integration, router := testDriver(t, client, transport)
 116                 ctx, cancel := context.WithCancel(t.Context())
 117                 defer cancel()
 118
 119                 integration.Refresh()
 120                 integration.Refresh()
 121
 122                 done := make(chan error, 1)
 123
 124                 go func() {
 125                         done <- integration.Run(ctx)
 126                 }()
 127
 128                 synctest.Wait()
 129                 require.Len(t, transport.messages(t, protocol.Descriptor), 1)
 130                 require.Equal(t, 1, client.startCount())
 131                 time.Sleep(6 * time.Second)
 132                 synctest.Wait()
 133                 require.Equal(t, 2, client.startCount())
 134
 135                 command := testCommand()
 136                 publication, err := protocol.BuildCommand(command.DeviceID, command.Capability, command.Metadata, command.Payload)
 137                 require.NoError(t, err)
 138                 require.NoError(t, router.Handle(ctx, publication.Topic, publication.Data, false))
 139                 require.NoError(t, router.Handle(ctx, publication.Topic, publication.Data, false))
 140
 141                 time.Sleep(40 * time.Second)
 142                 synctest.Wait()
 143                 require.Len(t, transport.messages(t, protocol.Result), 2)
 144                 cancel()
 145                 require.ErrorIs(t, <-done, context.Canceled)
 146                 require.Len(t, client.actions, 3)
 147
 148                 messages := transport.messages(t, protocol.Availability)
 149
 150                 for _, message := range messages {
 151                         require.Equal(t, protocol.AvailabilityPayload{Status: "online"}, message.Payload)
 152                 }
 153         })
 154 }
 155
 156 func TestDriverConstructorValidatesAndCopiesSettings(t *testing.T) {
 157         router, err := driver.NewRouter([]string{testDeviceID, testSecondID})
 158         require.NoError(t, err)
 159
 160         publisher := driver.NewPublisher(&fakeTransport{})
 161         valid := Thermostat{
 162                 DeviceID: testDeviceID,
 163                 MAC:      testMAC,
 164         }
 165         testCases := []struct {
 166                 name     string
 167                 settings Settings
 168         }{
 169                 {
 170                         name: "empty gateway is rejected",
 171                 },
 172                 {
 173                         name: "invalid ID is rejected",
 174                         settings: Settings{Thermostats: []Thermostat{{
 175                                 DeviceID: "bad",
 176                                 MAC:      testMAC,
 177                         }}},
 178                 },
 179                 {
 180                         name: "invalid MAC is rejected",
 181                         settings: Settings{Thermostats: []Thermostat{{
 182                                 DeviceID: testDeviceID,
 183                                 MAC:      "bad",
 184                         }}},
 185                 },
 186                 {
 187                         name:     "duplicate ID is rejected",
 188                         settings: Settings{Thermostats: []Thermostat{valid, valid}},
 189                 },
 190                 {
 191                         name: "MAC uniqueness ignores letter case",
 192                         settings: Settings{Thermostats: []Thermostat{valid, {
 193                                 DeviceID: testSecondID,
 194                                 MAC:      strings.ToUpper(testMAC),
 195                         }}},
 196                 },
 197                 {
 198                         name: "poll interval cannot exceed five minutes",
 199                         settings: Settings{
 200                                 PollInterval: time.Hour,
 201                                 Thermostats:  []Thermostat{valid},
 202                         },
 203                 },
 204                 {
 205                         name: "poll interval cannot be below ten seconds",
 206                         settings: Settings{
 207                                 PollInterval: time.Second,
 208                                 Thermostats:  []Thermostat{valid},
 209                         },
 210                 },
 211         }
 212
 213         for _, tc := range testCases {
 214                 t.Run(tc.name, func(t *testing.T) {
 215                         _, err = New(tc.settings, publisher, router, &fakeGateway{})
 216                         require.Error(t, err)
 217                 })
 218         }
 219
 220         emptyRouter, err := driver.NewRouter(nil)
 221         require.NoError(t, err)
 222
 223         _, err = New(Settings{Thermostats: []Thermostat{valid}}, publisher, emptyRouter, &fakeGateway{})
 224         require.ErrorContains(t, err, "queue")
 225
 226         settings := Settings{
 227                 PollInterval: 10 * time.Second,
 228                 Thermostats:  []Thermostat{valid},
 229         }
 230         integration, err := New(settings, publisher, router, &fakeGateway{})
 231         require.NoError(t, err)
 232
 233         settings.Thermostats[0].MAC = "bad"
 234         require.Equal(t, testMAC, integration.thermostats[0].settings.MAC)
 235         require.Equal(t, 10*time.Second, integration.pollInterval)
 236 }
 237
 238 func TestDriverReturnsPublicationFailuresAndCallerCancellation(t *testing.T) {
 239         synctest.Test(t, func(t *testing.T) {
 240                 failure := errors.New("publication unavailable")
 241                 transport := &fakeTransport{
 242                         failKind: protocol.Descriptor,
 243                         failure:  failure,
 244                 }
 245                 integration, _ := testDriver(t, &fakeGateway{states: []State{testState()}}, transport)
 246                 require.ErrorIs(t, integration.Run(t.Context()), failure)
 247
 248                 transport.failKind = protocol.Reported
 249
 250                 require.ErrorIs(t, integration.Run(t.Context()), failure)
 251
 252                 transport.failKind = protocol.Availability
 253                 integration.thermostats[0].missed = 2
 254                 integration.gateway.client = &fakeGateway{}
 255                 require.ErrorIs(t, integration.poll(t.Context()), failure)
 256
 257                 ctx, cancel := context.WithCancel(t.Context())
 258                 cancel()
 259
 260                 require.ErrorIs(t, integration.poll(ctx), context.Canceled)
 261         })
 262 }
 263
 264 func TestDriverReturnsRefreshAndPeriodicPublicationFailures(t *testing.T) {
 265         for _, refresh := range []bool{false, true} {
 266                 t.Run(publicationFailureName(refresh), func(t *testing.T) {
 267                         synctest.Test(t, func(t *testing.T) {
 268                                 client := &fakeGateway{states: []State{testState()}}
 269                                 transport := &fakeTransport{}
 270                                 integration, _ := testDriver(t, client, transport)
 271                                 ctx, cancel := context.WithCancel(t.Context())
 272                                 defer cancel()
 273
 274                                 done := make(chan error, 1)
 275
 276                                 go func() {
 277                                         done <- integration.Run(ctx)
 278                                 }()
 279
 280                                 synctest.Wait()
 281
 282                                 failure := errors.New("publication unavailable")
 283
 284                                 transport.mu.Lock()
 285
 286                                 transport.failKind = protocol.Reported
 287                                 transport.failure = failure
 288                                 transport.mu.Unlock()
 289                                 client.mu.Lock()
 290
 291                                 client.states[0].TargetTemperature = 25
 292                                 client.mu.Unlock()
 293
 294                                 if refresh {
 295                                         integration.Refresh()
 296                                 }
 297
 298                                 require.ErrorIs(t, <-done, failure)
 299                         })
 300                 })
 301         }
 302 }
 303
 304 func TestDriverReturnsCommandPublicationFailuresAndJoinsForwarders(t *testing.T) {
 305         synctest.Test(t, func(t *testing.T) {
 306                 client := &fakeGateway{states: []State{testState()}}
 307                 transport := &fakeTransport{
 308                         failKind: protocol.Result,
 309                         failure:  errors.New("result unavailable"),
 310                 }
 311                 integration, router := testDriver(t, client, transport)
 312                 command := testCommand()
 313                 publication, err := protocol.BuildCommand(command.DeviceID, command.Capability, command.Metadata, command.Payload)
 314                 require.NoError(t, err)
 315                 require.NoError(t, router.Handle(t.Context(), publication.Topic, publication.Data, false))
 316                 require.ErrorIs(t, integration.Run(t.Context()), transport.failure)
 317         })
 318 }
 319
 320 func TestDriverPollKeepsIndependentFailureDuringCancellation(t *testing.T) {
 321         failure := errors.New("independent gateway failure")
 322         ctx, cancel := context.WithCancel(t.Context())
 323         defer cancel()
 324
 325         client := &cancelingGateway{
 326                 cancel:  cancel,
 327                 failure: failure,
 328         }
 329         integration, _ := testDriver(t, client, &fakeTransport{})
 330         err := integration.poll(ctx)
 331         require.ErrorIs(t, err, context.Canceled)
 332         require.ErrorIs(t, err, failure)
 333 }
 334
 335 func publicationFailureName(refresh bool) string {
 336         if refresh {
 337                 return "refresh failure returns to runner"
 338         }
 339
 340         return "periodic failure returns to runner"
 341 }