git.giacomolli.uk - onehouse/blob - driver/router_test.go

   1 package driver
   2
   3 import (
   4         "context"
   5         "encoding/json"
   6         "testing"
   7         "testing/synctest"
   8         "time"
   9
  10         "github.com/stretchr/testify/require"
  11
  12         "onehouse/protocol"
  13 )
  14
  15 func TestRouterKeepsSinglesAndBatchesInOneDeviceOrder(t *testing.T) {
  16         router, err := NewRouter([]string{testDeviceID})
  17         require.NoError(t, err)
  18
  19         queue, exists := router.Queue(testDeviceID)
  20         require.True(t, exists)
  21
  22         first := routerCommand(t, testDeviceID, "relay", time.Minute)
  23         second := routerCommand(t, testDeviceID, "output", time.Minute)
  24         third := routerCommand(t, testDeviceID, "pulse", time.Minute)
  25         last := routerCommand(t, testDeviceID, "batch", time.Minute)
  26         batch, err := protocol.BuildCommandBatch(testDeviceID, second.Metadata, protocol.BatchPayload{
  27                 Commands: []protocol.BatchCommand{
  28                         {
  29                                 Metadata:   second.Metadata,
  30                                 Capability: second.Capability,
  31                                 Payload:    second.Payload,
  32                         },
  33                         {
  34                                 Metadata:   third.Metadata,
  35                                 Capability: third.Capability,
  36                                 Payload:    third.Payload,
  37                         },
  38                 },
  39         })
  40         require.NoError(t, err)
  41
  42         for _, publication := range []protocol.Publication{commandPublication(t, first), batch, commandPublication(t, last)} {
  43                 require.NoError(t, router.Handle(t.Context(), publication.Topic, publication.Data, publication.Retain))
  44         }
  45
  46         require.Len(t, queue, 3)
  47         require.Equal(t, []Command{first}, <-queue)
  48         require.Equal(t, []Command{second, third}, <-queue)
  49         require.Equal(t, []Command{last}, <-queue)
  50 }
  51
  52 func TestRouterBuildsOnlyExplicitDeviceSubscriptions(t *testing.T) {
  53         router, err := NewRouter([]string{testOtherDeviceID, testDeviceID})
  54         require.NoError(t, err)
  55
  56         first, err := protocol.CommandSubscription(testDeviceID)
  57         require.NoError(t, err)
  58
  59         second, err := protocol.CommandSubscription(testOtherDeviceID)
  60         require.NoError(t, err)
  61
  62         subscriptions := router.Subscriptions()
  63         require.Equal(t, []protocol.Subscription{first, second}, subscriptions)
  64
  65         subscriptions[0].Filter = "changed"
  66
  67         require.Equal(t, first, router.Subscriptions()[0])
  68
  69         queue, exists := router.Queue("unknown")
  70         require.False(t, exists)
  71         require.Nil(t, queue)
  72
  73         _, err = NewRouter([]string{"invalid"})
  74         require.Error(t, err)
  75
  76         _, err = NewRouter([]string{testDeviceID, testDeviceID})
  77         require.Error(t, err)
  78
  79         readOnly, err := NewRouter(nil)
  80         require.NoError(t, err)
  81         require.Empty(t, readOnly.Subscriptions())
  82 }
  83
  84 func TestRouterRejectsInvalidUnownedAndExpiredDeliveries(t *testing.T) {
  85         router, err := NewRouter([]string{testDeviceID})
  86         require.NoError(t, err)
  87
  88         queue, exists := router.Queue(testDeviceID)
  89         require.True(t, exists)
  90
  91         valid := commandPublication(t, routerCommand(t, testDeviceID, "relay", time.Minute))
  92         unowned := commandPublication(t, routerCommand(t, testOtherDeviceID, "relay", time.Minute))
  93         expired := commandPublication(t, routerCommand(t, testDeviceID, "relay", -time.Second))
  94         value, err := protocol.BuildReported(testDeviceID, "relay", protocolMetadata(t), protocol.ValuePayload{Value: json.RawMessage("true")})
  95         require.NoError(t, err)
  96
  97         availability, err := protocol.BuildAvailability(testDeviceID, protocolMetadata(t), protocol.AvailabilityPayload{Status: "online"})
  98         require.NoError(t, err)
  99
 100         testCases := []struct {
 101                 name        string
 102                 publication protocol.Publication
 103         }{
 104                 {
 105                         name: "retained command rejects",
 106                         publication: protocol.Publication{
 107                                 Topic:  valid.Topic,
 108                                 Data:   valid.Data,
 109                                 Retain: true,
 110                         },
 111                 },
 112                 {
 113                         name: "malformed payload rejects",
 114                         publication: protocol.Publication{
 115                                 Topic: valid.Topic,
 116                                 Data:  []byte(`{}`),
 117                         },
 118                 },
 119                 {
 120                         name: "unknown topic rejects",
 121                         publication: protocol.Publication{
 122                                 Topic: "invalid",
 123                                 Data:  valid.Data,
 124                         },
 125                 },
 126                 {
 127                         name:        "unowned device rejects",
 128                         publication: unowned,
 129                 },
 130                 {
 131                         name:        "expired command rejects",
 132                         publication: expired,
 133                 },
 134                 {
 135                         name:        "reported value is not command intake",
 136                         publication: value,
 137                 },
 138                 {
 139                         name:        "availability is not command intake",
 140                         publication: availability,
 141                 },
 142         }
 143
 144         for _, tc := range testCases {
 145                 t.Run(tc.name, func(t *testing.T) {
 146                         require.Error(t, router.Handle(t.Context(), tc.publication.Topic, tc.publication.Data, tc.publication.Retain))
 147                         require.Empty(t, queue)
 148                 })
 149         }
 150 }
 151
 152 func TestRouterRejectsWholeBatchWhenOneMemberExpires(t *testing.T) {
 153         router, err := NewRouter([]string{testDeviceID})
 154         require.NoError(t, err)
 155
 156         first := routerCommand(t, testDeviceID, "relay", time.Minute)
 157         second := routerCommand(t, testDeviceID, "pulse", -time.Second)
 158         batch, err := protocol.BuildCommandBatch(testDeviceID, first.Metadata, protocol.BatchPayload{
 159                 Commands: []protocol.BatchCommand{
 160                         {
 161                                 Metadata:   first.Metadata,
 162                                 Capability: first.Capability,
 163                                 Payload:    first.Payload,
 164                         },
 165                         {
 166                                 Metadata:   second.Metadata,
 167                                 Capability: second.Capability,
 168                                 Payload:    second.Payload,
 169                         },
 170                 },
 171         })
 172         require.NoError(t, err)
 173         require.Error(t, router.Handle(t.Context(), batch.Topic, batch.Data, false))
 174
 175         queue, exists := router.Queue(testDeviceID)
 176         require.True(t, exists)
 177         require.Empty(t, queue)
 178 }
 179
 180 func TestRouterBackpressureHonorsCancellationAndEarliestExpiry(t *testing.T) {
 181         synctest.Test(t, func(t *testing.T) {
 182                 router, err := NewRouter([]string{testDeviceID})
 183                 require.NoError(t, err)
 184
 185                 valid := commandPublication(t, routerCommand(t, testDeviceID, "relay", time.Hour))
 186
 187                 for range commandQueueCapacity {
 188                         require.NoError(t, router.Handle(t.Context(), valid.Topic, valid.Data, false))
 189                 }
 190
 191                 ctx, cancel := context.WithCancel(t.Context())
 192                 defer cancel()
 193
 194                 finished := make(chan error, 1)
 195                 go func() {
 196                         finished <- router.Handle(ctx, valid.Topic, valid.Data, false)
 197                 }()
 198
 199                 synctest.Wait()
 200                 require.Empty(t, finished)
 201                 cancel()
 202                 require.ErrorIs(t, <-finished, context.Canceled)
 203
 204                 short := commandPublication(t, routerCommand(t, testDeviceID, "pulse", time.Second))
 205                 require.ErrorIs(t, router.Handle(t.Context(), short.Topic, short.Data, false), context.DeadlineExceeded)
 206
 207                 first := routerCommand(t, testDeviceID, "relay", time.Minute)
 208                 second := routerCommand(t, testDeviceID, "pulse", time.Second)
 209                 batch, err := protocol.BuildCommandBatch(testDeviceID, first.Metadata, protocol.BatchPayload{Commands: []protocol.BatchCommand{
 210                         {
 211                                 Metadata:   first.Metadata,
 212                                 Capability: first.Capability,
 213                                 Payload:    first.Payload,
 214                         },
 215                         {
 216                                 Metadata:   second.Metadata,
 217                                 Capability: second.Capability,
 218                                 Payload:    second.Payload,
 219                         },
 220                 }})
 221                 require.NoError(t, err)
 222
 223                 start := time.Now()
 224
 225                 require.ErrorIs(t, router.Handle(t.Context(), batch.Topic, batch.Data, false), context.DeadlineExceeded)
 226                 require.Equal(t, time.Second, time.Since(start))
 227
 228                 queue, exists := router.Queue(testDeviceID)
 229                 require.True(t, exists)
 230                 require.Len(t, queue, commandQueueCapacity)
 231                 require.ErrorIs(t, router.Handle(ctx, valid.Topic, valid.Data, false), context.Canceled)
 232                 <-queue
 233                 require.NoError(t, router.Handle(t.Context(), valid.Topic, valid.Data, false))
 234         })
 235 }
 236
 237 func routerCommand(t *testing.T, deviceID string, capability string, lifetime time.Duration) Command {
 238         t.Helper()
 239
 240         metadata := protocolMetadata(t)
 241         expiry := time.Now().UTC().Add(lifetime)
 242         metadata.ExpiresAt = &expiry
 243
 244         return Command{
 245                 DeviceID:   deviceID,
 246                 Capability: capability,
 247                 Metadata:   metadata,
 248                 Payload:    protocol.ValuePayload{Value: json.RawMessage("true")},
 249         }
 250 }
 251
 252 func commandPublication(t *testing.T, command Command) protocol.Publication {
 253         t.Helper()
 254
 255         publication, err := protocol.BuildCommand(command.DeviceID, command.Capability, command.Metadata, command.Payload)
 256         require.NoError(t, err)
 257
 258         return publication
 259 }
 260
 261 func protocolMetadata(t *testing.T) protocol.Metadata {
 262         t.Helper()
 263
 264         now := time.Now().UTC()
 265         id, err := protocol.NewID(now)
 266         require.NoError(t, err)
 267
 268         return protocol.Metadata{
 269                 MessageID: id,
 270                 SentAt:    &now,
 271         }
 272 }