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 }
 242
 243 func TestTypedParsersKeepValidationAfterTopicDispatch(t *testing.T) {
 244         deadline := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
 245         metadata := Metadata{
 246                 MessageID:     messageID,
 247                 ExpiresAt:     &deadline,
 248                 CorrelationID: commandID,
 249         }
 250         testCases := []struct {
 251                 name    string
 252                 kind    Kind
 253                 target  string
 254                 payload any
 255                 parse   func(string, []byte, bool) (any, error)
 256         }{
 257                 {
 258                         name:    "descriptor keeps its machine metadata",
 259                         kind:    Descriptor,
 260                         payload: testDescriptor(),
 261                         parse: func(topic string, data []byte, retained bool) (any, error) {
 262                                 parsed, err := ParseDescriptor(topic, data, retained)
 263                                 return parsed.Payload, err
 264                         },
 265                 },
 266                 {
 267                         name:    "availability keeps its reachability",
 268                         kind:    Availability,
 269                         payload: AvailabilityPayload{Status: "online"},
 270                         parse: func(topic string, data []byte, retained bool) (any, error) {
 271                                 parsed, err := ParseAvailability(topic, data, retained)
 272                                 return parsed.Payload, err
 273                         },
 274                 },
 275                 {
 276                         name:    "reported keeps its complete value",
 277                         kind:    Reported,
 278                         target:  "relay",
 279                         payload: ValuePayload{Value: json.RawMessage(`true`)},
 280                         parse: func(topic string, data []byte, retained bool) (any, error) {
 281                                 parsed, err := ParseReported(topic, data, retained)
 282                                 return parsed.Payload, err
 283                         },
 284                 },
 285                 {
 286                         name:    "telemetry keeps its sample",
 287                         kind:    Telemetry,
 288                         target:  "power",
 289                         payload: ValuePayload{Value: json.RawMessage(`12`)},
 290                         parse: func(topic string, data []byte, retained bool) (any, error) {
 291                                 parsed, err := ParseTelemetry(topic, data, retained)
 292                                 return parsed.Payload, err
 293                         },
 294                 },
 295                 {
 296                         name:    "event keeps its open fields",
 297                         kind:    Event,
 298                         target:  "press",
 299                         payload: EventPayload{"action": json.RawMessage(`"press"`)},
 300                         parse: func(topic string, data []byte, retained bool) (any, error) {
 301                                 parsed, err := ParseEvent(topic, data, retained)
 302                                 return parsed.Payload, err
 303                         },
 304                 },
 305                 {
 306                         name:    "command keeps its requested value",
 307                         kind:    Command,
 308                         target:  "relay",
 309                         payload: ValuePayload{Value: json.RawMessage(`true`)},
 310                         parse: func(topic string, data []byte, retained bool) (any, error) {
 311                                 parsed, err := ParseCommand(topic, data, retained)
 312                                 return parsed.Payload, err
 313                         },
 314                 },
 315                 {
 316                         name:    "command batch keeps its member order",
 317                         kind:    CommandBatch,
 318                         payload: testBatch(deadline),
 319                         parse: func(topic string, data []byte, retained bool) (any, error) {
 320                                 parsed, err := ParseCommandBatch(topic, data, retained)
 321                                 return parsed.Payload, err
 322                         },
 323                 },
 324                 {
 325                         name:    "result keeps its correlated outcome",
 326                         kind:    Result,
 327                         target:  commandID,
 328                         payload: ResultPayload{Status: "completed"},
 329                         parse: func(topic string, data []byte, retained bool) (any, error) {
 330                                 parsed, err := ParseResult(topic, data, retained)
 331                                 return parsed.Payload, err
 332                         },
 333                 },
 334                 {
 335                         name: "hub status keeps its snapshot",
 336                         kind: HubStatus,
 337                         payload: HubStatusPayload{
 338                                 ID:      deviceID,
 339                                 Address: "example.com",
 340                                 Status:  "ready",
 341                         },
 342                         parse: func(topic string, data []byte, retained bool) (any, error) {
 343                                 parsed, err := ParseHubStatus(topic, data, retained)
 344                                 return parsed.Payload, err
 345                         },
 346                 },
 347         }
 348
 349         for _, tc := range testCases {
 350                 t.Run(tc.name, func(t *testing.T) {
 351                         topic, err := (Address{
 352                                 Kind:   tc.kind,
 353                                 ID:     deviceID,
 354                                 Target: tc.target,
 355                         }).topic()
 356                         require.NoError(t, err)
 357
 358                         data, err := encodeEnvelope(metadata, tc.payload)
 359                         require.NoError(t, err)
 360
 361                         parsed, err := tc.parse(topic, data, false)
 362                         require.NoError(t, err)
 363                         require.Equal(t, tc.payload, parsed)
 364
 365                         _, err = tc.parse(topic, []byte(`{}`), false)
 366                         require.Error(t, err)
 367
 368                         wrongTopic := Root + "/devices/" + deviceID + "/availability"
 369                         if tc.kind == Availability {
 370                                 wrongTopic = Root + "/devices/" + deviceID + "/descriptor"
 371                         }
 372
 373                         _, err = tc.parse(wrongTopic, data, false)
 374                         require.Error(t, err)
 375                 })
 376         }
 377 }