Dmitry VerkhoturovandUmputun
95f59213e5
TestService_WithDrops and TestService_SubmitVerificationWithDrops submitted three items into a size-1 queue and asserted at least one was dropped, relying on the single consumer not draining the queue between submits. synctest does not fully pin this because the MockDest send path does real logging I/O outside the bubble, so under CI's -race scheduling the consumer occasionally drained all three, delivering everything and failing the "<= 2" assertion (~0.2% of CI runs). Add an optional gate channel to MockDest so a destination blocks in Send/SendVerification until released. The tests now submit one item (the consumer picks it up and blocks on the gate), fill the size-1 queue, submit an overflow item that is dropped, then release the gate — the drop is deterministic regardless of scheduling. Assert exactly two delivered instead of "at most two".
332 lines
12 KiB
Go
332 lines
12 KiB
Go
package notify
|
|
|
|
import (
|
|
"fmt"
|
|
"sync/atomic"
|
|
"testing"
|
|
"testing/synctest"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
|
|
"github.com/umputun/remark42/backend/app/store"
|
|
)
|
|
|
|
func TestService_NoDestinations(t *testing.T) {
|
|
s := NewService(nil, 0)
|
|
assert.Equal(t, defaultQueueSize, cap(s.queue))
|
|
assert.NotNil(t, s)
|
|
s.Submit(Request{Comment: store.Comment{ID: "123"}})
|
|
s.Submit(Request{Comment: store.Comment{ID: "123"}})
|
|
s.Submit(Request{Comment: store.Comment{ID: "123"}})
|
|
s.Close()
|
|
// second call should not result in panic
|
|
s.Close()
|
|
}
|
|
|
|
func TestService_WithDestinations(t *testing.T) {
|
|
synctest.Test(t, func(t *testing.T) {
|
|
d1, d2 := &MockDest{id: 1}, &MockDest{id: 2}
|
|
s := NewService(nil, 1, d1, d2)
|
|
assert.NotNil(t, s)
|
|
|
|
s.Submit(Request{Comment: store.Comment{ID: "100"}})
|
|
synctest.Wait()
|
|
s.Submit(Request{Comment: store.Comment{ID: "101"}})
|
|
synctest.Wait()
|
|
s.Submit(Request{Comment: store.Comment{ID: "102"}})
|
|
synctest.Wait()
|
|
s.Close()
|
|
|
|
require.Equal(t, 3, len(d1.Get()), "got all comments to d1")
|
|
require.Equal(t, 3, len(d2.Get()), "got all comments to d2")
|
|
|
|
assert.Equal(t, "100", d1.Get()[0].Comment.ID)
|
|
assert.Equal(t, "101", d1.Get()[1].Comment.ID)
|
|
assert.Equal(t, "102", d1.Get()[2].Comment.ID)
|
|
})
|
|
}
|
|
|
|
func TestService_WithDrops(t *testing.T) {
|
|
synctest.Test(t, func(t *testing.T) {
|
|
// gated destinations pin the consumer on the first item so the size-1 queue
|
|
// fills deterministically and the overflow is dropped regardless of scheduling
|
|
gate := make(chan struct{})
|
|
d1, d2 := &MockDest{id: 1, block: gate}, &MockDest{id: 2, block: gate}
|
|
s := NewService(nil, 1, d1, d2)
|
|
assert.NotNil(t, s)
|
|
|
|
s.Submit(Request{Comment: store.Comment{ID: "100"}}) // consumed, consumer blocks in Send on the gate
|
|
synctest.Wait()
|
|
s.Submit(Request{Comment: store.Comment{ID: "101"}}) // fills the size-1 queue
|
|
s.Submit(Request{Comment: store.Comment{ID: "102"}}) // queue full, dropped
|
|
synctest.Wait()
|
|
|
|
close(gate) // release the consumer: it finishes 100 then processes 101
|
|
synctest.Wait()
|
|
s.Close()
|
|
|
|
s.Submit(Request{Comment: store.Comment{ID: "111"}}) // safe to send after close
|
|
|
|
require.Len(t, d1.Get(), 2, "one comment of three dropped from d1, got: %v", d1.Get())
|
|
require.Len(t, d2.Get(), 2, "one comment of three dropped from d2, got: %v", d2.Get())
|
|
assert.Equal(t, "100", d1.Get()[0].Comment.ID)
|
|
assert.Equal(t, "101", d1.Get()[1].Comment.ID)
|
|
})
|
|
}
|
|
|
|
func TestService_SubmitVerificationWithDrops(t *testing.T) {
|
|
synctest.Test(t, func(t *testing.T) {
|
|
// gated destinations pin the consumer on the first item so the size-1 queue
|
|
// fills deterministically and the overflow is dropped regardless of scheduling
|
|
gate := make(chan struct{})
|
|
d1, d2 := &MockDest{id: 1, block: gate}, &MockDest{id: 2, block: gate}
|
|
s := NewService(nil, 1, d1, d2)
|
|
assert.NotNil(t, s)
|
|
|
|
s.SubmitVerification(VerificationRequest{
|
|
SiteID: "remark",
|
|
User: "testUser",
|
|
Email: "test@example.org",
|
|
Token: "testToken",
|
|
}) // consumed, consumer blocks in SendVerification on the gate
|
|
synctest.Wait()
|
|
s.SubmitVerification(VerificationRequest{User: "second"}) // fills the size-1 queue
|
|
s.SubmitVerification(VerificationRequest{User: "dropped"}) // queue full, dropped
|
|
synctest.Wait()
|
|
|
|
close(gate) // release the consumer: it finishes testUser then processes second
|
|
synctest.Wait()
|
|
s.Close()
|
|
|
|
s.SubmitVerification(VerificationRequest{}) // safe to send after close
|
|
|
|
require.Len(t, d2.GetVerify(), 2, "one request of three dropped from d2, got: %v", d2.GetVerify())
|
|
|
|
verifyDest := d1.GetVerify()
|
|
require.Len(t, verifyDest, 2, "one request of three dropped from d1, got: %v", verifyDest)
|
|
assert.Equal(t, "remark", verifyDest[0].SiteID)
|
|
assert.Equal(t, "testUser", verifyDest[0].User)
|
|
assert.Equal(t, "test@example.org", verifyDest[0].Email)
|
|
assert.Equal(t, "testToken", verifyDest[0].Token)
|
|
assert.Equal(t, "second", verifyDest[1].User)
|
|
})
|
|
}
|
|
|
|
func TestService_Many(t *testing.T) {
|
|
synctest.Test(t, func(t *testing.T) {
|
|
d1, d2 := &MockDest{id: 1}, &MockDest{id: 2}
|
|
s := NewService(nil, 5, d1, d2)
|
|
assert.NotNil(t, s)
|
|
|
|
for i := range 10 {
|
|
s.Submit(Request{Comment: store.Comment{ID: fmt.Sprintf("%d", 100+i)}})
|
|
s.SubmitVerification(VerificationRequest{User: fmt.Sprintf("%d", 100+i)})
|
|
}
|
|
s.Close()
|
|
|
|
assert.NotEqual(t, 10, len(d1.Get()), "some comments dropped from d1")
|
|
assert.NotEqual(t, 10, len(d1.GetVerify()), "some verifications dropped from d1")
|
|
assert.NotEqual(t, 10, len(d2.Get()), "some comments dropped from d2")
|
|
assert.NotEqual(t, 10, len(d2.GetVerify()), "some verifications dropped from d2")
|
|
})
|
|
}
|
|
|
|
func TestService_WithParent(t *testing.T) {
|
|
synctest.Test(t, func(t *testing.T) {
|
|
dest := &MockDest{id: 1}
|
|
dataStore := &mockStore{data: map[string]store.Comment{}}
|
|
|
|
dataStore.data["p1"] = store.Comment{ID: "p1"}
|
|
dataStore.data["p2"] = store.Comment{ID: "p2"}
|
|
|
|
s := NewService(dataStore, 1, dest)
|
|
assert.NotNil(t, s)
|
|
|
|
s.Submit(Request{Comment: store.Comment{ID: "c1", ParentID: "p1"}})
|
|
synctest.Wait()
|
|
s.Submit(Request{Comment: store.Comment{ID: "c11", ParentID: "p11"}})
|
|
synctest.Wait()
|
|
s.Close()
|
|
|
|
destRes := dest.Get()
|
|
require.Equal(t, 2, len(destRes), "two comment notified")
|
|
assert.Equal(t, "p1", destRes[0].Comment.ParentID)
|
|
assert.Equal(t, "p1", destRes[0].parent.ID)
|
|
assert.Equal(t, "p11", destRes[1].Comment.ParentID)
|
|
assert.Equal(t, "", destRes[1].parent.ID)
|
|
})
|
|
}
|
|
|
|
func TestService_EmailRetrieval(t *testing.T) {
|
|
synctest.Test(t, func(t *testing.T) {
|
|
dest := &MockDest{id: 1}
|
|
dataStore := &mockStore{data: map[string]store.Comment{}, userDetails: map[string]string{}}
|
|
|
|
dataStore.data["p1"] = store.Comment{ID: "p1", User: store.User{ID: "u1"}}
|
|
dataStore.data["p2"] = store.Comment{ID: "p2", ParentID: "p1", User: store.User{ID: "u1"}}
|
|
dataStore.data["p3"] = store.Comment{ID: "p3", ParentID: "p1", User: store.User{ID: "u2"}}
|
|
dataStore.data["p4"] = store.Comment{ID: "p4", ParentID: "p3", User: store.User{ID: "u1"}}
|
|
dataStore.userDetails["u1"] = "u1@example.com"
|
|
|
|
s := NewService(dataStore, 1, dest)
|
|
assert.NotNil(t, s)
|
|
|
|
// one comment, one notification
|
|
s.Submit(Request{Comment: dataStore.data["p1"]})
|
|
synctest.Wait()
|
|
|
|
destRes := dest.Get()
|
|
require.Equal(t, 1, len(destRes), "one comment notified")
|
|
assert.Equal(t, "p1", destRes[0].Comment.ID)
|
|
assert.Empty(t, destRes[0].parent)
|
|
assert.Empty(t, destRes[0].Emails)
|
|
|
|
// reply to the first comment, same comment as one in original comment
|
|
s.Submit(Request{Comment: dataStore.data["p2"]})
|
|
synctest.Wait()
|
|
|
|
destRes = dest.Get()
|
|
require.Equal(t, 2, len(destRes), "two comment notified")
|
|
assert.Equal(t, "p2", destRes[1].Comment.ID)
|
|
assert.Equal(t, "p1", destRes[1].parent.ID)
|
|
assert.Equal(t, "u1", destRes[1].parent.User.ID)
|
|
assert.Empty(t, destRes[1].Emails, "u1 is not notified they are the one who left the comment")
|
|
|
|
// another reply to the first comment, another user
|
|
s.Submit(Request{Comment: dataStore.data["p3"]})
|
|
synctest.Wait()
|
|
|
|
destRes = dest.Get()
|
|
require.Equal(t, 3, len(destRes), "three comment notified")
|
|
assert.Equal(t, "p3", destRes[2].Comment.ID)
|
|
assert.Equal(t, "p1", destRes[2].parent.ID)
|
|
assert.Equal(t, "u1", destRes[2].parent.User.ID)
|
|
assert.ElementsMatch(t, []string{"u1@example.com"}, destRes[2].Emails)
|
|
|
|
// reply to the last comment by another user, should trigger email retrieval error
|
|
s.Submit(Request{Comment: dataStore.data["p4"]})
|
|
synctest.Wait()
|
|
|
|
destRes = dest.Get()
|
|
require.Equal(t, 4, len(destRes), "four comment notified")
|
|
assert.Equal(t, "p4", destRes[3].Comment.ID)
|
|
assert.Equal(t, "p3", destRes[3].parent.ID)
|
|
assert.Equal(t, "u2", destRes[3].parent.User.ID)
|
|
assert.Empty(t, destRes[3].Emails, "no email can be retrieved for u2")
|
|
|
|
s.Close()
|
|
})
|
|
}
|
|
|
|
func TestService_Recursive(t *testing.T) {
|
|
synctest.Test(t, func(t *testing.T) {
|
|
dest := &MockDest{id: 1}
|
|
dataStore := &mockStore{data: map[string]store.Comment{}, userDetails: map[string]string{}}
|
|
|
|
dataStore.data["p1"] = store.Comment{ID: "p1", User: store.User{ID: "u1"}}
|
|
dataStore.data["p2"] = store.Comment{ID: "p2", ParentID: "p1", User: store.User{ID: "u2"}}
|
|
dataStore.data["p3"] = store.Comment{ID: "p3", ParentID: "p2", User: store.User{ID: "u3"}}
|
|
dataStore.data["p4"] = store.Comment{ID: "p4", ParentID: "p3", User: store.User{ID: "u1"}}
|
|
dataStore.data["p5"] = store.Comment{ID: "p5", ParentID: "p4", User: store.User{ID: "u4"}}
|
|
dataStore.userDetails["u1"] = "u1@example.com"
|
|
// second comment goes without email address for notification
|
|
dataStore.userDetails["u3"] = "u3@example.com"
|
|
|
|
s := NewService(dataStore, 1, dest)
|
|
assert.NotNil(t, s)
|
|
|
|
// one comment from u1 with email set
|
|
s.Submit(Request{Comment: dataStore.data["p1"]})
|
|
synctest.Wait()
|
|
|
|
destRes := dest.Get()
|
|
require.Equal(t, 1, len(destRes), "one comment notified")
|
|
assert.Equal(t, "p1", destRes[0].Comment.ID)
|
|
assert.Empty(t, destRes[0].parent)
|
|
assert.Empty(t, destRes[0].Emails)
|
|
|
|
// reply to the first comment from u2 without email set
|
|
s.Submit(Request{Comment: dataStore.data["p2"]})
|
|
synctest.Wait()
|
|
|
|
destRes = dest.Get()
|
|
require.Equal(t, 2, len(destRes), "two comment notified")
|
|
assert.Equal(t, "p2", destRes[1].Comment.ID)
|
|
assert.Equal(t, "p1", destRes[1].parent.ID)
|
|
assert.Equal(t, "u1", destRes[1].parent.User.ID)
|
|
assert.ElementsMatch(t, []string{"u1@example.com"}, destRes[1].Emails)
|
|
|
|
// reply to the second comment from u3 with email set
|
|
s.Submit(Request{Comment: dataStore.data["p3"]})
|
|
synctest.Wait()
|
|
|
|
destRes = dest.Get()
|
|
require.Equal(t, 3, len(destRes), "three comment notified")
|
|
assert.Equal(t, "p3", destRes[2].Comment.ID)
|
|
assert.Equal(t, "p2", destRes[2].parent.ID)
|
|
assert.Equal(t, "u2", destRes[2].parent.User.ID)
|
|
assert.ElementsMatch(t, []string{"u1@example.com"}, destRes[2].Emails)
|
|
|
|
// reply to the third comment from u1 (author of the first comment), only u3 should be notified
|
|
s.Submit(Request{Comment: dataStore.data["p4"]})
|
|
synctest.Wait()
|
|
|
|
destRes = dest.Get()
|
|
require.Equal(t, 4, len(destRes), "four comment notified once each")
|
|
assert.Equal(t, "p4", destRes[3].Comment.ID)
|
|
assert.Equal(t, "p3", destRes[3].parent.ID)
|
|
assert.Equal(t, "u3", destRes[3].parent.User.ID)
|
|
assert.ElementsMatch(t, []string{"u3@example.com"}, destRes[3].Emails, "u1 is not notified they are the one who left the comment")
|
|
|
|
// reply to the fourth comment from u4, u1 and u3 should be notified once as a result
|
|
s.Submit(Request{Comment: dataStore.data["p5"]})
|
|
synctest.Wait()
|
|
|
|
destRes = dest.Get()
|
|
require.Equal(t, 5, len(destRes), "four comment notified once each")
|
|
assert.Equal(t, "p5", destRes[4].Comment.ID)
|
|
assert.Equal(t, "p4", destRes[4].parent.ID)
|
|
assert.Equal(t, "u1", destRes[4].parent.User.ID)
|
|
assert.ElementsMatch(t, []string{"u1@example.com", "u3@example.com"}, destRes[4].Emails, "u3 and u1 notified once")
|
|
|
|
s.Close()
|
|
})
|
|
}
|
|
|
|
func TestService_Nop(t *testing.T) {
|
|
s := NopService
|
|
s.Submit(Request{Comment: store.Comment{}})
|
|
s.Close()
|
|
assert.Equal(t, uint32(1), atomic.LoadUint32(&s.closed))
|
|
}
|
|
|
|
type mockStore struct {
|
|
data map[string]store.Comment
|
|
userDetails map[string]string
|
|
}
|
|
|
|
func (m mockStore) getUserDetail(userID string) (string, error) {
|
|
detail, ok := m.userDetails[userID]
|
|
if !ok {
|
|
return "", fmt.Errorf("no such user")
|
|
}
|
|
return detail, nil
|
|
}
|
|
|
|
func (m mockStore) Get(_ store.Locator, id string, _ store.User) (store.Comment, error) {
|
|
res, ok := m.data[id]
|
|
if !ok {
|
|
return store.Comment{}, fmt.Errorf("no such id")
|
|
}
|
|
return res, nil
|
|
}
|
|
|
|
func (m mockStore) GetUserEmail(_, userID string) (string, error) {
|
|
return m.getUserDetail(userID)
|
|
}
|
|
|
|
func (m mockStore) GetUserTelegram(_, userID string) (string, error) {
|
|
return m.getUserDetail(userID)
|
|
}
|