webhooks_test.go 8.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360
  1. package webhooks
  2. import (
  3. "bytes"
  4. "encoding/json"
  5. "fmt"
  6. "net/http"
  7. "net/http/httptest"
  8. "os"
  9. "sync"
  10. "sync/atomic"
  11. "testing"
  12. "time"
  13. "github.com/owncast/owncast/core/chat/events"
  14. "github.com/owncast/owncast/core/data"
  15. "github.com/owncast/owncast/models"
  16. jsonpatch "gopkg.in/evanphx/json-patch.v5"
  17. )
  18. func fakeGetStatus() models.Status {
  19. return models.Status{
  20. Online: true,
  21. ViewerCount: 5,
  22. OverallMaxViewerCount: 420,
  23. SessionMaxViewerCount: 69,
  24. StreamTitle: "my stream",
  25. VersionNumber: "1.2.3",
  26. }
  27. }
  28. func TestMain(m *testing.M) {
  29. dbFile, err := os.CreateTemp(os.TempDir(), "owncast-test-db.db")
  30. if err != nil {
  31. panic(err)
  32. }
  33. dbFile.Close()
  34. defer os.Remove(dbFile.Name())
  35. if err := data.SetupPersistence(dbFile.Name()); err != nil {
  36. panic(err)
  37. }
  38. SetupWebhooks(fakeGetStatus)
  39. defer close(queue)
  40. m.Run()
  41. }
  42. // Because the other tests use `sendEventToWebhooks` with a `WaitGroup` to know when the test completes,
  43. // this test ensures that `SendToWebhooks` without a `WaitGroup` doesn't panic.
  44. func TestPublicSend(t *testing.T) {
  45. // Send enough events to be sure at least one worker delivers a second event.
  46. eventsCount := webhookWorkerPoolSize + 1
  47. var wg sync.WaitGroup
  48. wg.Add(eventsCount)
  49. svr := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
  50. wg.Done()
  51. }))
  52. defer svr.Close()
  53. hook, err := data.InsertWebhook(svr.URL, []models.EventType{models.MessageSent})
  54. if err != nil {
  55. t.Fatal(err)
  56. }
  57. defer func() {
  58. if err := data.DeleteWebhook(hook); err != nil {
  59. t.Error(err)
  60. }
  61. }()
  62. for i := 0; i < eventsCount; i++ {
  63. wh := WebhookEvent{
  64. EventData: struct{}{},
  65. Type: models.MessageSent,
  66. }
  67. SendEventToWebhooks(wh)
  68. }
  69. wg.Wait()
  70. }
  71. // Make sure that events are only sent to interested endpoints.
  72. func TestRouting(t *testing.T) {
  73. eventTypes := []models.EventType{models.ChatActionSent, models.UserJoined, events.UserParted}
  74. calls := map[models.EventType]int{}
  75. var lock sync.Mutex
  76. svr := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
  77. if len(r.URL.Path) < 1 || r.URL.Path[0] != '/' {
  78. t.Fatalf("Got unexpected path %v", r.URL.Path)
  79. }
  80. pathType := r.URL.Path[1:]
  81. var body WebhookEvent
  82. if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
  83. t.Fatal(err)
  84. }
  85. if body.Type != pathType {
  86. t.Fatalf("Got %v payload on %v endpoint", body.Type, pathType)
  87. }
  88. lock.Lock()
  89. defer lock.Unlock()
  90. calls[pathType] += 1
  91. }))
  92. defer svr.Close()
  93. for _, eventType := range eventTypes {
  94. hook, err := data.InsertWebhook(svr.URL+"/"+eventType, []models.EventType{eventType})
  95. if err != nil {
  96. t.Fatal(err)
  97. }
  98. defer func() {
  99. if err := data.DeleteWebhook(hook); err != nil {
  100. t.Error(err)
  101. }
  102. }()
  103. }
  104. var wg sync.WaitGroup
  105. for _, eventType := range eventTypes {
  106. wh := WebhookEvent{
  107. EventData: struct{}{},
  108. Type: eventType,
  109. }
  110. sendEventToWebhooks(wh, &wg)
  111. }
  112. wg.Wait()
  113. for _, eventType := range eventTypes {
  114. if calls[eventType] != 1 {
  115. t.Errorf("Expected %v to be called exactly once but it was called %v times", eventType, calls[eventType])
  116. }
  117. }
  118. }
  119. // Make sure that events are sent to all interested endpoints.
  120. func TestMultiple(t *testing.T) {
  121. const times = 2
  122. var calls uint32
  123. svr := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
  124. atomic.AddUint32(&calls, 1)
  125. }))
  126. defer svr.Close()
  127. for i := 0; i < times; i++ {
  128. hook, err := data.InsertWebhook(fmt.Sprintf("%v/%v", svr.URL, i), []models.EventType{models.MessageSent})
  129. if err != nil {
  130. t.Fatal(err)
  131. }
  132. defer func() {
  133. if err := data.DeleteWebhook(hook); err != nil {
  134. t.Error(err)
  135. }
  136. }()
  137. }
  138. var wg sync.WaitGroup
  139. wh := WebhookEvent{
  140. EventData: struct{}{},
  141. Type: models.MessageSent,
  142. }
  143. sendEventToWebhooks(wh, &wg)
  144. wg.Wait()
  145. if atomic.LoadUint32(&calls) != times {
  146. t.Errorf("Expected event to be sent exactly %v times but it was sent %v times", times, atomic.LoadUint32(&calls))
  147. }
  148. }
  149. // Make sure when a webhook is used its last used timestamp is updated.
  150. func TestTimestamps(t *testing.T) {
  151. const tolerance = time.Second
  152. start := time.Now()
  153. eventTypes := []models.EventType{models.StreamStarted, models.StreamStopped}
  154. handlerIds := []int{0, 0}
  155. handlers := []*models.Webhook{nil, nil}
  156. svr := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
  157. }))
  158. defer svr.Close()
  159. for i, eventType := range eventTypes {
  160. hook, err := data.InsertWebhook(svr.URL+"/"+eventType, []models.EventType{eventType})
  161. if err != nil {
  162. t.Fatal(err)
  163. }
  164. handlerIds[i] = hook
  165. defer func() {
  166. if err := data.DeleteWebhook(hook); err != nil {
  167. t.Error(err)
  168. }
  169. }()
  170. }
  171. var wg sync.WaitGroup
  172. wh := WebhookEvent{
  173. EventData: struct{}{},
  174. Type: eventTypes[0],
  175. }
  176. sendEventToWebhooks(wh, &wg)
  177. wg.Wait()
  178. hooks, err := data.GetWebhooks()
  179. if err != nil {
  180. t.Fatal(err)
  181. }
  182. for h, hook := range hooks {
  183. for i, handlerId := range handlerIds {
  184. if hook.ID == handlerId {
  185. handlers[i] = &hooks[h]
  186. }
  187. }
  188. }
  189. if handlers[0] == nil {
  190. t.Fatal("First handler was not found in registered handlers")
  191. }
  192. if handlers[1] == nil {
  193. t.Fatal("Second handler was not found in registered handlers")
  194. }
  195. end := time.Now()
  196. if handlers[0].Timestamp.Add(tolerance).Before(start) {
  197. t.Errorf("First handler timestamp %v should not be before start of test %v", handlers[0].Timestamp, start)
  198. }
  199. if handlers[0].Timestamp.Add(tolerance).Before(handlers[1].Timestamp) {
  200. t.Errorf("Second handler timestamp %v should not be before first handler timestamp %v", handlers[1].Timestamp, handlers[0].Timestamp)
  201. }
  202. if end.Add(tolerance).Before(handlers[1].Timestamp) {
  203. t.Errorf("End of test %v should not be before second handler timestamp %v", end, handlers[1].Timestamp)
  204. }
  205. if handlers[0].LastUsed == nil {
  206. t.Error("First handler last used should have been set")
  207. } else if handlers[0].LastUsed.Add(tolerance).Before(handlers[1].Timestamp) {
  208. t.Errorf("First handler last used %v should not be before second handler timestamp %v", handlers[0].LastUsed, handlers[1].Timestamp)
  209. } else if end.Add(tolerance).Before(*handlers[0].LastUsed) {
  210. t.Errorf("End of test %v should not be before first handler last used %v", end, handlers[0].LastUsed)
  211. }
  212. if handlers[1].LastUsed != nil {
  213. t.Error("Second handler last used should not have been set")
  214. }
  215. }
  216. // Make sure up to the expected number of events can be fired in parallel.
  217. func TestParallel(t *testing.T) {
  218. var calls uint32
  219. var wgStart sync.WaitGroup
  220. finished := make(chan int)
  221. wgStart.Add(webhookWorkerPoolSize)
  222. svr := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
  223. myId := atomic.AddUint32(&calls, 1)
  224. // We made it to the pool size + 1 event, so we're done with the test.
  225. if myId == uint32(webhookWorkerPoolSize)+1 {
  226. close(finished)
  227. return
  228. }
  229. // Wait until all the handlers are started.
  230. wgStart.Done()
  231. wgStart.Wait()
  232. // The first handler just returns so the pool size + 1 event can be handled.
  233. if myId != 1 {
  234. // The other handlers will wait for pool size + 1.
  235. _ = <-finished
  236. }
  237. }))
  238. defer svr.Close()
  239. hook, err := data.InsertWebhook(svr.URL, []models.EventType{models.MessageSent})
  240. if err != nil {
  241. t.Fatal(err)
  242. }
  243. defer func() {
  244. if err := data.DeleteWebhook(hook); err != nil {
  245. t.Error(err)
  246. }
  247. }()
  248. var wgMessages sync.WaitGroup
  249. for i := 0; i < webhookWorkerPoolSize+1; i++ {
  250. wh := WebhookEvent{
  251. EventData: struct{}{},
  252. Type: models.MessageSent,
  253. }
  254. sendEventToWebhooks(wh, &wgMessages)
  255. }
  256. wgMessages.Wait()
  257. }
  258. // Send an event, capture it, and verify that it has the expected payload.
  259. func checkPayload(t *testing.T, eventType models.EventType, send func(), expectedJson string) {
  260. eventChannel := make(chan WebhookEvent)
  261. // Set up a server.
  262. svr := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
  263. data := WebhookEvent{}
  264. json.NewDecoder(r.Body).Decode(&data)
  265. eventChannel <- data
  266. }))
  267. defer svr.Close()
  268. // Subscribe to the webhook.
  269. hook, err := data.InsertWebhook(svr.URL, []models.EventType{eventType})
  270. if err != nil {
  271. t.Fatal(err)
  272. }
  273. defer func() {
  274. if err := data.DeleteWebhook(hook); err != nil {
  275. t.Error(err)
  276. }
  277. }()
  278. // Send and capture the event.
  279. send()
  280. event := <-eventChannel
  281. if event.Type != eventType {
  282. t.Errorf("Got event type %v but expected %v", event.Type, eventType)
  283. }
  284. // Compare.
  285. payloadJson, err := json.MarshalIndent(event.EventData, "", " ")
  286. if err != nil {
  287. t.Fatal(err)
  288. }
  289. t.Logf("Actual payload:\n%s", payloadJson)
  290. if !jsonpatch.Equal(payloadJson, []byte(expectedJson)) {
  291. diff, err := jsonpatch.CreateMergePatch(payloadJson, []byte(expectedJson))
  292. if err != nil {
  293. t.Fatal(err)
  294. }
  295. var out bytes.Buffer
  296. if err := json.Indent(&out, diff, "", " "); err != nil {
  297. t.Fatal(err)
  298. }
  299. t.Errorf("Expected difference from actual payload:\n%s", out.Bytes())
  300. }
  301. }