Dispatch parsed addresses without topic retries

[onehouse] / protocol / builders_test.go

diff --git a/protocol/builders_test.go b/protocol/builders_test.go
index 159cca2bc7314b6be70ccdb2e3c85e2750aaf23b..20bd547b1b4ece5b5170a2331c35867c090109ab 100644 (file)
--- a/protocol/builders_test.go
+++ b/protocol/builders_test.go
@@ -182,7 +182,7 @@ func TestTypedPublicationsRoundTrip(t *testing.T) {
                        }
 
                        if tc.kind == Command {
                        }
 
                        if tc.kind == Command {

-                               response, err := ParseTopic(publication.ResponseTopic)
+                               response, err := parseTopic(publication.ResponseTopic)

                                require.NoError(t, err)
                                require.Equal(t, Result, response.Kind)
                                require.Equal(t, messageID, response.Target)
                                require.NoError(t, err)
                                require.Equal(t, Result, response.Kind)
                                require.Equal(t, messageID, response.Target)


@@ -239,3 +239,139 @@ func TestTypedParsersRejectWrongOperations(t *testing.T) {
                require.Error(t, parse(Root+"/devices/"+deviceID+"/unknown", []byte(`{}`), false))
        }
 }
                require.Error(t, parse(Root+"/devices/"+deviceID+"/unknown", []byte(`{}`), false))
        }
 }

+
+func TestTypedParsersKeepValidationAfterTopicDispatch(t *testing.T) {
+       deadline := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+       metadata := Metadata{
+               MessageID:     messageID,
+               ExpiresAt:     &deadline,
+               CorrelationID: commandID,
+       }
+       testCases := []struct {
+               name    string
+               kind    Kind
+               target  string
+               payload any
+               parse   func(string, []byte, bool) (any, error)
+       }{
+               {
+                       name:    "descriptor keeps its machine metadata",
+                       kind:    Descriptor,
+                       payload: testDescriptor(),
+                       parse: func(topic string, data []byte, retained bool) (any, error) {
+                               parsed, err := ParseDescriptor(topic, data, retained)
+                               return parsed.Payload, err
+                       },
+               },
+               {
+                       name:    "availability keeps its reachability",
+                       kind:    Availability,
+                       payload: AvailabilityPayload{Status: "online"},
+                       parse: func(topic string, data []byte, retained bool) (any, error) {
+                               parsed, err := ParseAvailability(topic, data, retained)
+                               return parsed.Payload, err
+                       },
+               },
+               {
+                       name:    "reported keeps its complete value",
+                       kind:    Reported,
+                       target:  "relay",
+                       payload: ValuePayload{Value: json.RawMessage(`true`)},
+                       parse: func(topic string, data []byte, retained bool) (any, error) {
+                               parsed, err := ParseReported(topic, data, retained)
+                               return parsed.Payload, err
+                       },
+               },
+               {
+                       name:    "telemetry keeps its sample",
+                       kind:    Telemetry,
+                       target:  "power",
+                       payload: ValuePayload{Value: json.RawMessage(`12`)},
+                       parse: func(topic string, data []byte, retained bool) (any, error) {
+                               parsed, err := ParseTelemetry(topic, data, retained)
+                               return parsed.Payload, err
+                       },
+               },
+               {
+                       name:    "event keeps its open fields",
+                       kind:    Event,
+                       target:  "press",
+                       payload: EventPayload{"action": json.RawMessage(`"press"`)},
+                       parse: func(topic string, data []byte, retained bool) (any, error) {
+                               parsed, err := ParseEvent(topic, data, retained)
+                               return parsed.Payload, err
+                       },
+               },
+               {
+                       name:    "command keeps its requested value",
+                       kind:    Command,
+                       target:  "relay",
+                       payload: ValuePayload{Value: json.RawMessage(`true`)},
+                       parse: func(topic string, data []byte, retained bool) (any, error) {
+                               parsed, err := ParseCommand(topic, data, retained)
+                               return parsed.Payload, err
+                       },
+               },
+               {
+                       name:    "command batch keeps its member order",
+                       kind:    CommandBatch,
+                       payload: testBatch(deadline),
+                       parse: func(topic string, data []byte, retained bool) (any, error) {
+                               parsed, err := ParseCommandBatch(topic, data, retained)
+                               return parsed.Payload, err
+                       },
+               },
+               {
+                       name:    "result keeps its correlated outcome",
+                       kind:    Result,
+                       target:  commandID,
+                       payload: ResultPayload{Status: "completed"},
+                       parse: func(topic string, data []byte, retained bool) (any, error) {
+                               parsed, err := ParseResult(topic, data, retained)
+                               return parsed.Payload, err
+                       },
+               },
+               {
+                       name: "hub status keeps its snapshot",
+                       kind: HubStatus,
+                       payload: HubStatusPayload{
+                               ID:      deviceID,
+                               Address: "example.com",
+                               Status:  "ready",
+                       },
+                       parse: func(topic string, data []byte, retained bool) (any, error) {
+                               parsed, err := ParseHubStatus(topic, data, retained)
+                               return parsed.Payload, err
+                       },
+               },
+       }
+
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       topic, err := (Address{
+                               Kind:   tc.kind,
+                               ID:     deviceID,
+                               Target: tc.target,
+                       }).topic()
+                       require.NoError(t, err)
+
+                       data, err := encodeEnvelope(metadata, tc.payload)
+                       require.NoError(t, err)
+
+                       parsed, err := tc.parse(topic, data, false)
+                       require.NoError(t, err)
+                       require.Equal(t, tc.payload, parsed)
+
+                       _, err = tc.parse(topic, []byte(`{}`), false)
+                       require.Error(t, err)
+
+                       wrongTopic := Root + "/devices/" + deviceID + "/availability"
+                       if tc.kind == Availability {
+                               wrongTopic = Root + "/devices/" + deviceID + "/descriptor"
+                       }
+
+                       _, err = tc.parse(wrongTopic, data, false)
+                       require.Error(t, err)
+               })
+       }
+}