woodpecker/server/pubsub/pub_test.go

106 lines
2.2 KiB
Go
Raw Normal View History

2019-04-06 13:44:04 +00:00
package pubsub
import (
"context"
"errors"
2019-04-06 13:44:04 +00:00
"sync"
"testing"
"time"
"github.com/stretchr/testify/assert"
2019-04-06 13:44:04 +00:00
)
func TestPubsub(t *testing.T) {
var (
wg sync.WaitGroup
testTopic = "test"
testMessage = Message{
Data: []byte("test"),
}
)
ctx, cancel := context.WithCancelCause(
2019-04-06 13:44:04 +00:00
context.Background(),
)
broker := New()
assert.NoError(t, broker.Create(ctx, testTopic))
2019-04-06 13:44:04 +00:00
go func() {
assert.NoError(t, broker.Subscribe(ctx, testTopic, func(message Message) { wg.Done() }))
2019-04-06 13:44:04 +00:00
}()
go func() {
assert.NoError(t, broker.Subscribe(ctx, testTopic, func(message Message) { wg.Done() }))
2019-04-06 13:44:04 +00:00
}()
<-time.After(500 * time.Millisecond)
2019-04-06 13:44:04 +00:00
if _, ok := broker.(*publisher).topics[testTopic]; !ok {
t.Errorf("Expect topic registered with publisher")
}
wg.Add(2)
go func() {
assert.NoError(t, broker.Publish(ctx, testTopic, testMessage))
2019-04-06 13:44:04 +00:00
}()
wg.Wait()
cancel(nil)
2019-04-06 13:44:04 +00:00
}
func TestPublishNotFound(t *testing.T) {
var (
testTopic = "test"
testMessage = Message{
Data: []byte("test"),
}
)
broker := New()
err := broker.Publish(context.Background(), testTopic, testMessage)
if !errors.Is(err, ErrNotFound) {
2019-04-06 13:44:04 +00:00
t.Errorf("Expect Not Found error when topic does not exist")
}
}
func TestSubscribeNotFound(t *testing.T) {
var (
testTopic = "test"
testCallback = func(message Message) {}
)
broker := New()
err := broker.Subscribe(context.Background(), testTopic, testCallback)
if !errors.Is(err, ErrNotFound) {
2019-04-06 13:44:04 +00:00
t.Errorf("Expect Not Found error when topic does not exist")
}
}
func TestSubscriptionClosed(t *testing.T) {
var (
wg sync.WaitGroup
testTopic = "test"
testCallback = func(Message) {}
)
broker := New()
assert.NoError(t, broker.Create(context.Background(), testTopic))
2019-04-06 13:44:04 +00:00
go func() {
assert.NoError(t, broker.Subscribe(context.Background(), testTopic, testCallback))
2019-04-06 13:44:04 +00:00
wg.Done()
}()
<-time.After(500 * time.Millisecond)
2019-04-06 13:44:04 +00:00
if _, ok := broker.(*publisher).topics[testTopic]; !ok {
t.Errorf("Expect topic registered with publisher")
}
wg.Add(1)
assert.NoError(t, broker.Remove(context.Background(), testTopic))
2019-04-06 13:44:04 +00:00
wg.Wait()
if _, ok := broker.(*publisher).topics[testTopic]; ok {
t.Errorf("Expect topic removed from publisher")
}
}