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

   1 package driver
   2
   3 import (
   4         "context"
   5         "encoding/json"
   6         "errors"
   7         "testing"
   8         "testing/synctest"
   9         "time"
  10
  11         "github.com/stretchr/testify/require"
  12
  13         "onehouse/protocol"
  14 )
  15
  16 func TestPublisherSuppressesEqualRetainedValuesAndKeepsCommandEvidence(t *testing.T) {
  17         transport := &recordingTransport{}
  18         publisher := NewPublisher(transport)
  19         metadata := protocol.Metadata{}
  20         descriptor := protocol.DescriptorPayload{
  21                 Manufacturer: "example",
  22                 Model:        "switch",
  23                 Firmware:     "1",
  24                 Capabilities: []protocol.Capability{{
  25                         ID:           "relay",
  26                         Type:         "switch",
  27                         ReportsState: true,
  28                 }},
  29         }
  30
  31         for range 2 {
  32                 require.NoError(t, publisher.Descriptor(t.Context(), testDeviceID, metadata, descriptor))
  33                 require.NoError(t, publisher.Availability(t.Context(), testDeviceID, metadata, protocol.AvailabilityPayload{Status: "offline"}))
  34                 require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "relay", metadata, protocol.ValuePayload{Value: json.RawMessage("false")}))
  35         }
  36
  37         require.Len(t, transport.messages(), 3)
  38         require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "relay", protocol.Metadata{CorrelationID: testCommandID}, protocol.ValuePayload{Value: json.RawMessage("false")}))
  39         require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "relay", protocol.Metadata{CausationID: testCommandID}, protocol.ValuePayload{Value: json.RawMessage("false")}))
  40         require.Len(t, transport.messages(), 5)
  41
  42         for _, publication := range transport.messages() {
  43                 require.Equal(t, byte(1), publication.QoS)
  44                 require.True(t, publication.Retain)
  45         }
  46 }
  47
  48 func TestPublisherRepeatsLatestMeasurementWithoutMovingDeadline(t *testing.T) {
  49         synctest.Test(t, func(t *testing.T) {
  50                 ctx, cancel := context.WithCancel(t.Context())
  51                 defer cancel()
  52
  53                 transport := &recordingTransport{}
  54                 publisher := NewPublisher(transport)
  55                 finished := make(chan error, 1)
  56
  57                 require.NoError(t, publisher.Reported(ctx, testDeviceID, "weather", protocol.Metadata{CorrelationID: testCommandID}, protocol.ValuePayload{Value: json.RawMessage(`{"temperature":20,"humidity":50}`)}))
  58
  59                 first := transport.messages()[0]
  60                 go func() {
  61                         finished <- publisher.Run(ctx)
  62                 }()
  63
  64                 for range 299 {
  65                         time.Sleep(time.Second)
  66                         require.NoError(t, publisher.Telemetry(ctx, testDeviceID, "weather", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage(`{ "humidity":50,"temperature":20 }`)}))
  67                         synctest.Wait()
  68                         require.Len(t, transport.messages(), 1)
  69                 }
  70
  71                 time.Sleep(time.Second)
  72                 synctest.Wait()
  73
  74                 packets := transport.messages()
  75                 require.Len(t, packets, 2)
  76
  77                 repeated, err := protocol.ParseReported(packets[1].Topic, packets[1].Data, packets[1].Retain)
  78                 require.NoError(t, err)
  79
  80                 original, err := protocol.ParseReported(first.Topic, first.Data, first.Retain)
  81                 require.NoError(t, err)
  82                 require.NotEqual(t, original.Metadata.MessageID, repeated.Metadata.MessageID)
  83                 require.Equal(t, time.Now().Add(-time.Second).UTC(), *repeated.Metadata.SentAt)
  84                 require.Empty(t, repeated.Metadata.CorrelationID)
  85                 require.Empty(t, repeated.Metadata.CausationID)
  86                 require.JSONEq(t, `{"temperature":20,"humidity":50}`, string(repeated.Payload.Value))
  87                 cancel()
  88                 require.ErrorIs(t, <-finished, context.Canceled)
  89         })
  90 }
  91
  92 func TestPublisherTelemetryKeepsSnapshotOnScheduleAndReconnect(t *testing.T) {
  93         synctest.Test(t, func(t *testing.T) {
  94                 ctx, cancel := context.WithCancel(t.Context())
  95                 defer cancel()
  96
  97                 transport := &recordingTransport{}
  98                 publisher := NewPublisher(transport)
  99                 finished := make(chan error, 1)
 100
 101                 require.NoError(t, publisher.Telemetry(ctx, testDeviceID, "power", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("10")}))
 102
 103                 go func() {
 104                         finished <- publisher.Run(ctx)
 105                 }()
 106
 107                 time.Sleep(4 * time.Minute)
 108                 require.NoError(t, publisher.Telemetry(ctx, testDeviceID, "power", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("20")}))
 109                 require.NoError(t, publisher.Telemetry(ctx, testOtherDeviceID, "power", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("30")}))
 110                 synctest.Wait()
 111
 112                 packets := transport.messages()
 113                 require.Len(t, packets, 3)
 114
 115                 for _, packet := range packets {
 116                         require.Equal(t, byte(0), packet.QoS)
 117                         require.False(t, packet.Retain)
 118                 }
 119
 120                 time.Sleep(time.Minute)
 121                 synctest.Wait()
 122
 123                 packets = transport.messages()
 124                 require.Len(t, packets, 4)
 125
 126                 snapshot, err := protocol.ParseReported(packets[3].Topic, packets[3].Data, packets[3].Retain)
 127                 require.NoError(t, err)
 128                 require.JSONEq(t, "20", string(snapshot.Payload.Value))
 129                 require.Equal(t, byte(1), packets[3].QoS)
 130                 require.Equal(t, time.Now().Add(-time.Minute).UTC(), *snapshot.Metadata.SentAt)
 131                 publisher.Refresh()
 132                 synctest.Wait()
 133                 require.Len(t, transport.messages(), 6)
 134                 cancel()
 135                 require.ErrorIs(t, <-finished, context.Canceled)
 136         })
 137 }
 138
 139 func TestPublisherRepeatsOnlineButNotBooleanOrOffline(t *testing.T) {
 140         synctest.Test(t, func(t *testing.T) {
 141                 ctx, cancel := context.WithCancel(t.Context())
 142                 defer cancel()
 143
 144                 transport := &recordingTransport{}
 145                 publisher := NewPublisher(transport)
 146                 finished := make(chan error, 1)
 147
 148                 require.NoError(t, publisher.Availability(ctx, testDeviceID, protocol.Metadata{}, protocol.AvailabilityPayload{Status: "online"}))
 149                 require.NoError(t, publisher.Availability(ctx, testOtherDeviceID, protocol.Metadata{}, protocol.AvailabilityPayload{Status: "offline"}))
 150                 require.NoError(t, publisher.Reported(ctx, testDeviceID, "relay", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("true")}))
 151
 152                 go func() {
 153                         finished <- publisher.Run(ctx)
 154                 }()
 155
 156                 time.Sleep(5 * time.Minute)
 157                 synctest.Wait()
 158
 159                 packets := transport.messages()
 160                 require.Len(t, packets, 4)
 161
 162                 online, err := protocol.ParseAvailability(packets[3].Topic, packets[3].Data, packets[3].Retain)
 163                 require.NoError(t, err)
 164                 require.Equal(t, "online", online.Payload.Status)
 165                 publisher.Refresh()
 166                 synctest.Wait()
 167                 require.Len(t, transport.messages(), 7)
 168                 cancel()
 169                 require.ErrorIs(t, <-finished, context.Canceled)
 170         })
 171 }
 172
 173 func TestPublisherDoesNotSuppressEventsResultsOrExpiringValues(t *testing.T) {
 174         transport := &recordingTransport{}
 175         publisher := NewPublisher(transport)
 176         expiry := time.Now().UTC().Add(time.Minute)
 177
 178         for range 2 {
 179                 require.NoError(t, publisher.Event(t.Context(), testDeviceID, "button", protocol.Metadata{}, protocol.EventPayload{"action": json.RawMessage(`"pressed"`)}))
 180                 require.NoError(t, publisher.Result(t.Context(), testDeviceID, testCommandID, protocol.Metadata{CorrelationID: testCommandID}, protocol.ResultPayload{Status: "completed"}))
 181                 require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "relay", protocol.Metadata{ExpiresAt: &expiry}, protocol.ValuePayload{Value: json.RawMessage("true")}))
 182         }
 183
 184         require.Len(t, transport.messages(), 6)
 185         require.NoError(t, publisher.repeat(t.Context(), true))
 186         require.Len(t, transport.messages(), 6)
 187 }
 188
 189 func TestPublisherKeepsFailedLatestSampleAndReturnsRepeatFailures(t *testing.T) {
 190         synctest.Test(t, func(t *testing.T) {
 191                 transport := &recordingTransport{}
 192                 publisher := NewPublisher(transport)
 193                 require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "temperature", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("20")}))
 194
 195                 failure := errors.New("broker unavailable")
 196                 transport.setError(failure)
 197                 require.ErrorIs(t, publisher.Telemetry(t.Context(), testDeviceID, "temperature", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("21")}), failure)
 198                 require.ErrorIs(t, publisher.Telemetry(t.Context(), testDeviceID, "temperature", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("22")}), failure)
 199                 publisher.Refresh()
 200                 require.ErrorIs(t, publisher.Run(t.Context()), failure)
 201                 transport.setError(nil)
 202                 require.NoError(t, publisher.repeat(t.Context(), false))
 203
 204                 packets := transport.messages()
 205                 last := packets[len(packets)-1]
 206                 snapshot, err := protocol.ParseReported(last.Topic, last.Data, last.Retain)
 207                 require.NoError(t, err)
 208                 require.JSONEq(t, "22", string(snapshot.Payload.Value))
 209         })
 210 }
 211
 212 func TestPublisherValidatesBeforeSendingAndHonorsCancellation(t *testing.T) {
 213         transport := &recordingTransport{}
 214         publisher := NewPublisher(transport)
 215         require.Error(t, publisher.Reported(t.Context(), "invalid", "relay", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("true")}))
 216         require.Error(t, publisher.Reported(t.Context(), testDeviceID, "relay", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("invalid")}))
 217
 218         past := time.Now().UTC().Add(-time.Second)
 219         require.ErrorContains(t, publisher.Event(t.Context(), testDeviceID, "button", protocol.Metadata{ExpiresAt: &past}, protocol.EventPayload{}), "expired")
 220
 221         ctx, cancel := context.WithCancel(t.Context())
 222         defer cancel()
 223
 224         cancel()
 225         require.ErrorIs(t, publisher.Reported(ctx, testDeviceID, "relay", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("true")}), context.Canceled)
 226         require.Empty(t, transport.messages())
 227 }
 228
 229 func TestPublisherSendTimeoutAndCallerCancellation(t *testing.T) {
 230         synctest.Test(t, func(t *testing.T) {
 231                 publisher := NewPublisher(&waitingTransport{})
 232                 err := publisher.Event(t.Context(), testDeviceID, "button", protocol.Metadata{}, protocol.EventPayload{})
 233                 require.ErrorIs(t, err, context.DeadlineExceeded)
 234
 235                 ctx, cancel := context.WithCancel(t.Context())
 236                 defer cancel()
 237
 238                 finished := make(chan error, 1)
 239                 go func() {
 240                         finished <- publisher.Event(ctx, testDeviceID, "button", protocol.Metadata{}, protocol.EventPayload{})
 241                 }()
 242
 243                 synctest.Wait()
 244                 cancel()
 245                 require.ErrorIs(t, <-finished, context.Canceled)
 246         })
 247 }
 248
 249 func TestPublisherCopiesSnapshotsAndAcceptsTypedNull(t *testing.T) {
 250         transport := &recordingTransport{}
 251         publisher := NewPublisher(transport)
 252         now := time.Now().UTC()
 253         metadata := protocol.Metadata{SentAt: &now}
 254         value := json.RawMessage(`{"power":10}`)
 255         require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "power", metadata, protocol.ValuePayload{Value: value}))
 256
 257         observation := now
 258         now = now.Add(time.Hour)
 259
 260         copy(value, `{"power":99}`)
 261         publisher.Refresh()
 262         publisher.Refresh()
 263         require.NoError(t, publisher.repeat(t.Context(), true))
 264
 265         packets := transport.messages()
 266         require.Len(t, packets, 2)
 267
 268         snapshot, err := protocol.ParseReported(packets[1].Topic, packets[1].Data, packets[1].Retain)
 269         require.NoError(t, err)
 270         require.JSONEq(t, `{"power":10}`, string(snapshot.Payload.Value))
 271         require.Equal(t, observation, *snapshot.Metadata.SentAt)
 272         require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "power", metadata, protocol.ValuePayload{}))
 273 }
 274
 275 func TestPublisherEqualTelemetryAtRepeatTimeSendsRetainedSnapshot(t *testing.T) {
 276         synctest.Test(t, func(t *testing.T) {
 277                 transport := &recordingTransport{}
 278                 publisher := NewPublisher(transport)
 279                 value := protocol.ValuePayload{Value: json.RawMessage("null")}
 280                 require.NoError(t, publisher.Telemetry(t.Context(), testDeviceID, "power", protocol.Metadata{}, value))
 281                 time.Sleep(protocol.MeasurementRepeatInterval)
 282                 require.NoError(t, publisher.Telemetry(t.Context(), testDeviceID, "power", protocol.Metadata{}, value))
 283
 284                 packets := transport.messages()
 285                 require.Len(t, packets, 2)
 286                 require.True(t, packets[1].Retain)
 287
 288                 snapshot, err := protocol.ParseReported(packets[1].Topic, packets[1].Data, packets[1].Retain)
 289                 require.NoError(t, err)
 290                 require.JSONEq(t, "null", string(snapshot.Payload.Value))
 291         })
 292 }
 293
 294 func TestPublisherPreservesDirectMeasurementEvidenceAtRepeatDeadline(t *testing.T) {
 295         testCases := []struct {
 296                 name          string
 297                 correlationID string
 298                 causationID   string
 299         }{
 300                 {
 301                         name:          "correlation keeps caller identity",
 302                         correlationID: testCommandID,
 303                 },
 304                 {
 305                         name:        "causation keeps caller identity",
 306                         causationID: testCommandID,
 307                 },
 308                 {
 309                         name:          "correlation and causation keep caller identity",
 310                         correlationID: testCommandID,
 311                         causationID:   testOtherDeviceID,
 312                 },
 313         }
 314
 315         for _, tc := range testCases {
 316                 t.Run(tc.name, func(t *testing.T) {
 317                         synctest.Test(t, func(t *testing.T) {
 318                                 transport := &recordingTransport{}
 319                                 publisher := NewPublisher(transport)
 320                                 value := protocol.ValuePayload{Value: json.RawMessage("20")}
 321                                 require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "temperature", protocol.Metadata{}, value))
 322                                 time.Sleep(protocol.MeasurementRepeatInterval)
 323
 324                                 observedAt := time.Now().UTC()
 325                                 metadata := protocol.Metadata{
 326                                         MessageID:     testOtherDeviceID,
 327                                         SentAt:        &observedAt,
 328                                         CorrelationID: tc.correlationID,
 329                                         CausationID:   tc.causationID,
 330                                 }
 331                                 require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "temperature", metadata, value))
 332
 333                                 packets := transport.messages()
 334                                 require.Len(t, packets, 2)
 335
 336                                 message, err := protocol.ParseReported(packets[1].Topic, packets[1].Data, packets[1].Retain)
 337                                 require.NoError(t, err)
 338                                 require.Equal(t, metadata, message.Metadata)
 339                                 require.Equal(t, value, message.Payload)
 340                         })
 341                 })
 342         }
 343 }
 344
 345 func TestPublisherRunsAsDriverAndRefreshesOnReconnect(t *testing.T) {
 346         synctest.Test(t, func(t *testing.T) {
 347                 ctx, cancel := context.WithCancel(t.Context())
 348                 defer cancel()
 349
 350                 transport := &recordingTransport{}
 351                 publisher := NewPublisher(transport)
 352                 require.NoError(t, publisher.Reported(ctx, testDeviceID, "relay", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("true")}))
 353
 354                 connected := make(chan struct{}, 1)
 355                 finished := make(chan error, 1)
 356                 go func() {
 357                         finished <- Run(ctx, []Driver{publisher}, connected)
 358                 }()
 359
 360                 synctest.Wait()
 361
 362                 connected <- struct{}{}
 363
 364                 synctest.Wait()
 365
 366                 packets := transport.messages()
 367                 require.Len(t, packets, 2)
 368                 require.NotEqual(t, packets[0].Data, packets[1].Data)
 369                 cancel()
 370                 require.NoError(t, <-finished)
 371         })
 372 }
 373
 374 type waitingTransport struct{}
 375
 376 func (p *waitingTransport) Publish(ctx context.Context, _ protocol.Publication) error {
 377         <-ctx.Done()
 378
 379         return ctx.Err()
 380 }