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

   1 package alamo
   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 TestCommandHandlerOrdersPowerAndTargetWithActionReadback(t *testing.T) {
  17         testCases := []struct {
  18                 name         string
  19                 initialPower bool
  20                 value        string
  21                 wantActions  []string
  22         }{
  23                 {
  24                         name:        "target precedes power on",
  25                         value:       `{"mode":"heat","target_temperature":22}`,
  26                         wantActions: []string{"read", "target", "on"},
  27                 },
  28                 {
  29                         name:         "power off precedes target change",
  30                         initialPower: true,
  31                         value:        `{"mode":"off","target_temperature":22}`,
  32                         wantActions:  []string{"read", "off", "target"},
  33                 },
  34                 {
  35                         name:         "active target change turns off then on",
  36                         initialPower: true,
  37                         value:        `{"mode":"heat","target_temperature":22}`,
  38                         wantActions:  []string{"read", "off", "target", "on"},
  39                 },
  40                 {
  41                         name:        "equal settings still get fresh readback",
  42                         value:       `{"mode":"off","target_temperature":20}`,
  43                         wantActions: []string{"read"},
  44                 },
  45         }
  46
  47         for _, tc := range testCases {
  48                 t.Run(tc.name, func(t *testing.T) {
  49                         synctest.Test(t, func(t *testing.T) {
  50                                 state := testState()
  51                                 state.Power = tc.initialPower
  52                                 client := &fakeGateway{states: []State{state}}
  53                                 transport := &fakeTransport{}
  54                                 integration, _ := testDriver(t, client, transport)
  55                                 handler := &commandHandler{
  56                                         gateway:    &integration.gateway,
  57                                         thermostat: integration.thermostats[0],
  58                                 }
  59                                 command := testCommand()
  60                                 command.Payload.Value = []byte(tc.value)
  61                                 require.NoError(t, handler.handle(t.Context(), []driver.Command{command}))
  62                                 require.Equal(t, tc.wantActions, actionNames(client.actions))
  63                                 require.Len(t, transport.messages(t, protocol.Result), 1)
  64
  65                                 result := transport.messages(t, protocol.Result)[0]
  66                                 require.Equal(t, protocol.ResultPayload{Status: "completed"}, result.Payload)
  67
  68                                 reported := transport.messages(t, protocol.Reported)
  69                                 require.Len(t, reported, 1)
  70                                 require.Equal(t, command.Metadata.MessageID, reported[0].Metadata.CausationID)
  71
  72                                 for index := 1; index < len(client.starts); index++ {
  73                                         require.GreaterOrEqual(t, client.starts[index].Sub(client.starts[index-1]), 6*time.Second)
  74                                 }
  75                         })
  76                 })
  77         }
  78 }
  79
  80 func TestCommandHandlerStopsAtUnmatchedIntermediateReadback(t *testing.T) {
  81         testCases := []struct {
  82                 name         string
  83                 power        bool
  84                 ignorePower  bool
  85                 ignoreTarget bool
  86                 wantActions  []string
  87         }{
  88                 {
  89                         name:        "failed off confirmation prevents target write",
  90                         power:       true,
  91                         ignorePower: true,
  92                         wantActions: []string{"read", "off"},
  93                 },
  94                 {
  95                         name:         "failed target confirmation prevents power on",
  96                         ignoreTarget: true,
  97                         wantActions:  []string{"read", "target"},
  98                 },
  99                 {
 100                         name:        "failed power on confirmation does not complete",
 101                         ignorePower: true,
 102                         wantActions: []string{"read", "target", "on"},
 103                 },
 104         }
 105
 106         for _, tc := range testCases {
 107                 t.Run(tc.name, func(t *testing.T) {
 108                         synctest.Test(t, func(t *testing.T) {
 109                                 state := testState()
 110                                 state.Power = tc.power
 111                                 client := &fakeGateway{
 112                                         states:       []State{state},
 113                                         ignorePower:  tc.ignorePower,
 114                                         ignoreTarget: tc.ignoreTarget,
 115                                 }
 116                                 transport := &fakeTransport{}
 117                                 integration, _ := testDriver(t, client, transport)
 118                                 handler := commandHandler{
 119                                         gateway:    &integration.gateway,
 120                                         thermostat: integration.thermostats[0],
 121                                 }
 122                                 require.NoError(t, handler.handle(t.Context(), []driver.Command{testCommand()}))
 123                                 require.Equal(t, tc.wantActions, actionNames(client.actions))
 124
 125                                 result, ok := transport.messages(t, protocol.Result)[0].Payload.(protocol.ResultPayload)
 126                                 require.True(t, ok)
 127                                 require.Equal(t, "failed", result.Status)
 128                                 require.NotEmpty(t, result.Reason)
 129                         })
 130                 })
 131         }
 132 }
 133
 134 func TestCommandHandlerReplaysFirstResultAfterFeedbackOrResultSendFailure(t *testing.T) {
 135         for _, kind := range []protocol.Kind{protocol.Reported, protocol.Result, protocol.Availability} {
 136                 t.Run(string(kind), func(t *testing.T) {
 137                         synctest.Test(t, func(t *testing.T) {
 138                                 failure := errors.New("transport unavailable")
 139                                 transport := &fakeTransport{
 140                                         failKind: kind,
 141                                         failure:  failure,
 142                                 }
 143                                 client := &fakeGateway{states: []State{testState()}}
 144                                 integration, _ := testDriver(t, client, transport)
 145                                 handler := commandHandler{
 146                                         gateway:    &integration.gateway,
 147                                         thermostat: integration.thermostats[0],
 148                                 }
 149                                 command := testCommand()
 150                                 require.ErrorIs(t, handler.handle(t.Context(), []driver.Command{command}), failure)
 151
 152                                 calls := len(client.actions)
 153                                 record, exists := integration.thermostats[0].ledger.Lookup(command.Metadata.MessageID)
 154                                 require.True(t, exists)
 155                                 require.Equal(t, "completed", record.Payload.Status)
 156
 157                                 transport.failKind = ""
 158
 159                                 require.NoError(t, handler.handle(t.Context(), []driver.Command{command, command}))
 160                                 require.Len(t, client.actions, calls)
 161
 162                                 messages := transport.messages(t, protocol.Result)
 163                                 last := messages[len(messages)-1]
 164                                 require.Equal(t, record.Metadata, last.Metadata)
 165                                 require.Equal(t, record.Payload, last.Payload)
 166                         })
 167                 })
 168         }
 169 }
 170
 171 func TestCommandHandlerRejectsInvalidCommandsWithoutGatewayWork(t *testing.T) {
 172         testCases := []struct {
 173                 name   string
 174                 change func(*driver.Command)
 175         }{
 176                 {
 177                         name: "expired work is rejected",
 178                         change: func(command *driver.Command) {
 179                                 expires := time.Now().Add(-time.Second).UTC()
 180                                 command.Metadata.ExpiresAt = &expires
 181                         },
 182                 },
 183                 {
 184                         name: "missing expiry is rejected",
 185                         change: func(command *driver.Command) {
 186                                 command.Metadata.ExpiresAt = nil
 187                         },
 188                 },
 189                 {
 190                         name: "another capability is rejected",
 191                         change: func(command *driver.Command) {
 192                                 command.Capability = "other"
 193                         },
 194                 },
 195                 {
 196                         name: "incomplete settings are rejected",
 197                         change: func(command *driver.Command) {
 198                                 command.Payload.Value = []byte(`{}`)
 199                         },
 200                 },
 201         }
 202
 203         for _, tc := range testCases {
 204                 t.Run(tc.name, func(t *testing.T) {
 205                         client := &fakeGateway{}
 206                         transport := &fakeTransport{}
 207                         integration, _ := testDriver(t, client, transport)
 208                         handler := commandHandler{
 209                                 gateway:    &integration.gateway,
 210                                 thermostat: integration.thermostats[0],
 211                         }
 212                         command := testCommand()
 213                         tc.change(&command)
 214                         require.NoError(t, handler.handle(t.Context(), []driver.Command{command, command}))
 215                         require.Empty(t, client.starts)
 216
 217                         messages := transport.messages(t, protocol.Result)
 218                         require.Len(t, messages, 2)
 219                         require.Equal(t, messages[0], messages[1])
 220
 221                         result, ok := messages[0].Payload.(protocol.ResultPayload)
 222                         require.True(t, ok)
 223                         require.Equal(t, "rejected", result.Status)
 224                 })
 225         }
 226 }
 227
 228 func TestCommandHandlerCommandExpiryStopsLaterActionsAndReplay(t *testing.T) {
 229         synctest.Test(t, func(t *testing.T) {
 230                 client := &fakeGateway{states: []State{testState()}}
 231                 transport := &fakeTransport{}
 232                 integration, _ := testDriver(t, client, transport)
 233                 handler := commandHandler{
 234                         gateway:    &integration.gateway,
 235                         thermostat: integration.thermostats[0],
 236                 }
 237                 command := testCommand()
 238                 expires := time.Now().Add(7 * time.Second).UTC()
 239                 command.Metadata.ExpiresAt = &expires
 240                 require.NoError(t, handler.handle(t.Context(), []driver.Command{command}))
 241                 require.Equal(t, []string{"read", "target"}, actionNames(client.actions))
 242                 require.False(t, client.states[0].Power)
 243
 244                 reported := transport.messages(t, protocol.Reported)
 245                 require.Len(t, reported, 1)
 246                 require.Equal(t, client.starts[1].UTC(), *reported[0].Metadata.SentAt)
 247                 require.NoError(t, handler.handle(t.Context(), []driver.Command{command}))
 248                 require.Len(t, client.actions, 2)
 249
 250                 messages := transport.messages(t, protocol.Result)
 251                 require.Len(t, messages, 2)
 252                 require.Equal(t, messages[0], messages[1])
 253                 require.Equal(t, protocol.ResultPayload{
 254                         Status: "failed",
 255                         Reason: "Command expired",
 256                 }, messages[0].Payload)
 257         })
 258 }
 259
 260 func TestCommandHandlerRetainsFailuresWithoutRepeatingPhysicalActions(t *testing.T) {
 261         for _, position := range []int{1, 2, 3, 4} {
 262                 t.Run(actionFailureName(position), func(t *testing.T) {
 263                         synctest.Test(t, func(t *testing.T) {
 264                                 state := testState()
 265                                 state.Power = true
 266                                 client := &fakeGateway{
 267                                         states:       []State{state},
 268                                         failActionAt: position,
 269                                 }
 270                                 transport := &fakeTransport{}
 271                                 integration, _ := testDriver(t, client, transport)
 272                                 handler := commandHandler{
 273                                         gateway:    &integration.gateway,
 274                                         thermostat: integration.thermostats[0],
 275                                 }
 276                                 command := testCommand()
 277                                 require.NoError(t, handler.handle(t.Context(), []driver.Command{command, command}))
 278                                 require.Len(t, client.actions, position)
 279
 280                                 messages := transport.messages(t, protocol.Result)
 281                                 require.Len(t, messages, 2)
 282                                 require.Equal(t, messages[0], messages[1])
 283
 284                                 payload, ok := messages[0].Payload.(protocol.ResultPayload)
 285                                 require.True(t, ok)
 286                                 require.Equal(t, "failed", payload.Status)
 287                         })
 288                 })
 289         }
 290 }
 291
 292 func TestCommandHandlerValidatesOwnershipAndReturnsPublicationFailures(t *testing.T) {
 293         client := &fakeGateway{}
 294         failure := errors.New("result unavailable")
 295         transport := &fakeTransport{
 296                 failKind: protocol.Result,
 297                 failure:  failure,
 298         }
 299         integration, _ := testDriver(t, client, transport)
 300         handler := commandHandler{
 301                 gateway:    &integration.gateway,
 302                 thermostat: integration.thermostats[0],
 303         }
 304         command := testCommand()
 305         command.DeviceID = testSecondID
 306         require.ErrorContains(t, handler.handle(t.Context(), []driver.Command{command}), "not owned")
 307
 308         command = testCommand()
 309         command.Capability = "invalid"
 310         require.ErrorIs(t, handler.handle(t.Context(), []driver.Command{command}), failure)
 311         require.ErrorIs(t, handler.handle(t.Context(), []driver.Command{command}), failure)
 312         require.Empty(t, client.starts)
 313 }
 314
 315 func TestCommandHandlerCancellationStopsBeforeAnotherActionAndKeepsOutcome(t *testing.T) {
 316         synctest.Test(t, func(t *testing.T) {
 317                 client := &fakeGateway{states: []State{testState()}}
 318                 transport := &fakeTransport{}
 319                 integration, _ := testDriver(t, client, transport)
 320                 handler := commandHandler{
 321                         gateway:    &integration.gateway,
 322                         thermostat: integration.thermostats[0],
 323                 }
 324                 ctx, cancel := context.WithCancel(t.Context())
 325                 defer cancel()
 326
 327                 done := make(chan error, 1)
 328                 command := testCommand()
 329
 330                 go func() {
 331                         done <- handler.handle(ctx, []driver.Command{command})
 332                 }()
 333
 334                 synctest.Wait()
 335                 cancel()
 336                 require.ErrorIs(t, <-done, context.Canceled)
 337                 require.Len(t, client.actions, 1)
 338
 339                 _, exists := integration.thermostats[0].ledger.Lookup(command.Metadata.MessageID)
 340                 require.True(t, exists)
 341                 require.Empty(t, transport.messages(t, protocol.Result))
 342         })
 343 }
 344
 345 func TestCommandHandlerKeepsIndependentFailureDuringCancellation(t *testing.T) {
 346         failure := errors.New("independent gateway failure")
 347         ctx, cancel := context.WithCancel(t.Context())
 348         defer cancel()
 349
 350         client := &cancelingGateway{
 351                 cancel:  cancel,
 352                 failure: failure,
 353         }
 354         integration, _ := testDriver(t, client, &fakeTransport{})
 355         handler := commandHandler{
 356                 gateway:    &integration.gateway,
 357                 thermostat: integration.thermostats[0],
 358         }
 359         err := handler.handle(ctx, []driver.Command{testCommand()})
 360         require.ErrorIs(t, err, context.Canceled)
 361         require.ErrorIs(t, err, failure)
 362 }
 363
 364 func actionNames(actions []Action) []string {
 365         var names []string
 366
 367         for _, action := range actions {
 368                 name := "read"
 369                 if action.TargetTemperature != nil {
 370                         name = "target"
 371                 }
 372
 373                 if action.Power != nil {
 374                         name = "off"
 375                         if *action.Power {
 376                                 name = "on"
 377                         }
 378                 }
 379
 380                 names = append(names, name)
 381         }
 382
 383         return names
 384 }
 385
 386 func actionFailureName(position int) string {
 387         return []string{"", "failed initial read", "failed power off", "failed target write", "failed power on"}[position]
 388 }