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

   1 package m31
   2
   3 import (
   4         "context"
   5         "errors"
   6         "fmt"
   7         "slices"
   8         "testing"
   9         "testing/synctest"
  10         "time"
  11
  12         "github.com/stretchr/testify/require"
  13
  14         "onehouse/driver"
  15         "onehouse/protocol"
  16 )
  17
  18 func TestCommandsWriteContiguousRangesAndReplayResults(t *testing.T) {
  19         c, client, transport := testConnection(t)
  20         h := &commandHandler{connection: c}
  21
  22         var commands []driver.Command
  23
  24         for address := 5; address < 18; address++ {
  25                 commands = append(commands, testCommand(t, c.owner.topology.outputs[address].id, "true"))
  26         }
  27
  28         slices.Reverse(commands)
  29         require.NoError(t, h.handle(t.Context(), commands))
  30         require.Len(t, client.writes, 1)
  31         require.EqualValues(t, 5, client.writes[0].address)
  32         require.True(t, client.writes[0].multiple)
  33         require.Len(t, client.writes[0].values, 13)
  34         require.Equal(t, 1, client.coilReads)
  35
  36         reports := transport.ofKind(protocol.Reported)
  37         results := transport.ofKind(protocol.Result)
  38
  39         require.Len(t, reports, 13)
  40         require.Len(t, results, 13)
  41
  42         for _, command := range commands {
  43                 for _, report := range reports {
  44                         if report.Address.Target == command.Capability {
  45                                 require.Equal(t, command.Metadata.MessageID, report.Metadata.CausationID)
  46                         }
  47                 }
  48         }
  49
  50         require.NoError(t, h.handle(t.Context(), commands))
  51         require.Len(t, client.writes, 1)
  52         require.Equal(t, 1, client.coilReads)
  53
  54         replayed := transport.ofKind(protocol.Result)[13:]
  55         byCommand := make(map[string]protocol.Metadata)
  56
  57         for _, result := range results {
  58                 byCommand[result.Address.Target] = result.Metadata
  59         }
  60
  61         for _, result := range replayed {
  62                 require.Equal(t, byCommand[result.Address.Target], result.Metadata)
  63         }
  64 }
  65
  66 func TestCommandsKeepGapsAndRejectNonsteadyMembers(t *testing.T) {
  67         c, client, transport := testConnection(t)
  68         h := &commandHandler{connection: c}
  69         commands := []driver.Command{
  70                 testCommand(t, "m0.do0", "true"),
  71                 testCommand(t, "m0.do1", "true"),
  72                 testCommand(t, "m0.do2", "true"),
  73                 testCommand(t, "m0.do4", "true"),
  74                 testCommand(t, "m2.ao0", "2"),
  75                 testCommand(t, "m0.di0", "true"),
  76         }
  77         require.NoError(t, h.handle(t.Context(), commands))
  78         require.Len(t, client.writes, 2)
  79         require.EqualValues(t, 0, client.writes[0].address)
  80         require.Len(t, client.writes[0].values, 2)
  81         require.EqualValues(t, 4, client.writes[1].address)
  82         require.False(t, client.writes[1].multiple)
  83         require.False(t, client.coils[2])
  84         require.False(t, client.coils[3])
  85         require.Len(t, transport.ofKind(protocol.Result), 6)
  86 }
  87
  88 func TestCommandsFailAllOnTransportErrorAndOnlyMismatchOtherwise(t *testing.T) {
  89         testCases := []struct {
  90                 name             string
  91                 change           func(*fakeClient)
  92                 transportFailure bool
  93         }{
  94                 {name: "fails all on write failure", transportFailure: true, change: func(c *fakeClient) {
  95                         c.writeErr = errors.New("write failed")
  96                 }},
  97                 {name: "fails all after a partial write", transportFailure: true, change: func(c *fakeClient) {
  98                         c.writeErr = errors.New("later write failed")
  99                         c.failAfterWrites = 1
 100                 }},
 101                 {name: "fails all on read failure", transportFailure: true, change: func(c *fakeClient) {
 102                         c.readErr = errors.New("read failed")
 103                 }},
 104                 {name: "fails all on short read", transportFailure: true, change: func(c *fakeClient) {
 105                         c.shortRead = true
 106                 }},
 107                 {name: "fails only mismatching member", change: func(c *fakeClient) {
 108                         c.mismatch = 1
 109                 }},
 110         }
 111
 112         for _, tc := range testCases {
 113                 t.Run(tc.name, func(t *testing.T) {
 114                         c, client, transport := testConnection(t)
 115                         tc.change(client)
 116
 117                         h := &commandHandler{connection: c}
 118                         commands := []driver.Command{testCommand(t, "m0.do0", "true"), testCommand(t, "m0.do1", "true"), testCommand(t, "m1.do7", "true")}
 119                         err := h.handle(t.Context(), commands)
 120                         if tc.transportFailure {
 121                                 var failure *deviceError
 122
 123                                 require.ErrorAs(t, err, &failure)
 124                         } else {
 125                                 require.NoError(t, err)
 126                         }
 127
 128                         require.Len(t, transport.ofKind(protocol.Result), 3)
 129
 130                         for index, command := range commands {
 131                                 record, exists := c.owner.ledger.Lookup(command.Metadata.MessageID)
 132                                 require.True(t, exists)
 133
 134                                 status := "completed"
 135                                 if tc.transportFailure || index == 1 {
 136                                         status = "failed"
 137                                 }
 138
 139                                 require.Equal(t, status, record.Payload.Status)
 140                         }
 141                 })
 142         }
 143 }
 144
 145 func TestCommandsCompletePulseWithoutReadbackOrFeedback(t *testing.T) {
 146         c, client, transport := testConnection(t)
 147         h := &commandHandler{connection: c}
 148         command := testCommand(t, "m0.do2", "true")
 149         require.NoError(t, h.handle(t.Context(), []driver.Command{command}))
 150         require.Len(t, client.writes, 1)
 151         require.Zero(t, client.coilReads)
 152         require.Empty(t, transport.ofKind(protocol.Reported))
 153         require.NoError(t, h.handle(t.Context(), []driver.Command{command}))
 154         require.Len(t, client.writes, 1)
 155
 156         command = testCommand(t, "m0.do2", "false")
 157         require.NoError(t, h.handle(t.Context(), []driver.Command{command}))
 158         require.Len(t, client.writes, 1)
 159
 160         record, exists := c.owner.ledger.Lookup(command.Metadata.MessageID)
 161         require.True(t, exists)
 162         require.Equal(t, "rejected", record.Payload.Status)
 163 }
 164
 165 func TestCommandsWriteVoltageAndCheckFeedback(t *testing.T) {
 166         testCases := []struct {
 167                 name       string
 168                 change     func(*fakeClient)
 169                 wantStatus string
 170                 wantError  bool
 171         }{
 172                 {name: "completes verified voltage", wantStatus: "completed", change: func(*fakeClient) {
 173                 }},
 174                 {name: "fails mismatching voltage", wantStatus: "failed", change: func(c *fakeClient) {
 175                         c.mismatch = 3
 176                 }},
 177                 {name: "fails register transport", wantStatus: "failed", wantError: true, change: func(c *fakeClient) {
 178                         c.registerErr = errors.New("register failed")
 179                 }},
 180                 {name: "fails analog read transport", wantStatus: "failed", wantError: true, change: func(c *fakeClient) {
 181                         c.registerReadErr = errors.New("read failed")
 182                 }},
 183                 {name: "fails unsafe feedback", wantStatus: "failed", wantError: true, change: func(c *fakeClient) {
 184                         c.analog[0] = 10001
 185                 }},
 186                 {name: "fails incomplete feedback", wantStatus: "failed", wantError: true, change: func(c *fakeClient) {
 187                         c.analog = c.analog[:4]
 188                 }},
 189         }
 190
 191         for _, tc := range testCases {
 192                 t.Run(tc.name, func(t *testing.T) {
 193                         c, client, transport := testConnection(t)
 194                         tc.change(client)
 195
 196                         h := &commandHandler{connection: c}
 197                         command := testCommand(t, "m2.ao3", "5.5")
 198                         err := h.handle(t.Context(), []driver.Command{command})
 199                         if tc.wantError {
 200                                 require.Error(t, err)
 201                         } else {
 202                                 require.NoError(t, err)
 203                         }
 204
 205                         record, exists := c.owner.ledger.Lookup(command.Metadata.MessageID)
 206                         require.True(t, exists)
 207                         require.Equal(t, tc.wantStatus, record.Payload.Status)
 208
 209                         if tc.wantStatus == "completed" {
 210                                 require.EqualValues(t, 5500, client.analog[3])
 211                                 require.Len(t, transport.messages, 2)
 212                                 require.Equal(t, protocol.Reported, transport.messages[0].Address.Kind)
 213                                 require.Equal(t, command.Metadata.MessageID, transport.messages[0].Metadata.CausationID)
 214                         }
 215                 })
 216         }
 217 }
 218
 219 func TestCommandsRememberBeforeFailedPublication(t *testing.T) {
 220         for _, kind := range []protocol.Kind{protocol.Reported, protocol.Result} {
 221                 t.Run(fmt.Sprint("preserves first outcome after ", kind, " failure"), func(t *testing.T) {
 222                         c, client, transport := testConnection(t)
 223                         h := &commandHandler{connection: c}
 224                         failure := errors.New("publication failed")
 225                         transport.failKind, transport.err = kind, failure
 226                         commands := []driver.Command{testCommand(t, "m0.do0", "true"), testCommand(t, "m0.do1", "true")}
 227                         require.ErrorIs(t, h.handle(t.Context(), commands), failure)
 228
 229                         record, exists := c.owner.ledger.Lookup(commands[1].Metadata.MessageID)
 230                         require.True(t, exists)
 231                         require.Equal(t, "completed", record.Payload.Status)
 232
 233                         transport.err = nil
 234
 235                         require.NoError(t, h.handle(t.Context(), commands))
 236                         require.Len(t, client.writes, 1)
 237                 })
 238         }
 239 }
 240
 241 func TestCommandsRecheckExpiryBeforeEachWriteAndHonorCancellation(t *testing.T) {
 242         synctest.Test(t, func(t *testing.T) {
 243                 c, client, _ := testConnection(t)
 244                 h := &commandHandler{connection: c}
 245                 first := testCommand(t, "m0.do0", "true")
 246                 second := testCommand(t, "m0.do4", "true")
 247                 expires := time.Now().Add(time.Millisecond)
 248                 second.Metadata.ExpiresAt = &expires
 249                 client.writeDelay = 2 * time.Millisecond
 250
 251                 require.NoError(t, h.handle(t.Context(), []driver.Command{first, second}))
 252                 require.Len(t, client.writes, 1)
 253
 254                 record, exists := c.owner.ledger.Lookup(second.Metadata.MessageID)
 255                 require.True(t, exists)
 256                 require.Equal(t, "rejected", record.Payload.Status)
 257
 258                 ctx, cancel := context.WithCancel(t.Context())
 259                 cancel()
 260
 261                 third := testCommand(t, "m0.do1", "true")
 262                 require.ErrorIs(t, h.handle(ctx, []driver.Command{third}), context.Canceled)
 263                 require.Len(t, client.writes, 1)
 264         })
 265 }
 266
 267 func TestCommandsCompleteUnchangedSingleWithoutState(t *testing.T) {
 268         c, client, transport := testConnection(t)
 269         h := &commandHandler{connection: c}
 270         command := testCommand(t, "m0.do0", "false")
 271         require.NoError(t, h.handle(t.Context(), []driver.Command{command}))
 272         require.Len(t, client.writes, 1)
 273         require.False(t, client.writes[0].multiple)
 274         require.Len(t, transport.messages, 1)
 275         require.Equal(t, protocol.Result, transport.messages[0].Address.Kind)
 276 }
 277
 278 func TestCommandsReturnPublicationErrorsWithoutLosingOutcomes(t *testing.T) {
 279         testCases := []struct {
 280                 name            string
 281                 capability      string
 282                 value           string
 283                 offline         bool
 284                 physicalFailure bool
 285         }{
 286                 {name: "retains completed pulse after result failure", capability: "m0.do2", value: "true"},
 287                 {name: "retains voltage after result failure", capability: "m2.ao0", value: "5"},
 288                 {name: "returns rejection failure", capability: "m0.di0", value: "true"},
 289                 {name: "returns offline result failure", capability: "m0.do0", value: "true", offline: true},
 290                 {name: "returns offline rejection failure", capability: "m0.di0", value: "true", offline: true},
 291                 {name: "joins pulse and result failure", capability: "m0.do2", value: "true", physicalFailure: true},
 292                 {name: "joins batch and result failure", capability: "m0.do0", value: "true", physicalFailure: true},
 293         }
 294
 295         for _, tc := range testCases {
 296                 t.Run(tc.name, func(t *testing.T) {
 297                         c, client, transport := testConnection(t)
 298                         h := &commandHandler{connection: c}
 299                         command := testCommand(t, tc.capability, tc.value)
 300                         sendErr := errors.New("send failed")
 301                         physicalErr := errors.New("physical failure")
 302                         transport.failKind, transport.err = protocol.Result, sendErr
 303                         if tc.physicalFailure {
 304                                 client.writeErr = physicalErr
 305                         }
 306
 307                         var err error
 308
 309                         if tc.offline {
 310                                 err = h.unavailable(t.Context(), []driver.Command{command})
 311                         } else {
 312                                 err = h.handle(t.Context(), []driver.Command{command})
 313                         }
 314
 315                         require.ErrorIs(t, err, sendErr)
 316
 317                         if tc.physicalFailure {
 318                                 require.ErrorIs(t, err, physicalErr)
 319
 320                                 var physical *deviceError
 321
 322                                 require.NotErrorAs(t, err, &physical)
 323                         }
 324
 325                         record, exists := c.owner.ledger.Lookup(command.Metadata.MessageID)
 326                         require.True(t, exists)
 327
 328                         transport.err, client.writeErr = nil, nil
 329                         count := len(client.writes)
 330
 331                         require.NoError(t, h.handle(t.Context(), []driver.Command{command}))
 332                         require.Len(t, client.writes, count)
 333
 334                         replay := transport.ofKind(protocol.Result)
 335                         require.Equal(t, record.Metadata, replay[len(replay)-1].Metadata)
 336                 })
 337         }
 338 }