connection_test.go 4.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167
  1. package oscar
  2. import (
  3. "bytes"
  4. "context"
  5. "io"
  6. "log/slog"
  7. "sync"
  8. "testing"
  9. "github.com/stretchr/testify/assert"
  10. "github.com/stretchr/testify/mock"
  11. "github.com/mk6i/retro-aim-server/state"
  12. "github.com/mk6i/retro-aim-server/wire"
  13. )
  14. func TestHandleChatConnection_MessageRelay(t *testing.T) {
  15. sessionManager := state.NewInMemorySessionManager(slog.Default())
  16. // add a user to session that will receive relayed messages
  17. sess, _ := sessionManager.AddSession(nil, "bob")
  18. // start the server connection handler in the background
  19. serverReader, _ := io.Pipe()
  20. clientReader, serverWriter := io.Pipe()
  21. go func() {
  22. flapc := wire.NewFlapClient(0, nil, serverWriter)
  23. err := dispatchIncomingMessages(context.Background(), sess, flapc, serverReader, slog.Default(), nil)
  24. assert.NoError(t, err)
  25. }()
  26. inboundMsgs := []wire.SNACMessage{
  27. {
  28. Frame: wire.SNACFrame{
  29. FoodGroup: wire.Chat,
  30. SubGroup: wire.ChatUsersJoined,
  31. },
  32. Body: wire.SNAC_0x0E_0x03_ChatUsersJoined{
  33. Users: []wire.TLVUserInfo{
  34. {
  35. ScreenName: "screenname1",
  36. },
  37. },
  38. },
  39. },
  40. {
  41. Frame: wire.SNACFrame{
  42. FoodGroup: wire.Chat,
  43. SubGroup: wire.ChatUsersLeft,
  44. },
  45. Body: wire.SNAC_0x0E_0x03_ChatUsersJoined{
  46. Users: []wire.TLVUserInfo{
  47. {
  48. ScreenName: "screenname2",
  49. },
  50. },
  51. },
  52. },
  53. }
  54. // relay messages to user session
  55. for _, msg := range inboundMsgs {
  56. sessionManager.RelayToScreenName(context.Background(), state.NewIdentScreenName("bob"), msg)
  57. }
  58. // consume and verify the relayed messages
  59. for i := 0; i < len(inboundMsgs); i++ {
  60. flap := wire.FLAPFrame{}
  61. assert.NoError(t, wire.UnmarshalBE(&flap, clientReader))
  62. frame := wire.SNACFrame{}
  63. buf := bytes.NewBuffer(flap.Payload)
  64. assert.NoError(t, wire.UnmarshalBE(&frame, buf))
  65. assert.Equal(t, inboundMsgs[i].Frame, frame)
  66. body := wire.SNAC_0x0E_0x03_ChatUsersJoined{}
  67. assert.NoError(t, wire.UnmarshalBE(&body, buf))
  68. assert.Equal(t, inboundMsgs[i].Body, body)
  69. }
  70. // stop the session, which terminates the connection handler goroutine
  71. sess.Close()
  72. <-sess.Closed()
  73. // verify the connection handler sends client disconnection message before
  74. // terminating
  75. flap := wire.FLAPFrame{}
  76. assert.NoError(t, wire.UnmarshalBE(&flap, clientReader))
  77. assert.Equal(t, wire.FLAPFrameSignoff, flap.FrameType)
  78. }
  79. func TestHandleChatConnection_ClientRequest(t *testing.T) {
  80. sessionManager := state.NewInMemorySessionManager(slog.Default())
  81. // add session so that the function can terminate upon closure
  82. sess, _ := sessionManager.AddSession(nil, "bob")
  83. inboundMsgs := []wire.SNACMessage{
  84. {
  85. Frame: wire.SNACFrame{
  86. FoodGroup: wire.Chat,
  87. SubGroup: wire.ChatUsersJoined,
  88. },
  89. Body: wire.SNAC_0x0E_0x03_ChatUsersJoined{
  90. Users: []wire.TLVUserInfo{
  91. {
  92. ScreenName: "screenname1",
  93. },
  94. },
  95. },
  96. },
  97. {
  98. Frame: wire.SNACFrame{
  99. FoodGroup: wire.Chat,
  100. SubGroup: wire.ChatUsersLeft,
  101. },
  102. Body: wire.SNAC_0x0E_0x03_ChatUsersJoined{
  103. Users: []wire.TLVUserInfo{
  104. {
  105. ScreenName: "screenname2",
  106. },
  107. },
  108. },
  109. },
  110. }
  111. wg := &sync.WaitGroup{}
  112. wg.Add(len(inboundMsgs))
  113. // set up mock handlers to receive messages and verify their contents
  114. router := newMockHandler(t)
  115. for _, msg := range inboundMsgs {
  116. msg := msg
  117. router.EXPECT().
  118. Handle(mock.Anything, sess, msg.Frame, mock.Anything, mock.Anything).
  119. Run(func(ctx context.Context, sess *state.Session, inFrame wire.SNACFrame, r io.Reader, rw ResponseWriter) {
  120. defer wg.Done()
  121. body := wire.SNAC_0x0E_0x03_ChatUsersJoined{}
  122. assert.NoError(t, wire.UnmarshalBE(&body, r))
  123. assert.Equal(t, msg.Body, body)
  124. }).
  125. Return(nil)
  126. }
  127. // start the server connection handler in the background
  128. serverReader, clientWriter := io.Pipe()
  129. clientReader, serverWriter := io.Pipe()
  130. go func() {
  131. flapc := wire.NewFlapClient(0, nil, serverWriter)
  132. assert.NoError(t, dispatchIncomingMessages(context.Background(), sess, flapc, serverReader, slog.Default(), router))
  133. }()
  134. // send client messages
  135. flapc := wire.NewFlapClient(0, nil, clientWriter)
  136. for _, msg := range inboundMsgs {
  137. err := flapc.SendSNAC(msg.Frame, msg.Body)
  138. assert.NoError(t, err)
  139. }
  140. wg.Wait()
  141. // stop the session, which terminates the connection handler goroutine
  142. sess.Close()
  143. <-sess.Closed()
  144. // verify the connection handler sends client disconnection message before
  145. // terminating
  146. flap := wire.FLAPFrame{}
  147. assert.NoError(t, wire.UnmarshalBE(&flap, clientReader))
  148. assert.Equal(t, wire.FLAPFrameSignoff, flap.FrameType)
  149. }