Antonio Mika
·
2026-10-11
1package pipe
2
3import (
4 "bytes"
5 "context"
6 "crypto/ed25519"
7 "crypto/rand"
8 "errors"
9 "fmt"
10 "io"
11 "log/slog"
12 "net"
13 "os"
14 "strings"
15 "sync"
16 "testing"
17 "time"
18
19 "github.com/antoniomika/syncmap"
20 "github.com/picosh/pico/pkg/db"
21 "github.com/picosh/pico/pkg/db/stub"
22 "github.com/picosh/pico/pkg/pssh"
23 psub "github.com/picosh/pico/pkg/pubsub"
24 "github.com/picosh/pico/pkg/shared"
25 "golang.org/x/crypto/ssh"
26)
27
28type TestDB struct {
29 *stub.StubDB
30 mu sync.RWMutex
31 Users []*db.User
32 Pubkeys []*db.PublicKey
33 Features []*db.FeatureFlag
34 PipeMonitors []*db.PipeMonitor
35}
36
37func NewTestDB(logger *slog.Logger) *TestDB {
38 return &TestDB{
39 StubDB: stub.NewStubDB(logger),
40 }
41}
42
43func (t *TestDB) FindUserByPubkey(key string) (*db.User, error) {
44 t.mu.RLock()
45 defer t.mu.RUnlock()
46 for _, pk := range t.Pubkeys {
47 if pk.Key == key {
48 return t.findUserLocked(pk.UserID)
49 }
50 }
51 return nil, fmt.Errorf("user not found for pubkey")
52}
53
54func (t *TestDB) findUserLocked(userID string) (*db.User, error) {
55 for _, user := range t.Users {
56 if user.ID == userID {
57 cp := *user
58 return &cp, nil
59 }
60 }
61 return nil, fmt.Errorf("user not found")
62}
63
64func (t *TestDB) FindUser(userID string) (*db.User, error) {
65 t.mu.RLock()
66 defer t.mu.RUnlock()
67 return t.findUserLocked(userID)
68}
69
70func (t *TestDB) FindUserByName(name string) (*db.User, error) {
71 t.mu.RLock()
72 defer t.mu.RUnlock()
73 for _, user := range t.Users {
74 if user.Name == name {
75 cp := *user
76 return &cp, nil
77 }
78 }
79 return nil, fmt.Errorf("user not found")
80}
81
82func (t *TestDB) FindFeature(userID, name string) (*db.FeatureFlag, error) {
83 t.mu.RLock()
84 defer t.mu.RUnlock()
85 for _, ff := range t.Features {
86 if ff.UserID == userID && ff.Name == name {
87 cp := *ff
88 return &cp, nil
89 }
90 }
91 return nil, fmt.Errorf("feature not found")
92}
93
94func (t *TestDB) InsertAccessLog(_ *db.AccessLog) error {
95 return nil
96}
97
98func (t *TestDB) Close() error {
99 return nil
100}
101
102func (t *TestDB) AddUser(user *db.User) {
103 t.mu.Lock()
104 defer t.mu.Unlock()
105 cp := *user
106 t.Users = append(t.Users, &cp)
107}
108
109func (t *TestDB) AddPubkey(pubkey *db.PublicKey) {
110 t.mu.Lock()
111 defer t.mu.Unlock()
112 cp := *pubkey
113 t.Pubkeys = append(t.Pubkeys, &cp)
114}
115
116func (t *TestDB) UpsertPipeMonitor(userID, topic string, dur time.Duration, winEnd *time.Time) error {
117 t.mu.Lock()
118 defer t.mu.Unlock()
119 var winEndCopy *time.Time
120 if winEnd != nil {
121 w := *winEnd
122 winEndCopy = &w
123 }
124 for _, m := range t.PipeMonitors {
125 if m.UserId == userID && m.Topic == topic {
126 m.WindowDur = dur
127 m.WindowEnd = winEndCopy
128 now := time.Now()
129 m.UpdatedAt = &now
130 return nil
131 }
132 }
133 now := time.Now()
134 t.PipeMonitors = append(t.PipeMonitors, &db.PipeMonitor{
135 ID: fmt.Sprintf("monitor-%s-%s", userID, topic),
136 UserId: userID,
137 Topic: topic,
138 WindowDur: dur,
139 WindowEnd: winEndCopy,
140 CreatedAt: &now,
141 UpdatedAt: &now,
142 })
143 return nil
144}
145
146func (t *TestDB) UpdatePipeMonitorLastPing(userID, topic string, lastPing *time.Time) error {
147 t.mu.Lock()
148 defer t.mu.Unlock()
149 var lastPingCopy *time.Time
150 if lastPing != nil {
151 p := *lastPing
152 lastPingCopy = &p
153 }
154 for _, m := range t.PipeMonitors {
155 if m.UserId == userID && m.Topic == topic {
156 m.LastPing = lastPingCopy
157 now := time.Now()
158 m.UpdatedAt = &now
159 return nil
160 }
161 }
162 return fmt.Errorf("monitor not found")
163}
164
165func (t *TestDB) RemovePipeMonitor(userID, topic string) error {
166 t.mu.Lock()
167 defer t.mu.Unlock()
168 for i, m := range t.PipeMonitors {
169 if m.UserId == userID && m.Topic == topic {
170 t.PipeMonitors = append(t.PipeMonitors[:i], t.PipeMonitors[i+1:]...)
171 return nil
172 }
173 }
174 return fmt.Errorf("monitor not found")
175}
176
177func copyPipeMonitor(m *db.PipeMonitor) *db.PipeMonitor {
178 if m == nil {
179 return nil
180 }
181 cp := *m
182 if m.WindowEnd != nil {
183 w := *m.WindowEnd
184 cp.WindowEnd = &w
185 }
186 if m.LastPing != nil {
187 p := *m.LastPing
188 cp.LastPing = &p
189 }
190 if m.CreatedAt != nil {
191 c := *m.CreatedAt
192 cp.CreatedAt = &c
193 }
194 if m.UpdatedAt != nil {
195 u := *m.UpdatedAt
196 cp.UpdatedAt = &u
197 }
198 return &cp
199}
200
201func (t *TestDB) FindPipeMonitorByTopic(userID, topic string) (*db.PipeMonitor, error) {
202 t.mu.RLock()
203 defer t.mu.RUnlock()
204 for _, m := range t.PipeMonitors {
205 if m.UserId == userID && m.Topic == topic {
206 return copyPipeMonitor(m), nil
207 }
208 }
209 return nil, fmt.Errorf("monitor not found")
210}
211
212func (t *TestDB) FindPipeMonitorsByUser(userID string) ([]*db.PipeMonitor, error) {
213 t.mu.RLock()
214 defer t.mu.RUnlock()
215 var monitors []*db.PipeMonitor
216 for _, m := range t.PipeMonitors {
217 if m.UserId == userID {
218 monitors = append(monitors, copyPipeMonitor(m))
219 }
220 }
221 return monitors, nil
222}
223
224func (t *TestDB) InsertPipeMonitorHistory(monitorID string, windowDur time.Duration, windowEnd, lastPing *time.Time) error {
225 return nil
226}
227
228func (t *TestDB) FindPipeMonitorHistory(monitorID string, from, to time.Time) ([]*db.PipeMonitorHistory, error) {
229 return nil, nil
230}
231
232type TestSSHServer struct {
233 Cfg *Config
234 DBPool *TestDB
235 PipeHandler *CliHandler
236 Cancel context.CancelFunc
237}
238
239func NewTestSSHServer(t *testing.T) *TestSSHServer {
240 t.Helper()
241
242 opts := &slog.HandlerOptions{
243 AddSource: true,
244 Level: slog.LevelDebug,
245 }
246 logger := slog.New(slog.NewTextHandler(os.Stdout, opts))
247
248 dbpool := NewTestDB(logger)
249
250 cfg := &Config{
251 Domain: "pipe.test",
252 Port: "2225",
253 PortOverride: "2225",
254 Protocol: "ssh",
255 Logger: logger,
256 Space: "pipe",
257 }
258
259 ctx, cancel := context.WithCancel(context.Background())
260
261 pubsub := psub.NewMulticast(logger)
262 handler := &CliHandler{
263 Logger: logger,
264 DBPool: dbpool,
265 PubSub: pubsub,
266 Cfg: cfg,
267 Waiters: syncmap.New[string, []string](),
268 Access: syncmap.New[string, []string](),
269 }
270
271 sshAuth := shared.NewSshAuthHandler(dbpool, logger, "pipe")
272
273 server, err := pssh.NewSSHServerWithConfig(
274 ctx,
275 logger,
276 "pipe-ssh-test",
277 "localhost",
278 cfg.Port,
279 "",
280 "../../ssh_data/term_info_ed25519",
281 func(conn ssh.ConnMetadata, key ssh.PublicKey) (*ssh.Permissions, error) {
282 perms, _ := sshAuth.PubkeyAuthHandler(conn, key)
283 if perms == nil {
284 perms = &ssh.Permissions{
285 Extensions: map[string]string{
286 "pubkey": shared.KeyForKeyText(key),
287 },
288 }
289 }
290 return perms, nil
291 },
292 []pssh.SSHServerMiddleware{
293 Middleware(handler),
294 pssh.LogMiddleware(handler, dbpool),
295 },
296 nil,
297 nil,
298 )
299
300 if err != nil {
301 t.Fatalf("failed to create ssh server: %v", err)
302 }
303
304 go func() {
305 if err := server.ListenAndServe(); err != nil {
306 logger.Error("serve", "err", err.Error())
307 }
308 }()
309
310 waitForPort(t, cfg.Port, true)
311
312 return &TestSSHServer{
313 Cfg: cfg,
314 DBPool: dbpool,
315 PipeHandler: handler,
316 Cancel: cancel,
317 }
318}
319
320func (s *TestSSHServer) Shutdown() {
321 s.Cancel()
322 waitForPort(nil, s.Cfg.Port, false)
323}
324
325// waitForPort waits until the test server's port accepts connections, or
326// until it stops accepting them so the next test can listen on it.
327func waitForPort(t *testing.T, port string, open bool) {
328 deadline := time.Now().Add(5 * time.Second)
329 for time.Now().Before(deadline) {
330 conn, err := net.DialTimeout("tcp", "localhost:"+port, 100*time.Millisecond)
331 if err == nil {
332 _ = conn.Close()
333 }
334 if (err == nil) == open {
335 return
336 }
337 time.Sleep(5 * time.Millisecond)
338 }
339 if t != nil {
340 t.Fatalf("port %s open = %v, want %v", port, !open, open)
341 }
342}
343
344// waitForClients waits until n subscribers and pipes are on their topics,
345// so a message published next reaches them.
346func (s *TestSSHServer) waitForClients(t *testing.T, n int) {
347 t.Helper()
348 ps := s.PipeHandler.PubSub.(*psub.Multicast)
349 deadline := time.Now().Add(5 * time.Second)
350 for {
351 joined := 0
352 for range ps.GetSubs() {
353 joined++
354 }
355 for range ps.GetPipes() {
356 joined++
357 }
358 if joined == n {
359 return
360 }
361 if time.Now().After(deadline) {
362 t.Fatalf("%d subscribers and pipes, want %d", joined, n)
363 }
364 time.Sleep(5 * time.Millisecond)
365 }
366}
367
368type UserSSH struct {
369 username string
370 signer ssh.Signer
371 privateKey []byte
372}
373
374func GenerateUser(username string) UserSSH {
375 _, userKey, err := ed25519.GenerateKey(rand.Reader)
376 if err != nil {
377 panic(err)
378 }
379
380 b, err := ssh.MarshalPrivateKey(userKey, "")
381 if err != nil {
382 panic(err)
383 }
384
385 userSigner, err := ssh.NewSignerFromKey(userKey)
386 if err != nil {
387 panic(err)
388 }
389
390 return UserSSH{
391 username: username,
392 signer: userSigner,
393 privateKey: b.Bytes,
394 }
395}
396
397func (u UserSSH) PublicKey() string {
398 return shared.KeyForKeyText(u.signer.PublicKey())
399}
400
401func (u UserSSH) NewClient() (*ssh.Client, error) {
402 config := &ssh.ClientConfig{
403 User: u.username,
404 Auth: []ssh.AuthMethod{
405 ssh.PublicKeys(u.signer),
406 },
407 HostKeyCallback: ssh.InsecureIgnoreHostKey(),
408 }
409
410 return ssh.Dial("tcp", "localhost:2225", config)
411}
412
413func (u UserSSH) RunCommand(client *ssh.Client, cmd string) (string, error) {
414 session, err := client.NewSession()
415 if err != nil {
416 return "", err
417 }
418 defer func() { _ = session.Close() }()
419
420 stdoutPipe, err := session.StdoutPipe()
421 if err != nil {
422 return "", err
423 }
424
425 stderrPipe, err := session.StderrPipe()
426 if err != nil {
427 return "", err
428 }
429
430 if err := session.Start(cmd); err != nil {
431 return "", err
432 }
433
434 stdout := new(strings.Builder)
435 stderr := new(strings.Builder)
436 _, _ = io.Copy(stdout, stdoutPipe)
437 _, _ = io.Copy(stderr, stderrPipe)
438
439 _ = session.Wait()
440 return stdout.String() + stderr.String(), nil
441}
442
443func (u UserSSH) RunCommandWithStdin(client *ssh.Client, cmd string, stdin string) (string, error) {
444 session, err := client.NewSession()
445 if err != nil {
446 return "", err
447 }
448 defer func() { _ = session.Close() }()
449
450 stdinPipe, err := session.StdinPipe()
451 if err != nil {
452 return "", err
453 }
454
455 stdoutPipe, err := session.StdoutPipe()
456 if err != nil {
457 return "", err
458 }
459
460 if err := session.Start(cmd); err != nil {
461 return "", err
462 }
463
464 _, err = stdinPipe.Write([]byte(stdin))
465 if err != nil {
466 return "", err
467 }
468 _ = stdinPipe.Close()
469
470 buf := new(strings.Builder)
471 _, err = io.Copy(buf, stdoutPipe)
472 if err != nil {
473 return "", err
474 }
475
476 _ = session.Wait()
477 return buf.String(), nil
478}
479
480func RegisterUserWithServer(server *TestSSHServer, user UserSSH) {
481 dbUser := &db.User{
482 ID: user.username + "-id",
483 Name: user.username,
484 }
485 server.DBPool.AddUser(dbUser)
486 server.DBPool.AddPubkey(&db.PublicKey{
487 ID: user.username + "-pubkey-id",
488 UserID: dbUser.ID,
489 Key: user.PublicKey(),
490 })
491}
492
493func TestLs_UnauthenticatedUserDenied(t *testing.T) {
494 server := NewTestSSHServer(t)
495 defer server.Shutdown()
496
497 user := GenerateUser("anonymous")
498
499 client, err := user.NewClient()
500 if err != nil {
501 t.Fatalf("failed to connect: %v", err)
502 }
503 defer func() { _ = client.Close() }()
504
505 output, err := user.RunCommand(client, "ls")
506 if err != nil {
507 t.Logf("command error (expected): %v", err)
508 }
509
510 if !strings.Contains(output, "access denied") {
511 t.Errorf("expected 'access denied', got: %s", output)
512 }
513}
514
515func TestLs_AuthenticatedUser(t *testing.T) {
516 server := NewTestSSHServer(t)
517 defer server.Shutdown()
518
519 user := GenerateUser("alice")
520 RegisterUserWithServer(server, user)
521
522 client, err := user.NewClient()
523 if err != nil {
524 t.Fatalf("failed to connect: %v", err)
525 }
526 defer func() { _ = client.Close() }()
527
528 output, err := user.RunCommand(client, "ls")
529 if err != nil {
530 t.Logf("command completed with: %v", err)
531 }
532
533 if strings.Contains(output, "access denied") {
534 t.Errorf("authenticated user should not get access denied, got: %s", output)
535 }
536
537 if !strings.Contains(output, "no pubsub channels found") {
538 t.Errorf("expected 'no pubsub channels found' for empty state, got: %s", output)
539 }
540}
541
542func TestPubSub_BasicFlow(t *testing.T) {
543 server := NewTestSSHServer(t)
544 defer server.Shutdown()
545
546 user := GenerateUser("alice")
547 RegisterUserWithServer(server, user)
548
549 subClient, err := user.NewClient()
550 if err != nil {
551 t.Fatalf("failed to connect subscriber: %v", err)
552 }
553 defer func() { _ = subClient.Close() }()
554
555 pubClient, err := user.NewClient()
556 if err != nil {
557 t.Fatalf("failed to connect publisher: %v", err)
558 }
559 defer func() { _ = pubClient.Close() }()
560
561 subSession, err := subClient.NewSession()
562 if err != nil {
563 t.Fatalf("failed to create sub session: %v", err)
564 }
565 defer func() { _ = subSession.Close() }()
566
567 subStdout, err := subSession.StdoutPipe()
568 if err != nil {
569 t.Fatalf("failed to get sub stdout: %v", err)
570 }
571
572 if err := subSession.Start("sub testtopic -c"); err != nil {
573 t.Fatalf("failed to start sub: %v", err)
574 }
575
576 server.waitForClients(t, 1)
577
578 testMessage := "hello from pub"
579 _, err = user.RunCommandWithStdin(pubClient, "pub testtopic -c", testMessage)
580 if err != nil {
581 t.Logf("pub command completed: %v", err)
582 }
583
584 received := make([]byte, len(testMessage)+10)
585 n, err := subStdout.Read(received)
586 if err != nil && err != io.EOF {
587 t.Logf("read error: %v", err)
588 }
589
590 receivedStr := string(received[:n])
591 if !strings.Contains(receivedStr, testMessage) {
592 t.Errorf("subscriber did not receive message, got: %q, want: %q", receivedStr, testMessage)
593 }
594}
595
596func TestPubSub_PublicTopic(t *testing.T) {
597 server := NewTestSSHServer(t)
598 defer server.Shutdown()
599
600 alice := GenerateUser("alice")
601 bob := GenerateUser("bob")
602 RegisterUserWithServer(server, alice)
603 RegisterUserWithServer(server, bob)
604
605 subClient, err := bob.NewClient()
606 if err != nil {
607 t.Fatalf("failed to connect subscriber: %v", err)
608 }
609 defer func() { _ = subClient.Close() }()
610
611 pubClient, err := alice.NewClient()
612 if err != nil {
613 t.Fatalf("failed to connect publisher: %v", err)
614 }
615 defer func() { _ = pubClient.Close() }()
616
617 subSession, err := subClient.NewSession()
618 if err != nil {
619 t.Fatalf("failed to create sub session: %v", err)
620 }
621 defer func() { _ = subSession.Close() }()
622
623 subStdout, err := subSession.StdoutPipe()
624 if err != nil {
625 t.Fatalf("failed to get sub stdout: %v", err)
626 }
627
628 if err := subSession.Start("sub publictopic -p -c"); err != nil {
629 t.Fatalf("failed to start sub: %v", err)
630 }
631
632 server.waitForClients(t, 1)
633
634 testMessage := "public message"
635 _, err = alice.RunCommandWithStdin(pubClient, "pub publictopic -p -c", testMessage)
636 if err != nil {
637 t.Logf("pub command completed: %v", err)
638 }
639
640 received := make([]byte, len(testMessage)+10)
641 n, err := subStdout.Read(received)
642 if err != nil && err != io.EOF {
643 t.Logf("read error: %v", err)
644 }
645
646 receivedStr := string(received[:n])
647 if !strings.Contains(receivedStr, testMessage) {
648 t.Errorf("subscriber did not receive public message, got: %q, want: %q", receivedStr, testMessage)
649 }
650}
651
652func TestPipe_Bidirectional(t *testing.T) {
653 server := NewTestSSHServer(t)
654 defer server.Shutdown()
655
656 alice := GenerateUser("alice")
657 bob := GenerateUser("bob")
658 RegisterUserWithServer(server, alice)
659 RegisterUserWithServer(server, bob)
660
661 aliceClient, err := alice.NewClient()
662 if err != nil {
663 t.Fatalf("failed to connect alice: %v", err)
664 }
665 defer func() { _ = aliceClient.Close() }()
666
667 bobClient, err := bob.NewClient()
668 if err != nil {
669 t.Fatalf("failed to connect bob: %v", err)
670 }
671 defer func() { _ = bobClient.Close() }()
672
673 aliceSession, err := aliceClient.NewSession()
674 if err != nil {
675 t.Fatalf("failed to create alice session: %v", err)
676 }
677 defer func() { _ = aliceSession.Close() }()
678
679 aliceStdin, err := aliceSession.StdinPipe()
680 if err != nil {
681 t.Fatalf("failed to get alice stdin: %v", err)
682 }
683
684 aliceStdout, err := aliceSession.StdoutPipe()
685 if err != nil {
686 t.Fatalf("failed to get alice stdout: %v", err)
687 }
688
689 if err := aliceSession.Start("pipe pipetopic -p -c"); err != nil {
690 t.Fatalf("failed to start alice pipe: %v", err)
691 }
692
693 server.waitForClients(t, 1)
694
695 bobSession, err := bobClient.NewSession()
696 if err != nil {
697 t.Fatalf("failed to create bob session: %v", err)
698 }
699 defer func() { _ = bobSession.Close() }()
700
701 bobStdin, err := bobSession.StdinPipe()
702 if err != nil {
703 t.Fatalf("failed to get bob stdin: %v", err)
704 }
705
706 bobStdout, err := bobSession.StdoutPipe()
707 if err != nil {
708 t.Fatalf("failed to get bob stdout: %v", err)
709 }
710
711 if err := bobSession.Start("pipe pipetopic -p -c"); err != nil {
712 t.Fatalf("failed to start bob pipe: %v", err)
713 }
714
715 server.waitForClients(t, 2)
716
717 aliceMsg := "hello from alice\n"
718 _, err = aliceStdin.Write([]byte(aliceMsg))
719 if err != nil {
720 t.Fatalf("alice failed to write: %v", err)
721 }
722
723 bobReceived := make([]byte, 100)
724 n, err := bobStdout.Read(bobReceived)
725 if err != nil && err != io.EOF {
726 t.Logf("bob read error: %v", err)
727 }
728 if !strings.Contains(string(bobReceived[:n]), "hello from alice") {
729 t.Errorf("bob did not receive alice's message, got: %q", string(bobReceived[:n]))
730 }
731
732 bobMsg := "hello from bob\n"
733 _, err = bobStdin.Write([]byte(bobMsg))
734 if err != nil {
735 t.Fatalf("bob failed to write: %v", err)
736 }
737
738 aliceReceived := make([]byte, 100)
739 n, err = aliceStdout.Read(aliceReceived)
740 if err != nil && err != io.EOF {
741 t.Logf("alice read error: %v", err)
742 }
743 if !strings.Contains(string(aliceReceived[:n]), "hello from bob") {
744 t.Errorf("alice did not receive bob's message, got: %q", string(aliceReceived[:n]))
745 }
746
747 // When alice disconnects, bob's session should terminate cleanly without hanging
748 _ = aliceStdin.Close()
749 _ = aliceSession.Close()
750 _ = bobStdin.Close()
751
752 bobDone := make(chan error, 1)
753 go func() {
754 bobDone <- bobSession.Wait()
755 }()
756
757 select {
758 case <-bobDone:
759 // Bob's session terminated cleanly
760 case <-time.After(3 * time.Second):
761 t.Fatal("bob's pipe session hung after alice disconnected")
762 }
763}
764
765func TestPipe_AutoGeneratedTopic(t *testing.T) {
766 server := NewTestSSHServer(t)
767 defer server.Shutdown()
768
769 user := GenerateUser("alice")
770 RegisterUserWithServer(server, user)
771
772 client, err := user.NewClient()
773 if err != nil {
774 t.Fatalf("failed to connect: %v", err)
775 }
776 defer func() { _ = client.Close() }()
777
778 session, err := client.NewSession()
779 if err != nil {
780 t.Fatalf("failed to create session: %v", err)
781 }
782 defer func() { _ = session.Close() }()
783
784 stdout, err := session.StdoutPipe()
785 if err != nil {
786 t.Fatalf("failed to get stdout: %v", err)
787 }
788
789 if err := session.Start("pipe"); err != nil {
790 t.Fatalf("failed to start pipe: %v", err)
791 }
792
793 received := make([]byte, 200)
794 n, err := stdout.Read(received)
795 if err != nil && err != io.EOF {
796 t.Logf("read error: %v", err)
797 }
798
799 output := string(received[:n])
800 if !strings.Contains(output, "subscribe to this topic") {
801 t.Errorf("expected topic subscription instructions, got: %q", output)
802 }
803}
804
805func TestAccessControl_AllowedUserViaFullPath(t *testing.T) {
806 server := NewTestSSHServer(t)
807 defer server.Shutdown()
808
809 alice := GenerateUser("alice")
810 bob := GenerateUser("bob")
811 RegisterUserWithServer(server, alice)
812 RegisterUserWithServer(server, bob)
813
814 aliceClient, err := alice.NewClient()
815 if err != nil {
816 t.Fatalf("failed to connect alice: %v", err)
817 }
818 defer func() { _ = aliceClient.Close() }()
819
820 aliceSession, err := aliceClient.NewSession()
821 if err != nil {
822 t.Fatalf("failed to create alice session: %v", err)
823 }
824 defer func() { _ = aliceSession.Close() }()
825
826 aliceStdout, err := aliceSession.StdoutPipe()
827 if err != nil {
828 t.Fatalf("failed to get alice stdout: %v", err)
829 }
830
831 if err := aliceSession.Start("sub sharedtopic -a alice,bob -c"); err != nil {
832 t.Fatalf("failed to start alice sub: %v", err)
833 }
834
835 server.waitForClients(t, 1)
836
837 bobClient, err := bob.NewClient()
838 if err != nil {
839 t.Fatalf("failed to connect bob: %v", err)
840 }
841 defer func() { _ = bobClient.Close() }()
842
843 _, err = bob.RunCommandWithStdin(bobClient, "pub alice/sharedtopic -c", "bob allowed")
844 if err != nil {
845 t.Logf("bob pub completed: %v", err)
846 }
847
848 aliceReceived := make([]byte, 100)
849 n, _ := aliceStdout.Read(aliceReceived)
850
851 if !strings.Contains(string(aliceReceived[:n]), "bob allowed") {
852 t.Errorf("alice should receive bob's message on shared topic, got: %q", string(aliceReceived[:n]))
853 }
854}
855
856func TestPubSub_BlockingWaitsForSubscriber(t *testing.T) {
857 server := NewTestSSHServer(t)
858 defer server.Shutdown()
859
860 user := GenerateUser("alice")
861 RegisterUserWithServer(server, user)
862
863 pubClient, err := user.NewClient()
864 if err != nil {
865 t.Fatalf("failed to connect publisher: %v", err)
866 }
867 defer func() { _ = pubClient.Close() }()
868
869 subClient, err := user.NewClient()
870 if err != nil {
871 t.Fatalf("failed to connect subscriber: %v", err)
872 }
873 defer func() { _ = subClient.Close() }()
874
875 pubSession, err := pubClient.NewSession()
876 if err != nil {
877 t.Fatalf("failed to create pub session: %v", err)
878 }
879 defer func() { _ = pubSession.Close() }()
880
881 pubStdin, err := pubSession.StdinPipe()
882 if err != nil {
883 t.Fatalf("failed to get pub stdin: %v", err)
884 }
885
886 pubStdout, err := pubSession.StdoutPipe()
887 if err != nil {
888 t.Fatalf("failed to get pub stdout: %v", err)
889 }
890
891 // Start publisher with blocking enabled (default -b=true)
892 // Publisher should wait for subscriber
893 if err := pubSession.Start("pub blockingtopic"); err != nil {
894 t.Fatalf("failed to start pub: %v", err)
895 }
896
897 // Read output until we see "waiting" message or timeout
898 // Need to read in a loop because Read() may return partial data
899 var output string
900 readDone := make(chan struct{})
901 go func() {
902 buf := make([]byte, 1024)
903 for {
904 n, err := pubStdout.Read(buf)
905 if n > 0 {
906 output += string(buf[:n])
907 if strings.Contains(output, "waiting") {
908 close(readDone)
909 return
910 }
911 }
912 if err != nil {
913 close(readDone)
914 return
915 }
916 }
917 }()
918
919 select {
920 case <-readDone:
921 case <-time.After(2 * time.Second):
922 t.Fatalf("timeout waiting for 'waiting' message, got: %q", output)
923 }
924
925 if !strings.Contains(output, "waiting") {
926 t.Errorf("expected 'waiting' message for blocking pub, got: %q", output)
927 }
928
929 // Now start subscriber - this should unblock the publisher
930 subSession, err := subClient.NewSession()
931 if err != nil {
932 t.Fatalf("failed to create sub session: %v", err)
933 }
934 defer func() { _ = subSession.Close() }()
935
936 subStdout, err := subSession.StdoutPipe()
937 if err != nil {
938 t.Fatalf("failed to get sub stdout: %v", err)
939 }
940
941 if err := subSession.Start("sub blockingtopic -c"); err != nil {
942 t.Fatalf("failed to start sub: %v", err)
943 }
944
945 server.waitForClients(t, 1)
946
947 // Now send the message
948 testMessage := "blocking message"
949 _, err = pubStdin.Write([]byte(testMessage))
950 if err != nil {
951 t.Fatalf("failed to write message: %v", err)
952 }
953 _ = pubStdin.Close()
954
955 // Subscriber should receive the message
956 received := make([]byte, 100)
957 nRead, err := subStdout.Read(received)
958 if err != nil && err != io.EOF {
959 t.Logf("read error: %v", err)
960 }
961
962 if !strings.Contains(string(received[:nRead]), testMessage) {
963 t.Errorf("subscriber did not receive blocking message, got: %q, want: %q", string(received[:nRead]), testMessage)
964 }
965}
966
967func TestPubSub_NonBlockingDoesNotWait(t *testing.T) {
968 server := NewTestSSHServer(t)
969 defer server.Shutdown()
970
971 user := GenerateUser("alice")
972 RegisterUserWithServer(server, user)
973
974 client, err := user.NewClient()
975 if err != nil {
976 t.Fatalf("failed to connect: %v", err)
977 }
978 defer func() { _ = client.Close() }()
979
980 // Publish with -b=false (non-blocking) and no subscriber
981 // Should complete immediately without waiting
982 done := make(chan struct{})
983 var output string
984 var cmdErr error
985
986 go func() {
987 output, cmdErr = user.RunCommandWithStdin(client, "pub nonblockingtopic -b=false -c", "non-blocking message")
988 close(done)
989 }()
990
991 select {
992 case <-done:
993 // Command completed - this is expected for non-blocking
994 if cmdErr != nil {
995 t.Logf("non-blocking pub completed with: %v", cmdErr)
996 }
997 t.Logf("non-blocking pub output: %q", output)
998 case <-time.After(2 * time.Second):
999 t.Errorf("non-blocking pub should complete immediately, but it blocked")
1000 }
1001}
1002
1003func TestPubSub_BlockingTimeout(t *testing.T) {
1004 server := NewTestSSHServer(t)
1005 defer server.Shutdown()
1006
1007 user := GenerateUser("alice")
1008 RegisterUserWithServer(server, user)
1009
1010 client, err := user.NewClient()
1011 if err != nil {
1012 t.Fatalf("failed to connect: %v", err)
1013 }
1014 defer func() { _ = client.Close() }()
1015
1016 // Publish with blocking and short timeout, no subscriber
1017 // Should timeout after the specified duration
1018 done := make(chan struct{})
1019 var output string
1020
1021 go func() {
1022 output, _ = user.RunCommandWithStdin(client, "pub timeouttopic -b=true -t=500ms", "timeout message")
1023 close(done)
1024 }()
1025
1026 select {
1027 case <-done:
1028 // Command completed due to timeout
1029 if !strings.Contains(output, "timeout") && !strings.Contains(output, "waiting") {
1030 t.Logf("blocking pub with timeout output: %q", output)
1031 }
1032 case <-time.After(3 * time.Second):
1033 t.Errorf("blocking pub with timeout should have timed out after 500ms")
1034 }
1035}
1036
1037func TestSub_WaitsForPublisher(t *testing.T) {
1038 server := NewTestSSHServer(t)
1039 defer server.Shutdown()
1040
1041 user := GenerateUser("alice")
1042 RegisterUserWithServer(server, user)
1043
1044 subClient, err := user.NewClient()
1045 if err != nil {
1046 t.Fatalf("failed to connect subscriber: %v", err)
1047 }
1048 defer func() { _ = subClient.Close() }()
1049
1050 pubClient, err := user.NewClient()
1051 if err != nil {
1052 t.Fatalf("failed to connect publisher: %v", err)
1053 }
1054 defer func() { _ = pubClient.Close() }()
1055
1056 // Start subscriber first - it should wait for publisher
1057 subSession, err := subClient.NewSession()
1058 if err != nil {
1059 t.Fatalf("failed to create sub session: %v", err)
1060 }
1061 defer func() { _ = subSession.Close() }()
1062
1063 subStdout, err := subSession.StdoutPipe()
1064 if err != nil {
1065 t.Fatalf("failed to get sub stdout: %v", err)
1066 }
1067
1068 if err := subSession.Start("sub waitfortopic -c"); err != nil {
1069 t.Fatalf("failed to start sub: %v", err)
1070 }
1071
1072 // Subscriber is now waiting - give it a moment
1073 server.waitForClients(t, 1)
1074
1075 // Now publish - subscriber should receive it
1076 testMessage := "delayed publish"
1077 _, err = user.RunCommandWithStdin(pubClient, "pub waitfortopic -c", testMessage)
1078 if err != nil {
1079 t.Logf("pub completed: %v", err)
1080 }
1081
1082 received := make([]byte, 100)
1083 n, err := subStdout.Read(received)
1084 if err != nil && err != io.EOF {
1085 t.Logf("read error: %v", err)
1086 }
1087
1088 if !strings.Contains(string(received[:n]), testMessage) {
1089 t.Errorf("subscriber waiting for publisher did not receive message, got: %q, want: %q", string(received[:n]), testMessage)
1090 }
1091}
1092
1093func TestSub_KeepAliveReceivesMultipleMessages(t *testing.T) {
1094 server := NewTestSSHServer(t)
1095 defer server.Shutdown()
1096
1097 user := GenerateUser("alice")
1098 RegisterUserWithServer(server, user)
1099
1100 subClient, err := user.NewClient()
1101 if err != nil {
1102 t.Fatalf("failed to connect subscriber: %v", err)
1103 }
1104 defer func() { _ = subClient.Close() }()
1105
1106 pubClient1, err := user.NewClient()
1107 if err != nil {
1108 t.Fatalf("failed to connect publisher 1: %v", err)
1109 }
1110 defer func() { _ = pubClient1.Close() }()
1111
1112 pubClient2, err := user.NewClient()
1113 if err != nil {
1114 t.Fatalf("failed to connect publisher 2: %v", err)
1115 }
1116 defer func() { _ = pubClient2.Close() }()
1117
1118 // Start subscriber with keepAlive (-k) flag
1119 subSession, err := subClient.NewSession()
1120 if err != nil {
1121 t.Fatalf("failed to create sub session: %v", err)
1122 }
1123 defer func() { _ = subSession.Close() }()
1124
1125 subStdout, err := subSession.StdoutPipe()
1126 if err != nil {
1127 t.Fatalf("failed to get sub stdout: %v", err)
1128 }
1129
1130 if err := subSession.Start("sub keepalivetopic -k -c"); err != nil {
1131 t.Fatalf("failed to start sub: %v", err)
1132 }
1133
1134 server.waitForClients(t, 1)
1135
1136 // Send first message
1137 msg1 := "first message\n"
1138 _, err = user.RunCommandWithStdin(pubClient1, "pub keepalivetopic -c", msg1)
1139 if err != nil {
1140 t.Logf("pub 1 completed: %v", err)
1141 }
1142
1143 received1 := make([]byte, 100)
1144 n1, _ := subStdout.Read(received1)
1145 if !strings.Contains(string(received1[:n1]), "first message") {
1146 t.Errorf("subscriber did not receive first message, got: %q", string(received1[:n1]))
1147 }
1148
1149 // Send second message - subscriber with keepAlive should still receive it
1150 msg2 := "second message\n"
1151 _, err = user.RunCommandWithStdin(pubClient2, "pub keepalivetopic -c", msg2)
1152 if err != nil {
1153 t.Logf("pub 2 completed: %v", err)
1154 }
1155
1156 received2 := make([]byte, 100)
1157 n2, _ := subStdout.Read(received2)
1158 if !strings.Contains(string(received2[:n2]), "second message") {
1159 t.Errorf("subscriber with keepAlive did not receive second message, got: %q", string(received2[:n2]))
1160 }
1161}
1162
1163func TestSub_WithoutKeepAliveExitsAfterPublisher(t *testing.T) {
1164 server := NewTestSSHServer(t)
1165 defer server.Shutdown()
1166
1167 user := GenerateUser("alice")
1168 RegisterUserWithServer(server, user)
1169
1170 subClient, err := user.NewClient()
1171 if err != nil {
1172 t.Fatalf("failed to connect subscriber: %v", err)
1173 }
1174 defer func() { _ = subClient.Close() }()
1175
1176 pubClient, err := user.NewClient()
1177 if err != nil {
1178 t.Fatalf("failed to connect publisher: %v", err)
1179 }
1180 defer func() { _ = pubClient.Close() }()
1181
1182 // Start subscriber without keepAlive
1183 subSession, err := subClient.NewSession()
1184 if err != nil {
1185 t.Fatalf("failed to create sub session: %v", err)
1186 }
1187
1188 subStdout, err := subSession.StdoutPipe()
1189 if err != nil {
1190 t.Fatalf("failed to get sub stdout: %v", err)
1191 }
1192
1193 if err := subSession.Start("sub exitaftertopic -c"); err != nil {
1194 t.Fatalf("failed to start sub: %v", err)
1195 }
1196
1197 server.waitForClients(t, 1)
1198
1199 // Publish a message
1200 testMessage := "single message"
1201 _, err = user.RunCommandWithStdin(pubClient, "pub exitaftertopic -c", testMessage)
1202 if err != nil {
1203 t.Logf("pub completed: %v", err)
1204 }
1205
1206 // Read the message
1207 received := make([]byte, 100)
1208 n, _ := subStdout.Read(received)
1209 if !strings.Contains(string(received[:n]), testMessage) {
1210 t.Errorf("subscriber did not receive message, got: %q", string(received[:n]))
1211 }
1212
1213 // Subscriber session should exit after publisher disconnects
1214 done := make(chan error)
1215 go func() {
1216 done <- subSession.Wait()
1217 }()
1218
1219 select {
1220 case err := <-done:
1221 // Session ended as expected
1222 t.Logf("subscriber session ended: %v", err)
1223 case <-time.After(2 * time.Second):
1224 t.Errorf("subscriber without keepAlive should have exited after publisher disconnected")
1225 _ = subSession.Close()
1226 }
1227}
1228
1229func TestPub_EmptyMessage(t *testing.T) {
1230 server := NewTestSSHServer(t)
1231 defer server.Shutdown()
1232
1233 user := GenerateUser("alice")
1234 RegisterUserWithServer(server, user)
1235
1236 subClient, err := user.NewClient()
1237 if err != nil {
1238 t.Fatalf("failed to connect subscriber: %v", err)
1239 }
1240 defer func() { _ = subClient.Close() }()
1241
1242 pubClient, err := user.NewClient()
1243 if err != nil {
1244 t.Fatalf("failed to connect publisher: %v", err)
1245 }
1246 defer func() { _ = pubClient.Close() }()
1247
1248 // Start subscriber
1249 subSession, err := subClient.NewSession()
1250 if err != nil {
1251 t.Fatalf("failed to create sub session: %v", err)
1252 }
1253 defer func() { _ = subSession.Close() }()
1254
1255 subStdout, err := subSession.StdoutPipe()
1256 if err != nil {
1257 t.Fatalf("failed to get sub stdout: %v", err)
1258 }
1259
1260 if err := subSession.Start("sub emptytopic -c"); err != nil {
1261 t.Fatalf("failed to start sub: %v", err)
1262 }
1263
1264 server.waitForClients(t, 1)
1265
1266 // Publish with -e flag (empty message) - should not require stdin
1267 output, err := user.RunCommand(pubClient, "pub emptytopic -e -c")
1268 if err != nil {
1269 t.Logf("pub -e completed: %v, output: %s", err, output)
1270 }
1271
1272 // Subscriber should receive something (even if empty/minimal)
1273 // The -e flag sends a 1-byte buffer
1274 received := make([]byte, 10)
1275 n, err := subStdout.Read(received)
1276 if err != nil && err != io.EOF {
1277 t.Logf("read result: n=%d, err=%v", n, err)
1278 }
1279
1280 // With -e flag, we expect to receive at least 1 byte
1281 if n < 1 {
1282 t.Errorf("subscriber should receive empty message signal, got %d bytes", n)
1283 }
1284}
1285
1286func TestPipe_AccessControl(t *testing.T) {
1287 server := NewTestSSHServer(t)
1288 defer server.Shutdown()
1289
1290 alice := GenerateUser("alice")
1291 bob := GenerateUser("bob")
1292 RegisterUserWithServer(server, alice)
1293 RegisterUserWithServer(server, bob)
1294
1295 aliceClient, err := alice.NewClient()
1296 if err != nil {
1297 t.Fatalf("failed to connect alice: %v", err)
1298 }
1299 defer func() { _ = aliceClient.Close() }()
1300
1301 bobClient, err := bob.NewClient()
1302 if err != nil {
1303 t.Fatalf("failed to connect bob: %v", err)
1304 }
1305 defer func() { _ = bobClient.Close() }()
1306
1307 // Alice creates a pipe with access control allowing bob
1308 aliceSession, err := aliceClient.NewSession()
1309 if err != nil {
1310 t.Fatalf("failed to create alice session: %v", err)
1311 }
1312 defer func() { _ = aliceSession.Close() }()
1313
1314 aliceStdin, err := aliceSession.StdinPipe()
1315 if err != nil {
1316 t.Fatalf("failed to get alice stdin: %v", err)
1317 }
1318
1319 aliceStdout, err := aliceSession.StdoutPipe()
1320 if err != nil {
1321 t.Fatalf("failed to get alice stdout: %v", err)
1322 }
1323
1324 if err := aliceSession.Start("pipe accesspipe -a alice,bob -c"); err != nil {
1325 t.Fatalf("failed to start alice pipe: %v", err)
1326 }
1327
1328 server.waitForClients(t, 1)
1329
1330 // Bob joins the pipe using alice's namespace
1331 bobSession, err := bobClient.NewSession()
1332 if err != nil {
1333 t.Fatalf("failed to create bob session: %v", err)
1334 }
1335 defer func() { _ = bobSession.Close() }()
1336
1337 bobStdin, err := bobSession.StdinPipe()
1338 if err != nil {
1339 t.Fatalf("failed to get bob stdin: %v", err)
1340 }
1341
1342 bobStdout, err := bobSession.StdoutPipe()
1343 if err != nil {
1344 t.Fatalf("failed to get bob stdout: %v", err)
1345 }
1346
1347 if err := bobSession.Start("pipe alice/accesspipe -c"); err != nil {
1348 t.Fatalf("failed to start bob pipe: %v", err)
1349 }
1350
1351 server.waitForClients(t, 2)
1352
1353 // Alice sends message to bob
1354 aliceMsg := "hello bob\n"
1355 _, err = aliceStdin.Write([]byte(aliceMsg))
1356 if err != nil {
1357 t.Fatalf("alice failed to write: %v", err)
1358 }
1359
1360 bobReceived := make([]byte, 100)
1361 n, _ := bobStdout.Read(bobReceived)
1362 if !strings.Contains(string(bobReceived[:n]), "hello bob") {
1363 t.Errorf("bob did not receive alice's message, got: %q", string(bobReceived[:n]))
1364 }
1365
1366 // Bob sends message to alice
1367 bobMsg := "hello alice\n"
1368 _, err = bobStdin.Write([]byte(bobMsg))
1369 if err != nil {
1370 t.Fatalf("bob failed to write: %v", err)
1371 }
1372
1373 aliceReceived := make([]byte, 100)
1374 n, _ = aliceStdout.Read(aliceReceived)
1375 if !strings.Contains(string(aliceReceived[:n]), "hello alice") {
1376 t.Errorf("alice did not receive bob's message, got: %q", string(aliceReceived[:n]))
1377 }
1378}
1379
1380func TestPipe_Replay(t *testing.T) {
1381 server := NewTestSSHServer(t)
1382 defer server.Shutdown()
1383
1384 user := GenerateUser("alice")
1385 RegisterUserWithServer(server, user)
1386
1387 client, err := user.NewClient()
1388 if err != nil {
1389 t.Fatalf("failed to connect: %v", err)
1390 }
1391 defer func() { _ = client.Close() }()
1392
1393 // Start pipe with replay flag (-r)
1394 session, err := client.NewSession()
1395 if err != nil {
1396 t.Fatalf("failed to create session: %v", err)
1397 }
1398 defer func() { _ = session.Close() }()
1399
1400 stdin, err := session.StdinPipe()
1401 if err != nil {
1402 t.Fatalf("failed to get stdin: %v", err)
1403 }
1404
1405 stdout, err := session.StdoutPipe()
1406 if err != nil {
1407 t.Fatalf("failed to get stdout: %v", err)
1408 }
1409
1410 if err := session.Start("pipe replaytopic -r -c"); err != nil {
1411 t.Fatalf("failed to start pipe: %v", err)
1412 }
1413
1414 server.waitForClients(t, 1)
1415
1416 // Send a message - with -r flag, should receive it back
1417 testMsg := "echo back\n"
1418 _, err = stdin.Write([]byte(testMsg))
1419 if err != nil {
1420 t.Fatalf("failed to write: %v", err)
1421 }
1422
1423 received := make([]byte, 100)
1424 n, err := stdout.Read(received)
1425 if err != nil && err != io.EOF {
1426 t.Logf("read error: %v", err)
1427 }
1428
1429 if !strings.Contains(string(received[:n]), "echo back") {
1430 t.Errorf("with -r flag, sender should receive own message back, got: %q", string(received[:n]))
1431 }
1432}
1433
1434func TestAccessControl_UnauthorizedUserDenied(t *testing.T) {
1435 server := NewTestSSHServer(t)
1436 defer server.Shutdown()
1437
1438 alice := GenerateUser("alice")
1439 bob := GenerateUser("bob")
1440 charlie := GenerateUser("charlie")
1441 RegisterUserWithServer(server, alice)
1442 RegisterUserWithServer(server, bob)
1443 RegisterUserWithServer(server, charlie)
1444
1445 aliceClient, err := alice.NewClient()
1446 if err != nil {
1447 t.Fatalf("failed to connect alice: %v", err)
1448 }
1449 defer func() { _ = aliceClient.Close() }()
1450
1451 charlieClient, err := charlie.NewClient()
1452 if err != nil {
1453 t.Fatalf("failed to connect charlie: %v", err)
1454 }
1455 defer func() { _ = charlieClient.Close() }()
1456
1457 // Alice creates a topic with access only for alice and bob (not charlie)
1458 aliceSession, err := aliceClient.NewSession()
1459 if err != nil {
1460 t.Fatalf("failed to create alice session: %v", err)
1461 }
1462 defer func() { _ = aliceSession.Close() }()
1463
1464 if err := aliceSession.Start("sub restrictedtopic -a alice,bob -c"); err != nil {
1465 t.Fatalf("failed to start alice sub: %v", err)
1466 }
1467
1468 server.waitForClients(t, 1)
1469
1470 // Charlie tries to publish to alice's restricted topic - should be denied
1471 output, err := charlie.RunCommandWithStdin(charlieClient, "pub alice/restrictedtopic -c", "unauthorized message")
1472 if err != nil {
1473 t.Logf("charlie pub completed with error (expected): %v", err)
1474 }
1475
1476 // Charlie should get access denied or the message should not be delivered
1477 if strings.Contains(output, "access denied") {
1478 t.Logf("charlie correctly received access denied")
1479 } else {
1480 t.Logf("charlie output: %q (access control may work differently)", output)
1481 }
1482}
1483
1484func TestPubSub_MultipleSubscribers(t *testing.T) {
1485 server := NewTestSSHServer(t)
1486 defer server.Shutdown()
1487
1488 user := GenerateUser("alice")
1489 RegisterUserWithServer(server, user)
1490
1491 pubClient, err := user.NewClient()
1492 if err != nil {
1493 t.Fatalf("failed to connect publisher: %v", err)
1494 }
1495 defer func() { _ = pubClient.Close() }()
1496
1497 sub1Client, err := user.NewClient()
1498 if err != nil {
1499 t.Fatalf("failed to connect subscriber 1: %v", err)
1500 }
1501 defer func() { _ = sub1Client.Close() }()
1502
1503 sub2Client, err := user.NewClient()
1504 if err != nil {
1505 t.Fatalf("failed to connect subscriber 2: %v", err)
1506 }
1507 defer func() { _ = sub2Client.Close() }()
1508
1509 sub3Client, err := user.NewClient()
1510 if err != nil {
1511 t.Fatalf("failed to connect subscriber 3: %v", err)
1512 }
1513 defer func() { _ = sub3Client.Close() }()
1514
1515 // Start three subscribers
1516 sub1Session, err := sub1Client.NewSession()
1517 if err != nil {
1518 t.Fatalf("failed to create sub1 session: %v", err)
1519 }
1520 defer func() { _ = sub1Session.Close() }()
1521
1522 sub1Stdout, err := sub1Session.StdoutPipe()
1523 if err != nil {
1524 t.Fatalf("failed to get sub1 stdout: %v", err)
1525 }
1526
1527 if err := sub1Session.Start("sub fanout -c"); err != nil {
1528 t.Fatalf("failed to start sub1: %v", err)
1529 }
1530
1531 sub2Session, err := sub2Client.NewSession()
1532 if err != nil {
1533 t.Fatalf("failed to create sub2 session: %v", err)
1534 }
1535 defer func() { _ = sub2Session.Close() }()
1536
1537 sub2Stdout, err := sub2Session.StdoutPipe()
1538 if err != nil {
1539 t.Fatalf("failed to get sub2 stdout: %v", err)
1540 }
1541
1542 if err := sub2Session.Start("sub fanout -c"); err != nil {
1543 t.Fatalf("failed to start sub2: %v", err)
1544 }
1545
1546 sub3Session, err := sub3Client.NewSession()
1547 if err != nil {
1548 t.Fatalf("failed to create sub3 session: %v", err)
1549 }
1550 defer func() { _ = sub3Session.Close() }()
1551
1552 sub3Stdout, err := sub3Session.StdoutPipe()
1553 if err != nil {
1554 t.Fatalf("failed to get sub3 stdout: %v", err)
1555 }
1556
1557 if err := sub3Session.Start("sub fanout -c"); err != nil {
1558 t.Fatalf("failed to start sub3: %v", err)
1559 }
1560
1561 server.waitForClients(t, 3)
1562
1563 // Publish a single message
1564 testMessage := "broadcast message"
1565 _, err = user.RunCommandWithStdin(pubClient, "pub fanout -c", testMessage)
1566 if err != nil {
1567 t.Logf("pub completed: %v", err)
1568 }
1569
1570 // All three subscribers should receive the message
1571 received1 := make([]byte, 100)
1572 n1, _ := sub1Stdout.Read(received1)
1573 if !strings.Contains(string(received1[:n1]), testMessage) {
1574 t.Errorf("subscriber 1 did not receive message, got: %q", string(received1[:n1]))
1575 }
1576
1577 received2 := make([]byte, 100)
1578 n2, _ := sub2Stdout.Read(received2)
1579 if !strings.Contains(string(received2[:n2]), testMessage) {
1580 t.Errorf("subscriber 2 did not receive message, got: %q", string(received2[:n2]))
1581 }
1582
1583 received3 := make([]byte, 100)
1584 n3, _ := sub3Stdout.Read(received3)
1585 if !strings.Contains(string(received3[:n3]), testMessage) {
1586 t.Errorf("subscriber 3 did not receive message, got: %q", string(received3[:n3]))
1587 }
1588}
1589
1590// Monitor CLI Tests
1591
1592func TestMonitor_UnauthenticatedUserDenied(t *testing.T) {
1593 server := NewTestSSHServer(t)
1594 defer server.Shutdown()
1595
1596 user := GenerateUser("anonymous")
1597
1598 client, err := user.NewClient()
1599 if err != nil {
1600 t.Fatalf("failed to connect: %v", err)
1601 }
1602 defer func() { _ = client.Close() }()
1603
1604 output, err := user.RunCommand(client, "monitor my-service 1h")
1605 if err != nil {
1606 t.Logf("command error (expected): %v", err)
1607 }
1608
1609 if !strings.Contains(output, "access denied") {
1610 t.Errorf("expected 'access denied', got: %s", output)
1611 }
1612}
1613
1614func TestMonitor_CreateMonitor(t *testing.T) {
1615 server := NewTestSSHServer(t)
1616 defer server.Shutdown()
1617
1618 user := GenerateUser("alice")
1619 RegisterUserWithServer(server, user)
1620
1621 client, err := user.NewClient()
1622 if err != nil {
1623 t.Fatalf("failed to connect: %v", err)
1624 }
1625 defer func() { _ = client.Close() }()
1626
1627 output, err := user.RunCommand(client, "monitor pico-uptime 24h")
1628 if err != nil {
1629 t.Logf("command completed: %v", err)
1630 }
1631
1632 if strings.Contains(output, "access denied") {
1633 t.Errorf("authenticated user should not get access denied, got: %s", output)
1634 }
1635
1636 // Verify monitor was created in DB (topic is stored with user prefix)
1637 monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/pico-uptime")
1638 if err != nil {
1639 t.Fatalf("monitor should exist in DB: %v", err)
1640 }
1641
1642 if monitor.WindowDur != 24*time.Hour {
1643 t.Errorf("expected window duration 24h, got: %v", monitor.WindowDur)
1644 }
1645
1646 if !strings.Contains(output, "alice/pico-uptime") || !strings.Contains(output, "24h") {
1647 t.Errorf("output should confirm monitor creation, got: %s", output)
1648 }
1649}
1650
1651func TestMonitor_UpdateMonitor(t *testing.T) {
1652 server := NewTestSSHServer(t)
1653 defer server.Shutdown()
1654
1655 user := GenerateUser("alice")
1656 RegisterUserWithServer(server, user)
1657
1658 client, err := user.NewClient()
1659 if err != nil {
1660 t.Fatalf("failed to connect: %v", err)
1661 }
1662 defer func() { _ = client.Close() }()
1663
1664 // Create initial monitor
1665 _, err = user.RunCommand(client, "monitor my-cron 1h")
1666 if err != nil {
1667 t.Logf("create command completed: %v", err)
1668 }
1669
1670 // Upsert with new duration
1671 output, err := user.RunCommand(client, "monitor my-cron 6h")
1672 if err != nil {
1673 t.Logf("update command completed: %v", err)
1674 }
1675
1676 // Verify monitor was updated (topic is stored with user prefix)
1677 monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/my-cron")
1678 if err != nil {
1679 t.Fatalf("monitor should exist in DB: %v", err)
1680 }
1681
1682 if monitor.WindowDur != 6*time.Hour {
1683 t.Errorf("expected window duration 6h after update, got: %v", monitor.WindowDur)
1684 }
1685
1686 if !strings.Contains(output, "6h") {
1687 t.Errorf("output should confirm updated duration, got: %s", output)
1688 }
1689}
1690
1691func TestMonitor_DeleteMonitor(t *testing.T) {
1692 server := NewTestSSHServer(t)
1693 defer server.Shutdown()
1694
1695 user := GenerateUser("alice")
1696 RegisterUserWithServer(server, user)
1697
1698 client, err := user.NewClient()
1699 if err != nil {
1700 t.Fatalf("failed to connect: %v", err)
1701 }
1702 defer func() { _ = client.Close() }()
1703
1704 // Create monitor first
1705 _, err = user.RunCommand(client, "monitor to-delete 1h")
1706 if err != nil {
1707 t.Logf("create command completed: %v", err)
1708 }
1709
1710 // Verify it exists (topic is stored with user prefix)
1711 _, err = server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/to-delete")
1712 if err != nil {
1713 t.Fatalf("monitor should exist before deletion: %v", err)
1714 }
1715
1716 // Delete it
1717 output, err := user.RunCommand(client, "monitor to-delete -d")
1718 if err != nil {
1719 t.Logf("delete command completed: %v", err)
1720 }
1721
1722 // Verify it's gone (topic is stored with user prefix)
1723 _, err = server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/to-delete")
1724 if err == nil {
1725 t.Errorf("monitor should be deleted from DB")
1726 }
1727
1728 if !strings.Contains(output, "deleted") && !strings.Contains(output, "removed") {
1729 t.Logf("output should confirm deletion, got: %s", output)
1730 }
1731}
1732
1733func TestMonitor_InvalidDuration(t *testing.T) {
1734 server := NewTestSSHServer(t)
1735 defer server.Shutdown()
1736
1737 user := GenerateUser("alice")
1738 RegisterUserWithServer(server, user)
1739
1740 client, err := user.NewClient()
1741 if err != nil {
1742 t.Fatalf("failed to connect: %v", err)
1743 }
1744 defer func() { _ = client.Close() }()
1745
1746 output, err := user.RunCommand(client, "monitor my-service invaliduration")
1747 if err != nil {
1748 t.Logf("command error (expected): %v", err)
1749 }
1750
1751 if !strings.Contains(output, "invalid") && !strings.Contains(output, "duration") && !strings.Contains(output, "error") {
1752 t.Errorf("expected error about invalid duration, got: %s", output)
1753 }
1754}
1755
1756func TestMonitor_MissingTopic(t *testing.T) {
1757 server := NewTestSSHServer(t)
1758 defer server.Shutdown()
1759
1760 user := GenerateUser("alice")
1761 RegisterUserWithServer(server, user)
1762
1763 client, err := user.NewClient()
1764 if err != nil {
1765 t.Fatalf("failed to connect: %v", err)
1766 }
1767 defer func() { _ = client.Close() }()
1768
1769 output, err := user.RunCommand(client, "monitor")
1770 if err != nil {
1771 t.Logf("command error (expected): %v", err)
1772 }
1773
1774 // Should show usage or error about missing topic
1775 if !strings.Contains(output, "Usage") && !strings.Contains(output, "topic") && !strings.Contains(output, "error") {
1776 t.Errorf("expected usage info or error about missing topic, got: %s", output)
1777 }
1778}
1779
1780// Status CLI Tests
1781
1782func TestStatus_UnauthenticatedUserDenied(t *testing.T) {
1783 server := NewTestSSHServer(t)
1784 defer server.Shutdown()
1785
1786 user := GenerateUser("anonymous")
1787
1788 client, err := user.NewClient()
1789 if err != nil {
1790 t.Fatalf("failed to connect: %v", err)
1791 }
1792 defer func() { _ = client.Close() }()
1793
1794 output, err := user.RunCommand(client, "status")
1795 if err != nil {
1796 t.Logf("command error (expected): %v", err)
1797 }
1798
1799 if !strings.Contains(output, "access denied") {
1800 t.Errorf("expected 'access denied', got: %s", output)
1801 }
1802}
1803
1804func TestStatus_NoMonitors(t *testing.T) {
1805 server := NewTestSSHServer(t)
1806 defer server.Shutdown()
1807
1808 user := GenerateUser("alice")
1809 RegisterUserWithServer(server, user)
1810
1811 client, err := user.NewClient()
1812 if err != nil {
1813 t.Fatalf("failed to connect: %v", err)
1814 }
1815 defer func() { _ = client.Close() }()
1816
1817 output, err := user.RunCommand(client, "status")
1818 if err != nil {
1819 t.Logf("command completed: %v", err)
1820 }
1821
1822 if !strings.Contains(output, "no monitors") && !strings.Contains(output, "empty") {
1823 t.Errorf("expected message about no monitors, got: %s", output)
1824 }
1825}
1826
1827func TestStatus_ShowsMonitorStatus(t *testing.T) {
1828 server := NewTestSSHServer(t)
1829 defer server.Shutdown()
1830
1831 user := GenerateUser("alice")
1832 RegisterUserWithServer(server, user)
1833
1834 client, err := user.NewClient()
1835 if err != nil {
1836 t.Fatalf("failed to connect: %v", err)
1837 }
1838 defer func() { _ = client.Close() }()
1839
1840 // Create a monitor
1841 _, err = user.RunCommand(client, "monitor web-check 1h")
1842 if err != nil {
1843 t.Logf("create monitor completed: %v", err)
1844 }
1845
1846 // Check status
1847 output, err := user.RunCommand(client, "status")
1848 if err != nil {
1849 t.Logf("status command completed: %v", err)
1850 }
1851
1852 if !strings.Contains(output, "web-check") {
1853 t.Errorf("status should list the monitor topic, got: %s", output)
1854 }
1855}
1856
1857func TestStatus_ShowsHealthyUnhealthy(t *testing.T) {
1858 server := NewTestSSHServer(t)
1859 defer server.Shutdown()
1860
1861 user := GenerateUser("alice")
1862 RegisterUserWithServer(server, user)
1863
1864 // Create monitors directly in DB with different states
1865 now := time.Now()
1866 windowEnd := now.Add(1 * time.Hour)
1867 recentPing := now.Add(-30 * time.Minute) // within window - healthy
1868 oldPing := now.Add(-2 * time.Hour) // outside window - unhealthy
1869
1870 _ = server.DBPool.UpsertPipeMonitor("alice-id", "healthy-service", 1*time.Hour, &windowEnd)
1871 _ = server.DBPool.UpdatePipeMonitorLastPing("alice-id", "healthy-service", &recentPing)
1872
1873 _ = server.DBPool.UpsertPipeMonitor("alice-id", "unhealthy-service", 1*time.Hour, &windowEnd)
1874 _ = server.DBPool.UpdatePipeMonitorLastPing("alice-id", "unhealthy-service", &oldPing)
1875
1876 client, err := user.NewClient()
1877 if err != nil {
1878 t.Fatalf("failed to connect: %v", err)
1879 }
1880 defer func() { _ = client.Close() }()
1881
1882 output, err := user.RunCommand(client, "status")
1883 if err != nil {
1884 t.Logf("status command completed: %v", err)
1885 }
1886
1887 if !strings.Contains(output, "healthy-service") {
1888 t.Errorf("status should list healthy-service, got: %s", output)
1889 }
1890
1891 if !strings.Contains(output, "unhealthy-service") {
1892 t.Errorf("status should list unhealthy-service, got: %s", output)
1893 }
1894
1895 // Should indicate different health states
1896 if !strings.Contains(strings.ToLower(output), "healthy") && !strings.Contains(strings.ToLower(output), "ok") && !strings.Contains(output, "✓") {
1897 t.Logf("status output should indicate health state: %s", output)
1898 }
1899}
1900
1901// RSS CLI Tests
1902
1903func TestRss_UnauthenticatedUserDenied(t *testing.T) {
1904 server := NewTestSSHServer(t)
1905 defer server.Shutdown()
1906
1907 user := GenerateUser("anonymous")
1908
1909 client, err := user.NewClient()
1910 if err != nil {
1911 t.Fatalf("failed to connect: %v", err)
1912 }
1913 defer func() { _ = client.Close() }()
1914
1915 output, err := user.RunCommand(client, "rss")
1916 if err != nil {
1917 t.Logf("command error (expected): %v", err)
1918 }
1919
1920 if !strings.Contains(output, "access denied") {
1921 t.Errorf("expected 'access denied', got: %s", output)
1922 }
1923}
1924
1925func TestRss_GeneratesValidRSS(t *testing.T) {
1926 server := NewTestSSHServer(t)
1927 defer server.Shutdown()
1928
1929 user := GenerateUser("alice")
1930 RegisterUserWithServer(server, user)
1931
1932 // Create a monitor
1933 now := time.Now()
1934 windowEnd := now.Add(1 * time.Hour)
1935 _ = server.DBPool.UpsertPipeMonitor("alice-id", "rss-test-service", 1*time.Hour, &windowEnd)
1936
1937 client, err := user.NewClient()
1938 if err != nil {
1939 t.Fatalf("failed to connect: %v", err)
1940 }
1941 defer func() { _ = client.Close() }()
1942
1943 output, err := user.RunCommand(client, "rss")
1944 if err != nil {
1945 t.Logf("rss command completed: %v", err)
1946 }
1947
1948 // Should output valid RSS XML
1949 if !strings.Contains(output, "<?xml") || !strings.Contains(output, "<rss") {
1950 t.Errorf("expected RSS XML output, got: %s", output)
1951 }
1952
1953 if !strings.Contains(output, "rss-test-service") {
1954 t.Errorf("RSS should contain monitor topic, got: %s", output)
1955 }
1956}
1957
1958func TestRss_AlertsOnStaleMonitor(t *testing.T) {
1959 server := NewTestSSHServer(t)
1960 defer server.Shutdown()
1961
1962 user := GenerateUser("alice")
1963 RegisterUserWithServer(server, user)
1964
1965 // Create a stale monitor (last ping outside window)
1966 now := time.Now()
1967 windowEnd := now.Add(-30 * time.Minute) // window already ended
1968 oldPing := now.Add(-2 * time.Hour)
1969
1970 _ = server.DBPool.UpsertPipeMonitor("alice-id", "stale-service", 1*time.Hour, &windowEnd)
1971 _ = server.DBPool.UpdatePipeMonitorLastPing("alice-id", "stale-service", &oldPing)
1972
1973 client, err := user.NewClient()
1974 if err != nil {
1975 t.Fatalf("failed to connect: %v", err)
1976 }
1977 defer func() { _ = client.Close() }()
1978
1979 output, err := user.RunCommand(client, "rss")
1980 if err != nil {
1981 t.Logf("rss command completed: %v", err)
1982 }
1983
1984 // Should contain alert item for stale service
1985 if !strings.Contains(output, "stale-service") {
1986 t.Errorf("RSS should contain stale-service alert, got: %s", output)
1987 }
1988
1989 // Should have item element for the alert
1990 if !strings.Contains(output, "<item>") {
1991 t.Errorf("RSS should contain item element for alert, got: %s", output)
1992 }
1993}
1994
1995// Pub integration with Monitor
1996
1997func TestPub_UpdatesMonitorLastPing(t *testing.T) {
1998 server := NewTestSSHServer(t)
1999 defer server.Shutdown()
2000
2001 user := GenerateUser("alice")
2002 RegisterUserWithServer(server, user)
2003
2004 // Create a monitor first (topic is stored with user prefix)
2005 now := time.Now()
2006 windowEnd := now.Add(1 * time.Hour)
2007 _ = server.DBPool.UpsertPipeMonitor("alice-id", "alice/ping-test", 1*time.Hour, &windowEnd)
2008
2009 subClient, err := user.NewClient()
2010 if err != nil {
2011 t.Fatalf("failed to connect subscriber: %v", err)
2012 }
2013 defer func() { _ = subClient.Close() }()
2014
2015 pubClient, err := user.NewClient()
2016 if err != nil {
2017 t.Fatalf("failed to connect publisher: %v", err)
2018 }
2019 defer func() { _ = pubClient.Close() }()
2020
2021 // Start subscriber
2022 subSession, err := subClient.NewSession()
2023 if err != nil {
2024 t.Fatalf("failed to create sub session: %v", err)
2025 }
2026 defer func() { _ = subSession.Close() }()
2027
2028 if err := subSession.Start("sub ping-test -c"); err != nil {
2029 t.Fatalf("failed to start sub: %v", err)
2030 }
2031
2032 server.waitForClients(t, 1)
2033
2034 // Publish to the monitored topic
2035 _, err = user.RunCommandWithStdin(pubClient, "pub ping-test -c", "health check")
2036 if err != nil {
2037 t.Logf("pub command completed: %v", err)
2038 }
2039
2040 // Verify last_ping was updated (topic is stored with user prefix)
2041 monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/ping-test")
2042 if err != nil {
2043 t.Fatalf("monitor should exist: %v", err)
2044 }
2045
2046 if monitor.LastPing == nil {
2047 t.Errorf("last_ping should be set after pub")
2048 } else if time.Since(*monitor.LastPing) > 5*time.Second {
2049 t.Errorf("last_ping should be recent, got: %v", monitor.LastPing)
2050 }
2051}
2052
2053// Tests for monitor status edge cases
2054
2055func TestStatus_PingAtExactWindowStart(t *testing.T) {
2056 // Bug fix: Status() should use >= for windowStart comparison
2057 // A ping exactly at windowStart should be healthy
2058 now := time.Now().UTC()
2059 windowEnd := now.Add(1 * time.Hour)
2060 windowStart := windowEnd.Add(-1 * time.Hour) // equals now
2061
2062 monitor := &db.PipeMonitor{
2063 LastPing: &windowStart, // ping exactly at window start
2064 WindowEnd: &windowEnd,
2065 WindowDur: 1 * time.Hour,
2066 }
2067
2068 err := monitor.Status()
2069 if err != nil {
2070 t.Errorf("ping at exact window start should be healthy, got: %v", err)
2071 }
2072}
2073
2074func TestStatus_WindowExpired(t *testing.T) {
2075 // Bug fix: Status() should check if current time is past windowEnd
2076 now := time.Now().UTC()
2077 windowEnd := now.Add(-1 * time.Minute) // window ended 1 minute ago
2078 lastPing := now.Add(-30 * time.Second) // ping was 30 seconds ago
2079
2080 monitor := &db.PipeMonitor{
2081 LastPing: &lastPing,
2082 WindowEnd: &windowEnd,
2083 WindowDur: 1 * time.Hour,
2084 }
2085
2086 err := monitor.Status()
2087 if err == nil {
2088 t.Error("expired window should be unhealthy")
2089 }
2090 if !strings.Contains(err.Error(), "window expired") {
2091 t.Errorf("error should mention window expired, got: %v", err)
2092 }
2093}
2094
2095func TestStatus_PingResetsWindow(t *testing.T) {
2096 // Bug fix: Every ping should reset window to now + duration
2097 server := NewTestSSHServer(t)
2098 defer server.Shutdown()
2099
2100 user := GenerateUser("alice")
2101 RegisterUserWithServer(server, user)
2102
2103 // Create a monitor with an expired window
2104 expiredWindowEnd := time.Now().UTC().Add(-10 * time.Minute)
2105 _ = server.DBPool.UpsertPipeMonitor("alice-id", "alice/reset-test", 5*time.Minute, &expiredWindowEnd)
2106
2107 client, err := user.NewClient()
2108 if err != nil {
2109 t.Fatalf("failed to connect: %v", err)
2110 }
2111 defer func() { _ = client.Close() }()
2112
2113 // Start a subscriber first so pub doesn't block
2114 subClient, err := user.NewClient()
2115 if err != nil {
2116 t.Fatalf("failed to connect subscriber: %v", err)
2117 }
2118 defer func() { _ = subClient.Close() }()
2119
2120 subSession, err := subClient.NewSession()
2121 if err != nil {
2122 t.Fatalf("failed to create sub session: %v", err)
2123 }
2124 defer func() { _ = subSession.Close() }()
2125
2126 if err := subSession.Start("sub reset-test -c"); err != nil {
2127 t.Fatalf("failed to start sub: %v", err)
2128 }
2129
2130 server.waitForClients(t, 1)
2131
2132 // Pub to trigger monitor update
2133 _, err = user.RunCommandWithStdin(client, "pub reset-test -c", "ping")
2134 if err != nil {
2135 t.Logf("pub command completed: %v", err)
2136 }
2137
2138 // Check that window was reset
2139 monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/reset-test")
2140 if err != nil {
2141 t.Fatalf("monitor should exist: %v", err)
2142 }
2143
2144 if monitor.WindowEnd == nil {
2145 t.Fatal("window_end should be set")
2146 }
2147
2148 // Window end should now be in the future
2149 if !monitor.WindowEnd.After(time.Now().UTC()) {
2150 t.Errorf("window_end should be in the future after ping, got: %v", monitor.WindowEnd)
2151 }
2152}
2153
2154func TestStatus_HealthyImmediatelyAfterPing(t *testing.T) {
2155 // Bug fix: After a ping, status should immediately show healthy
2156 server := NewTestSSHServer(t)
2157 defer server.Shutdown()
2158
2159 user := GenerateUser("alice")
2160 RegisterUserWithServer(server, user)
2161
2162 client, err := user.NewClient()
2163 if err != nil {
2164 t.Fatalf("failed to connect: %v", err)
2165 }
2166 defer func() { _ = client.Close() }()
2167
2168 // Create monitor
2169 _, err = user.RunCommand(client, "monitor health-test 5m")
2170 if err != nil {
2171 t.Fatalf("failed to create monitor: %v", err)
2172 }
2173
2174 // Start subscriber
2175 subClient, err := user.NewClient()
2176 if err != nil {
2177 t.Fatalf("failed to connect subscriber: %v", err)
2178 }
2179 defer func() { _ = subClient.Close() }()
2180
2181 subSession, err := subClient.NewSession()
2182 if err != nil {
2183 t.Fatalf("failed to create sub session: %v", err)
2184 }
2185 defer func() { _ = subSession.Close() }()
2186
2187 if err := subSession.Start("sub health-test -c"); err != nil {
2188 t.Fatalf("failed to start sub: %v", err)
2189 }
2190
2191 server.waitForClients(t, 1)
2192
2193 // Pub to trigger ping
2194 pubClient, err := user.NewClient()
2195 if err != nil {
2196 t.Fatalf("failed to connect publisher: %v", err)
2197 }
2198 defer func() { _ = pubClient.Close() }()
2199
2200 _, err = user.RunCommandWithStdin(pubClient, "pub health-test -c", "ping")
2201 if err != nil {
2202 t.Logf("pub completed: %v", err)
2203 }
2204
2205 // Immediately check status
2206 statusClient, err := user.NewClient()
2207 if err != nil {
2208 t.Fatalf("failed to connect for status: %v", err)
2209 }
2210 defer func() { _ = statusClient.Close() }()
2211
2212 output, err := user.RunCommand(statusClient, "status")
2213 if err != nil {
2214 t.Logf("status completed: %v", err)
2215 }
2216
2217 if strings.Contains(output, "unhealthy") {
2218 t.Errorf("status should be healthy immediately after ping, got: %s", output)
2219 }
2220 if !strings.Contains(output, "healthy") {
2221 t.Errorf("status should show healthy, got: %s", output)
2222 }
2223}
2224
2225// TestMonitor_FixedWindowNonSliding verifies that pings within the same window
2226// do not slide the window forward. This is a regression test for a bug where
2227// each ping reset window_end to now+dur, creating a sliding window that never fails.
2228//
2229// Expected behavior:
2230// - last_ping: always updated to show most recent activity (user visibility).
2231// - window_end: only advances when current time exceeds it (health scheduling).
2232func TestMonitor_FixedWindowNonSliding(t *testing.T) {
2233 server := NewTestSSHServer(t)
2234 defer server.Shutdown()
2235
2236 user := GenerateUser("alice")
2237 RegisterUserWithServer(server, user)
2238
2239 client, err := user.NewClient()
2240 if err != nil {
2241 t.Fatalf("failed to connect: %v", err)
2242 }
2243 defer func() { _ = client.Close() }()
2244
2245 // Create a monitor with 1 hour window
2246 _, err = user.RunCommand(client, "monitor fixed-window-test 1h")
2247 if err != nil {
2248 t.Logf("create command completed: %v", err)
2249 }
2250
2251 // Get the initial window_end
2252 monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/fixed-window-test")
2253 if err != nil {
2254 t.Fatalf("monitor should exist: %v", err)
2255 }
2256 initialWindowEnd := *monitor.WindowEnd
2257
2258 // Simulate a ping by calling updateMonitor directly
2259 handler := server.PipeHandler
2260
2261 // Create a mock CliCmd
2262 mockUser := &db.User{ID: "alice-id", Name: "alice"}
2263 cmd := &CliCmd{
2264 userName: "alice",
2265 user: mockUser,
2266 }
2267
2268 // First ping - should record last_ping but NOT change window_end
2269 handler.updateMonitor(cmd, "alice/fixed-window-test")
2270
2271 monitor, err = server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/fixed-window-test")
2272 if err != nil {
2273 t.Fatalf("monitor should exist after first ping: %v", err)
2274 }
2275
2276 if monitor.LastPing == nil {
2277 t.Fatalf("last_ping should be set after first ping")
2278 }
2279 firstPingTime := *monitor.LastPing
2280 windowEndAfterFirstPing := *monitor.WindowEnd
2281
2282 // BUG CHECK: With the bug, window_end would have slid forward to now+1h
2283 // With the fix, window_end should remain at the original scheduled time
2284 if !windowEndAfterFirstPing.Equal(initialWindowEnd) {
2285 t.Errorf("BUG DETECTED: window_end should NOT change after first ping within window\n"+
2286 "initial window_end: %v\n"+
2287 "window_end after ping: %v\n"+
2288 "Window slid forward by: %v",
2289 initialWindowEnd.Format(time.RFC3339),
2290 windowEndAfterFirstPing.Format(time.RFC3339),
2291 windowEndAfterFirstPing.Sub(initialWindowEnd))
2292 }
2293
2294 // Second ping - last_ping SHOULD be updated (for user visibility)
2295 // but window_end should NOT change
2296 time.Sleep(10 * time.Millisecond) // Small delay to get different timestamp
2297 handler.updateMonitor(cmd, "alice/fixed-window-test")
2298
2299 monitor, err = server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/fixed-window-test")
2300 if err != nil {
2301 t.Fatalf("monitor should exist after second ping: %v", err)
2302 }
2303
2304 // last_ping SHOULD be updated to show most recent activity
2305 if monitor.LastPing.Equal(firstPingTime) {
2306 t.Errorf("last_ping SHOULD be updated for user visibility\n"+
2307 "first ping time: %v\n"+
2308 "last_ping after second call: %v",
2309 firstPingTime.Format(time.RFC3339Nano),
2310 monitor.LastPing.Format(time.RFC3339Nano))
2311 }
2312
2313 // But window_end should still be the original value (not sliding)
2314 if !monitor.WindowEnd.Equal(initialWindowEnd) {
2315 t.Errorf("BUG DETECTED: window_end should remain at original value\n"+
2316 "initial: %v\n"+
2317 "current: %v",
2318 initialWindowEnd.Format(time.RFC3339),
2319 monitor.WindowEnd.Format(time.RFC3339))
2320 }
2321}
2322
2323func TestSSH_WildcardSub(t *testing.T) {
2324 server := NewTestSSHServer(t)
2325 defer server.Shutdown()
2326
2327 user := GenerateUser("alice")
2328 dbUser := &db.User{ID: "alice-id", Name: "alice"}
2329 server.DBPool.AddUser(dbUser)
2330 server.DBPool.AddPubkey(&db.PublicKey{
2331 ID: "alice-pk",
2332 UserID: "alice-id",
2333 Key: user.PublicKey(),
2334 })
2335
2336 client, err := user.NewClient()
2337 if err != nil {
2338 t.Fatalf("failed to dial ssh server: %v", err)
2339 }
2340 defer func() { _ = client.Close() }()
2341
2342 // 1. Subscribe to wildcard topic: "sub metric-drain*"
2343 subSession, err := client.NewSession()
2344 if err != nil {
2345 t.Fatalf("failed to create sub session: %v", err)
2346 }
2347
2348 subOut, err := subSession.StdoutPipe()
2349 if err != nil {
2350 t.Fatalf("failed stdout pipe: %v", err)
2351 }
2352
2353 if err := subSession.Start("sub metric-drain*"); err != nil {
2354 t.Fatalf("failed to start sub: %v", err)
2355 }
2356
2357 var buf bytes.Buffer
2358 var bufMu sync.Mutex
2359 go func() {
2360 b := make([]byte, 1024)
2361 for {
2362 n, err := subOut.Read(b)
2363 if n > 0 {
2364 bufMu.Lock()
2365 buf.Write(b[:n])
2366 bufMu.Unlock()
2367 }
2368 if err != nil {
2369 break
2370 }
2371 }
2372 }()
2373
2374 // Wait for subscriber to be registered in the PubSub broker
2375 subReady := false
2376 for i := 0; i < 100; i++ {
2377 for topic, ch := range server.PipeHandler.PubSub.GetChannels() {
2378 if topic == "alice/metric-drain*" {
2379 for range ch.GetClients() {
2380 subReady = true
2381 break
2382 }
2383 }
2384 }
2385 if subReady {
2386 break
2387 }
2388 time.Sleep(10 * time.Millisecond)
2389 }
2390 if !subReady {
2391 t.Fatal("timed out waiting for wildcard subscriber to register")
2392 }
2393
2394 // 2. Publish to "metric-drain-pgs"
2395 pubClient1, err := user.NewClient()
2396 if err != nil {
2397 t.Fatalf("failed to dial pub client 1: %v", err)
2398 }
2399 defer func() { _ = pubClient1.Close() }()
2400
2401 pubSession1, err := pubClient1.NewSession()
2402 if err != nil {
2403 t.Fatalf("failed pub session 1: %v", err)
2404 }
2405 pubIn1, _ := pubSession1.StdinPipe()
2406 go func() {
2407 defer func() { _ = pubIn1.Close() }()
2408 _, _ = io.WriteString(pubIn1, "pgs-event\n")
2409 }()
2410 _ = pubSession1.Run("pub metric-drain-pgs -b=false")
2411
2412 // 3. Publish to "metric-drain-prose"
2413 pubClient2, err := user.NewClient()
2414 if err != nil {
2415 t.Fatalf("failed to dial pub client 2: %v", err)
2416 }
2417 defer func() { _ = pubClient2.Close() }()
2418
2419 pubSession2, err := pubClient2.NewSession()
2420 if err != nil {
2421 t.Fatalf("failed pub session 2: %v", err)
2422 }
2423 pubIn2, _ := pubSession2.StdinPipe()
2424 go func() {
2425 defer func() { _ = pubIn2.Close() }()
2426 _, _ = io.WriteString(pubIn2, "prose-event\n")
2427 }()
2428 _ = pubSession2.Run("pub metric-drain-prose -b=false")
2429
2430 // Wait for both events to be received by the subscriber
2431 deadline := time.Now().Add(5 * time.Second)
2432 var output string
2433 for time.Now().Before(deadline) {
2434 bufMu.Lock()
2435 output = buf.String()
2436 bufMu.Unlock()
2437 if strings.Contains(output, "pgs-event") && strings.Contains(output, "prose-event") {
2438 break
2439 }
2440 time.Sleep(10 * time.Millisecond)
2441 }
2442
2443 _ = subSession.Close()
2444
2445 if !strings.Contains(output, "pgs-event") {
2446 t.Errorf("expected SSH wildcard subscriber output to contain 'pgs-event', got: %q", output)
2447 }
2448 if !strings.Contains(output, "prose-event") {
2449 t.Errorf("expected SSH wildcard subscriber output to contain 'prose-event', got: %q", output)
2450 }
2451}
2452
2453// runStatus runs cmd with stdin and returns its exit status, or -1 when the
2454// session ended without one.
2455func runStatus(t *testing.T, client *ssh.Client, cmd, stdin string) int {
2456 t.Helper()
2457 session, err := client.NewSession()
2458 if err != nil {
2459 t.Fatal(err)
2460 }
2461 defer func() { _ = session.Close() }()
2462 session.Stdin = strings.NewReader(stdin)
2463 return exitStatus(t, session.Run(cmd))
2464}
2465
2466func exitStatus(t *testing.T, err error) int {
2467 t.Helper()
2468 var exitErr *ssh.ExitError
2469 var missing *ssh.ExitMissingError
2470 switch {
2471 case err == nil:
2472 return 0
2473 case errors.As(err, &exitErr):
2474 return exitErr.ExitStatus()
2475 case errors.As(err, &missing):
2476 return -1
2477 }
2478 t.Fatalf("command failed: %v", err)
2479 return 0
2480}
2481
2482func TestCommandsExitWithStatus(t *testing.T) {
2483 server := NewTestSSHServer(t)
2484 defer server.Shutdown()
2485
2486 alice := GenerateUser("alice")
2487 charlie := GenerateUser("charlie")
2488 RegisterUserWithServer(server, alice)
2489 RegisterUserWithServer(server, charlie)
2490 aliceClient, err := alice.NewClient()
2491 if err != nil {
2492 t.Fatal(err)
2493 }
2494 defer func() { _ = aliceClient.Close() }()
2495 charlieClient, err := charlie.NewClient()
2496 if err != nil {
2497 t.Fatal(err)
2498 }
2499 defer func() { _ = charlieClient.Close() }()
2500
2501 if got := runStatus(t, aliceClient, "pub nobody -b=false", "hello"); got != 0 {
2502 t.Errorf("pub exited %d, want 0", got)
2503 }
2504
2505 sub, err := aliceClient.NewSession()
2506 if err != nil {
2507 t.Fatal(err)
2508 }
2509 defer func() { _ = sub.Close() }()
2510 if err := sub.Start("sub news -a alice"); err != nil {
2511 t.Fatal(err)
2512 }
2513 server.waitForClients(t, 1)
2514 if got := runStatus(t, aliceClient, "pub news", "hello"); got != 0 {
2515 t.Errorf("pub to a sub exited %d, want 0", got)
2516 }
2517 if got := exitStatus(t, sub.Wait()); got != 0 {
2518 t.Errorf("sub exited %d after its publisher left, want 0", got)
2519 }
2520
2521 if err := (func() error {
2522 s, err := aliceClient.NewSession()
2523 if err != nil {
2524 return err
2525 }
2526 return s.Start("sub restricted -p -a alice -k")
2527 })(); err != nil {
2528 t.Fatal(err)
2529 }
2530 server.waitForClients(t, 1)
2531 if got := runStatus(t, charlieClient, "sub restricted -p", ""); got != 1 {
2532 t.Errorf("denied sub exited %d, want 1", got)
2533 }
2534}