]> git.giacomolli.uk - onehouse/blob - protocol/batch_payload_test.go
Run explicit drivers with signal cancellation
[onehouse] / protocol / batch_payload_test.go
1 package protocol
2
3 import (
4 "encoding/json"
5 "fmt"
6 "strings"
7 "testing"
8 "time"
9
10 "github.com/stretchr/testify/require"
11 )
12
13 func TestBatchKeepsMemberMetadataAndEarliestDeadline(t *testing.T) {
14 deadline := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
15 later := deadline.Add(time.Minute)
16 payload := testBatch(deadline)
17 payload.Commands[0].ExpiresAt = &later
18 payload.Commands[0].CorrelationID = deviceID
19 payload.Commands[1].CausationID = messageID
20 publication, err := BuildCommandBatch(deviceID, Metadata{MessageID: messageID}, payload)
21 require.NoError(t, err)
22 require.Equal(t, &deadline, publication.ExpiresAt)
23
24 parsed, err := ParseCommandBatch(publication.Topic, publication.Data, false)
25 require.NoError(t, err)
26 require.Equal(t, payload, parsed.Payload)
27
28 for index := 2; index < MaximumCommandBatchSize; index++ {
29 metadata := Metadata{
30 MessageID: fmt.Sprintf("%026d", index+20),
31 ExpiresAt: &deadline,
32 }
33 payload.Commands = append(payload.Commands, BatchCommand{
34 Metadata: metadata,
35 Capability: fmt.Sprintf("relay.%d", index+1),
36 Payload: ValuePayload{Value: json.RawMessage(`false`)},
37 })
38 }
39
40 _, err = BuildCommandBatch(deviceID, Metadata{MessageID: messageID}, payload)
41 require.NoError(t, err)
42
43 payload.Commands = append(payload.Commands, payload.Commands[0])
44 _, err = BuildCommandBatch(deviceID, Metadata{MessageID: messageID}, payload)
45 require.Error(t, err)
46 }
47
48 func TestBatchRejectsMalformedMembersBeforeReturningCommands(t *testing.T) {
49 deadline := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
50 metadata := Metadata{MessageID: messageID}
51 publication, err := BuildCommandBatch(deviceID, metadata, testBatch(deadline))
52 require.NoError(t, err)
53
54 testCases := []struct {
55 name string
56 old string
57 replacement string
58 }{
59 {
60 name: "rejects member schema",
61 old: `"schema":1`,
62 replacement: `"schema":2`,
63 },
64 {
65 name: "rejects unknown batch field",
66 old: `"commands":[`,
67 replacement: `"extra":1,"commands":[`,
68 },
69 {
70 name: "rejects unknown member field",
71 old: `"capability":"relay.1"`,
72 replacement: `"capability":"relay.1","extra":1`,
73 },
74 {
75 name: "rejects unknown member value field",
76 old: `"value":true`,
77 replacement: `"value":true,"extra":1`,
78 },
79 {
80 name: "rejects duplicate member ID",
81 old: commandID,
82 replacement: messageID,
83 },
84 {
85 name: "rejects duplicate target",
86 old: `"relay.2"`,
87 replacement: `"relay.1"`,
88 },
89 {
90 name: "rejects invalid target",
91 old: `"relay.2"`,
92 replacement: `"relay/+"`,
93 },
94 {
95 name: "requires member expiry",
96 old: `"expires_at":"2026-01-01T00:00:00Z",`,
97 replacement: ``,
98 },
99 {
100 name: "requires member value",
101 old: `"value":true`,
102 replacement: ``,
103 },
104 }
105
106 for _, tc := range testCases {
107 t.Run(tc.name, func(t *testing.T) {
108 data := strings.ReplaceAll(string(publication.Data), tc.old, tc.replacement)
109
110 var parsed Message[BatchPayload]
111
112 parsed, err = ParseCommandBatch(publication.Topic, []byte(data), false)
113 require.Error(t, err)
114 require.Empty(t, parsed.Payload.Commands)
115 })
116 }
117
118 payload := testBatch(deadline)
119 payload.Commands = payload.Commands[:1]
120 _, err = BuildCommandBatch(deviceID, metadata, payload)
121 require.Error(t, err)
122
123 _, err = BuildCommandBatch(deviceID, Metadata{MessageID: commandID}, testBatch(deadline))
124 require.Error(t, err)
125
126 data := strings.Replace(string(publication.Data), `"capability":"relay.1"`, `"capability":"relay.1","schema":2`, 1)
127 _, err = ParseCommandBatch(publication.Topic, []byte(data), false)
128 require.Error(t, err)
129 }