git.giacomolli.uk - onehouse/blob - protocol/builders_test.go

   1 package protocol
   2
   3 import (
   4         "encoding/json"
   5         "strings"
   6         "testing"
   7         "time"
   8
   9         "github.com/stretchr/testify/require"
  10 )
  11
  12 func TestTypedPublicationsRoundTrip(t *testing.T) {
  13         now := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
  14         deadline := now.Add(time.Minute)
  15         metadata := Metadata{
  16                 MessageID: messageID,
  17                 SentAt:    &now,
  18         }
  19         commandMetadata := Metadata{
  20                 MessageID: messageID,
  21                 ExpiresAt: &deadline,
  22         }
  23         resultMetadata := Metadata{
  24                 MessageID:     messageID,
  25                 CorrelationID: commandID,
  26         }
  27         testCases := []struct {
  28                 name    string
  29                 kind    Kind
  30                 build   func() (Publication, error)
  31                 payload any
  32                 retain  bool
  33                 qos     byte
  34         }{
  35                 {
  36                         name:   "preserves metadata contract",
  37                         kind:   Descriptor,
  38                         retain: true,
  39                         qos:    1,
  40
  41                         build: func() (Publication, error) {
  42                                 return BuildDescriptor(deviceID, metadata, testDescriptor())
  43                         },
  44                         payload: testDescriptor(),
  45                 },
  46                 {
  47                         name:   "preserves online reachability",
  48                         kind:   Availability,
  49                         retain: true,
  50                         qos:    1,
  51
  52                         build: func() (Publication, error) {
  53                                 return BuildAvailability(deviceID, metadata, AvailabilityPayload{Status: "online"})
  54                         },
  55                         payload: AvailabilityPayload{Status: "online"},
  56                 },
  57                 {
  58                         name:   "preserves retained object value",
  59                         kind:   Reported,
  60                         retain: true,
  61                         qos:    1,
  62
  63                         build: func() (Publication, error) {
  64                                 return BuildReported(deviceID, "weather", metadata, ValuePayload{Value: json.RawMessage(`{"temperature":20,"humidity":50}`)})
  65                         },
  66                         payload: ValuePayload{Value: json.RawMessage(`{"temperature":20,"humidity":50}`)},
  67                 },
  68                 {
  69                         name: "preserves unconfirmed sample",
  70                         kind: Telemetry,
  71                         qos:  0,
  72
  73                         build: func() (Publication, error) {
  74                                 return BuildTelemetry(deviceID, "power", metadata, ValuePayload{Value: json.RawMessage(`12.5`)})
  75                         },
  76                         payload: ValuePayload{Value: json.RawMessage(`12.5`)},
  77                 },
  78                 {
  79                         name: "preserves open event data",
  80                         kind: Event,
  81                         qos:  1,
  82
  83                         build: func() (Publication, error) {
  84                                 return BuildEvent(deviceID, "motion", metadata, EventPayload{
  85                                         "action": json.RawMessage(`"motion"`),
  86                                         "custom": json.RawMessage(`7`),
  87                                 })
  88                         },
  89                         payload: EventPayload{
  90                                 "action": json.RawMessage(`"motion"`),
  91                                 "custom": json.RawMessage(`7`),
  92                         },
  93                 },
  94                 {
  95                         name: "preserves command deadline",
  96                         kind: Command,
  97                         qos:  1,
  98
  99                         build: func() (Publication, error) {
 100                                 return BuildCommand(deviceID, "relay", commandMetadata, ValuePayload{Value: json.RawMessage(`true`)})
 101                         },
 102                         payload: ValuePayload{Value: json.RawMessage(`true`)},
 103                 },
 104                 {
 105                         name: "preserves ordered member requests",
 106                         kind: CommandBatch,
 107                         qos:  1,
 108
 109                         build: func() (Publication, error) {
 110                                 return BuildCommandBatch(deviceID, metadata, testBatch(deadline))
 111                         },
 112                         payload: testBatch(deadline),
 113                 },
 114                 {
 115                         name: "preserves correlated result",
 116                         kind: Result,
 117                         qos:  1,
 118
 119                         build: func() (Publication, error) {
 120                                 return BuildResult(deviceID, commandID, resultMetadata, ResultPayload{Status: "completed"})
 121                         },
 122                         payload: ResultPayload{Status: "completed"},
 123                 },
 124                 {
 125                         name:   "preserves expiring hub snapshot",
 126                         kind:   HubStatus,
 127                         retain: true,
 128                         qos:    1,
 129
 130                         build: func() (Publication, error) {
 131                                 return BuildHubStatus(deviceID, commandMetadata, HubStatusPayload{
 132                                         ID:      deviceID,
 133                                         Address: "example.com",
 134                                         Status:  "ready",
 135                                 })
 136                         },
 137                         payload: HubStatusPayload{
 138                                 ID:      deviceID,
 139                                 Address: "example.com",
 140                                 Status:  "ready",
 141                         },
 142                 },
 143         }
 144
 145         for _, tc := range testCases {
 146                 t.Run(tc.name, func(t *testing.T) {
 147                         publication, err := tc.build()
 148                         require.NoError(t, err)
 149                         require.Equal(t, tc.retain, publication.Retain)
 150                         require.Equal(t, tc.qos, publication.QoS)
 151
 152                         parsed, err := Parse(publication.Topic, publication.Data, publication.Retain)
 153                         require.NoError(t, err)
 154                         require.Equal(t, tc.kind, parsed.Address.Kind)
 155                         require.Equal(t, deviceID, parsed.Address.ID)
 156                         require.Equal(t, tc.payload, parsed.Payload)
 157                         require.Equal(t, messageID, parsed.Metadata.MessageID)
 158
 159                         live, err := Parse(publication.Topic, publication.Data, false)
 160                         require.NoError(t, err)
 161                         require.False(t, live.Retained)
 162
 163                         if !tc.retain {
 164                                 _, err = Parse(publication.Topic, publication.Data, true)
 165                                 require.Error(t, err)
 166                         }
 167
 168                         malformed := strings.Replace(publication.Topic, Root, "wrong/v1", 1)
 169                         _, err = Parse(malformed, publication.Data, false)
 170                         require.Error(t, err)
 171
 172                         unknown := strings.Replace(string(publication.Data), `"schema":1`, `"schema":1,"unknown":true`, 1)
 173                         _, err = Parse(publication.Topic, []byte(unknown), false)
 174                         require.Error(t, err)
 175
 176                         wrongSchema := strings.Replace(string(publication.Data), `"schema":1`, `"schema":2`, 1)
 177                         _, err = Parse(publication.Topic, []byte(wrongSchema), false)
 178                         require.Error(t, err)
 179
 180                         if tc.kind == Command || tc.kind == CommandBatch || tc.kind == HubStatus {
 181                                 require.Equal(t, &deadline, publication.ExpiresAt)
 182                         }
 183
 184                         if tc.kind == Command {
 185                                 response, err := ParseTopic(publication.ResponseTopic)
 186                                 require.NoError(t, err)
 187                                 require.Equal(t, Result, response.Kind)
 188                                 require.Equal(t, messageID, response.Target)
 189                                 require.Equal(t, messageID, publication.CorrelationData)
 190                         } else {
 191                                 require.Empty(t, publication.ResponseTopic)
 192                                 require.Empty(t, publication.CorrelationData)
 193                         }
 194                 })
 195         }
 196 }
 197
 198 func TestTypedParsersRejectWrongOperations(t *testing.T) {
 199         parsers := []func(string, []byte, bool) error{
 200                 func(topic string, data []byte, retained bool) error {
 201                         _, err := ParseDescriptor(topic, data, retained)
 202                         return err
 203                 },
 204                 func(topic string, data []byte, retained bool) error {
 205                         _, err := ParseAvailability(topic, data, retained)
 206                         return err
 207                 },
 208                 func(topic string, data []byte, retained bool) error {
 209                         _, err := ParseReported(topic, data, retained)
 210                         return err
 211                 },
 212                 func(topic string, data []byte, retained bool) error {
 213                         _, err := ParseTelemetry(topic, data, retained)
 214                         return err
 215                 },
 216                 func(topic string, data []byte, retained bool) error {
 217                         _, err := ParseEvent(topic, data, retained)
 218                         return err
 219                 },
 220                 func(topic string, data []byte, retained bool) error {
 221                         _, err := ParseCommand(topic, data, retained)
 222                         return err
 223                 },
 224                 func(topic string, data []byte, retained bool) error {
 225                         _, err := ParseCommandBatch(topic, data, retained)
 226                         return err
 227                 },
 228                 func(topic string, data []byte, retained bool) error {
 229                         _, err := ParseResult(topic, data, retained)
 230                         return err
 231                 },
 232                 func(topic string, data []byte, retained bool) error {
 233                         _, err := ParseHubStatus(topic, data, retained)
 234                         return err
 235                 },
 236         }
 237
 238         for _, parse := range parsers {
 239                 require.Error(t, parse(Root+"/devices/"+deviceID+"/unknown", []byte(`{}`), false))
 240         }
 241 }