1package zerobus
2
3import (
4 "errors"
5 "net"
6 "path/filepath"
7 "testing"
8 "time"
9)
10
11// socketPair returns a Conn whose socket is the client end of a Unix socket,
12// without the bus handshake, and the other end to write raw frames to it.
13func socketPair(t *testing.T) (*Conn, net.Conn) {
14 t.Helper()
15 l, err := net.Listen("unix", filepath.Join(t.TempDir(), "s"))
16 if err != nil {
17 t.Fatal(err)
18 }
19 defer func() { _ = l.Close() }()
20 uc, err := net.DialUnix("unix", nil, &net.UnixAddr{Name: l.Addr().String(), Net: "unix"})
21 if err != nil {
22 t.Fatal(err)
23 }
24 peer, err := l.Accept()
25 if err != nil {
26 t.Fatal(err)
27 }
28 t.Cleanup(func() { _ = uc.Close(); _ = peer.Close() })
29 return &Conn{c: uc, rbuf: make([]byte, defaultBufSz)}, peer
30}
31
32func TestNewSignalToSetsDestination(t *testing.T) {
33 var c Conn
34 c.NewSignalTo(":1.42", "/a", "a.b", "Hit", "s").Str("x")
35 b, err := c.enc.finish()
36 if err != nil {
37 t.Fatal(err)
38 }
39 var m Message
40 if err := parse(b, &m); err != nil {
41 t.Fatal(err)
42 }
43 if m.Type != TypeSignal || m.Destination != ":1.42" || m.Path != "/a" || m.Interface != "a.b" || m.Member != "Hit" {
44 t.Fatalf("message %+v", m)
45 }
46 if got := m.Body().Str(); got != "x" {
47 t.Fatalf("body %q", got)
48 }
49
50 c.NewSignal("/a", "a.b", "Hit", "")
51 b, err = c.enc.finish()
52 if err != nil {
53 t.Fatal(err)
54 }
55 if err := parse(b, &m); err != nil || m.Destination != "" {
56 t.Fatalf("broadcast signal has destination %q (%v)", m.Destination, err)
57 }
58
59 c.NewSignalTo("not a name", "/a", "a.b", "Hit", "")
60 if _, err := c.Send(); err == nil {
61 t.Fatal("invalid destination accepted")
62 }
63
64 c.NewSignalTo("", "/a", "a.b", "Hit", "")
65 if _, err := c.Send(); err == nil {
66 t.Fatal("directed signal without destination accepted")
67 }
68}
69
70func TestDirectedSignalReachesOnlyDestination(t *testing.T) {
71 addr := privateBus(t)
72 server := dial(t, addr)
73 target := dial(t, addr)
74 other := dial(t, addr)
75 const rule = "type='signal',interface='org.example.Directed'"
76 for _, c := range []*Conn{target, other} {
77 if err := c.AddMatch(rule); err != nil {
78 t.Fatal(err)
79 }
80 }
81
82 server.NewSignalTo(target.UniqueName(), "/a", "org.example.Directed", "Hit", "u").Uint32(7)
83 if _, err := server.Send(); err != nil {
84 t.Fatal(err)
85 }
86
87 // The bus routes in order: a signal for other would come before the
88 // reply to its round trip.
89 var leaked bool
90 other.Handle = func(m *Message) { leaked = leaked || m.Member == "Hit" }
91 server.NewCall(busName, busPath, busName, "GetId", "")
92 if _, err := server.Call(); err != nil {
93 t.Fatal(err)
94 }
95 other.NewCall(busName, busPath, busName, "GetId", "")
96 if _, err := other.Call(); err != nil {
97 t.Fatal(err)
98 }
99 if leaked {
100 t.Fatal("directed signal reached another peer")
101 }
102
103 for {
104 m, err := target.ReadMessage()
105 if err != nil {
106 t.Fatal(err)
107 }
108 if m.Member != "Hit" {
109 continue
110 }
111 if m.Type != TypeSignal || m.Sender != server.UniqueName() || m.Destination != target.UniqueName() || m.Body().Uint32() != 7 {
112 t.Fatalf("message %+v", m)
113 }
114 return
115 }
116}
117
118func TestAsyncMethodCallMatching(t *testing.T) {
119 addr := privateBus(t)
120 c := dial(t, addr)
121 other := dial(t, addr)
122
123 // Two requests are in flight at once; the loop only uses Send and
124 // ReadMessage and tells the replies apart by their serial.
125 want := map[uint32]string{}
126 for _, name := range []string{c.UniqueName(), other.UniqueName(), "org.example.Nobody"} {
127 e, serial := c.NewMethodCall(busName, busPath, busName, "GetNameOwner", "s")
128 e.Str(name)
129 sent, err := c.Send()
130 if err != nil {
131 t.Fatal(err)
132 }
133 if sent != serial {
134 t.Fatalf("NewMethodCall serial %d, Send serial %d", serial, sent)
135 }
136 if _, dup := want[serial]; dup {
137 t.Fatalf("serial %d reused", serial)
138 }
139 want[serial] = name
140 }
141
142 for len(want) > 0 {
143 m, err := c.ReadMessage()
144 if err != nil {
145 t.Fatal(err)
146 }
147 if m.Type != TypeMethodReturn && m.Type != TypeError {
148 continue // NameAcquired
149 }
150 name, ok := want[m.ReplySerial]
151 if !ok {
152 t.Fatalf("unexpected reply to serial %d", m.ReplySerial)
153 }
154 delete(want, m.ReplySerial)
155 switch name {
156 case "org.example.Nobody":
157 if m.Type != TypeError {
158 t.Fatalf("reply for %q is %d, want an error", name, m.Type)
159 }
160 default:
161 if m.Type != TypeMethodReturn || m.Body().Str() != name {
162 t.Fatalf("reply for %q is %d", name, m.Type)
163 }
164 }
165 }
166}
167
168func TestBufferedAndPollLoop(t *testing.T) {
169 c, peer := socketPair(t)
170 if c.Buffered() {
171 t.Fatal("empty connection is buffered")
172 }
173 two := append(signal("One"), signal("Two")...)
174 three := signal("Three")
175 // Two whole frames and the start of a third arrive in one write.
176 if _, err := peer.Write(append(two, three[:10]...)); err != nil {
177 t.Fatal(err)
178 }
179 waitReadable(t, c)
180 for _, want := range []string{"One", "Two"} {
181 m, err := c.ReadMessage()
182 if err != nil || m.Member != want {
183 t.Fatalf("got %v, %v; want %q", m, err, want)
184 }
185 if got, wantBuf := c.Buffered(), want == "One"; got != wantBuf {
186 t.Fatalf("Buffered after %q = %v, want %v", want, got, wantBuf)
187 }
188 }
189 // The head of a third frame is not a message.
190 if c.Buffered() {
191 t.Fatal("a partial frame counts as buffered")
192 }
193 if _, err := peer.Write(three[10:]); err != nil {
194 t.Fatal(err)
195 }
196 waitReadable(t, c)
197 m, err := c.ReadMessage()
198 if err != nil || m.Member != "Three" {
199 t.Fatalf("got %v, %v", m, err)
200 }
201 if c.Buffered() {
202 t.Fatal("drained connection is buffered")
203 }
204}
205
206func TestBufferedReportsStickyError(t *testing.T) {
207 c, peer := socketPair(t)
208 bad := signal("Bad")
209 bad[3] = 9 // protocol version
210 if _, err := peer.Write(append(bad, signal("Next")...)); err != nil {
211 t.Fatal(err)
212 }
213 if _, err := c.ReadMessage(); err == nil {
214 t.Fatal("malformed frame accepted")
215 }
216 if !c.Buffered() {
217 t.Fatal("a sticky error must not make a poll loop wait")
218 }
219}
220
221// waitReadable blocks until the socket of c is readable, as a poll loop does.
222func waitReadable(t *testing.T, c *Conn) {
223 t.Helper()
224 rc, err := c.SyscallConn()
225 if err != nil {
226 t.Fatal(err)
227 }
228 var fd int
229 if err := rc.Control(func(f uintptr) { fd = int(f) }); err != nil {
230 t.Fatal(err)
231 }
232 if err := waitFd(fd, 5*time.Second); errors.Is(err, errNoWaitFd) {
233 t.Skip(err)
234 } else if err != nil {
235 t.Fatal(err)
236 }
237}
238
239func TestSignalAllocations(t *testing.T) {
240 c := dial(t, privateBus(t))
241 send := func() {
242 e := c.NewSignalTo(":1.1", "/a", "org.example.Alloc", "Hit", "us")
243 e.Uint32(1)
244 e.Str("x")
245 if _, err := c.Send(); err != nil {
246 t.Fatal(err)
247 }
248 }
249 send() // grow the buffers once
250 if allocs := testing.AllocsPerRun(200, send); allocs != 0 {
251 t.Fatalf("%v allocations per directed signal, want 0", allocs)
252 }
253}
254
255func TestMethodCallAllocations(t *testing.T) {
256 c := dial(t, privateBus(t))
257 var serial uint32
258 call := func() {
259 e, s := c.NewMethodCall(busName, busPath, busName, "GetConnectionUnixProcessID", "s")
260 e.Str(c.UniqueName())
261 sent, err := c.Send()
262 if err != nil || sent != s {
263 t.Fatalf("Send = %d, %v; want %d", sent, err, s)
264 }
265 serial = s
266 }
267 call()
268 if allocs := testing.AllocsPerRun(200, call); allocs != 0 {
269 t.Fatalf("%v allocations per method call, want 0", allocs)
270 }
271 // The replies must be read for the test to leave the bus quiet; they
272 // also prove the calls were well formed.
273 if err := c.c.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil {
274 t.Fatal(err)
275 }
276 got := 0
277 for got < 202 {
278 m, err := c.ReadMessage()
279 if err != nil {
280 t.Fatalf("after %d of 202 replies: %v", got, err)
281 }
282 if m.Type == TypeMethodReturn && m.ReplySerial <= serial {
283 got++
284 }
285 }
286}
287
288func TestReadMessageAllocations(t *testing.T) {
289 addr := privateBus(t)
290 t.Run("method call", func(t *testing.T) {
291 client, server := dial(t, addr), dial(t, addr)
292 ping := func() {
293 e := client.NewCall(server.UniqueName(), "/a", "org.example.Alloc", "Ping", "s")
294 e.SetFlags(FlagNoReplyExpected)
295 e.Str("hello")
296 if _, err := client.Send(); err != nil {
297 t.Fatal(err)
298 }
299 for {
300 m, err := server.ReadMessage()
301 if err != nil {
302 t.Fatal(err)
303 }
304 if m.Type == TypeMethodCall && m.Member == "Ping" {
305 if m.Body().Str() != "hello" {
306 t.Fatal("bad body")
307 }
308 return
309 }
310 }
311 }
312 ping() // grow the buffers once
313 if allocs := testing.AllocsPerRun(200, ping); allocs != 0 {
314 t.Fatalf("%v allocations per method call read, want 0", allocs)
315 }
316 })
317 t.Run("method return", func(t *testing.T) {
318 c := dial(t, addr)
319 ask := func() {
320 e, serial := c.NewMethodCall(busName, busPath, busName, "GetNameOwner", "s")
321 e.Str(busName)
322 if _, err := c.Send(); err != nil {
323 t.Fatal(err)
324 }
325 for {
326 m, err := c.ReadMessage()
327 if err != nil {
328 t.Fatal(err)
329 }
330 if m.Type == TypeMethodReturn && m.ReplySerial == serial {
331 if m.Body().Str() == "" {
332 t.Fatal("empty owner")
333 }
334 return
335 }
336 }
337 }
338 ask()
339 if allocs := testing.AllocsPerRun(200, ask); allocs != 0 {
340 t.Fatalf("%v allocations per method return read, want 0", allocs)
341 }
342 })
343}