package queue
import (
"bytes"
"fmt"
"strings"
"testing"
"testing/synctest"
"time"
"blitiri.com.ar/go/chasquid/internal/aliases"
"blitiri.com.ar/go/chasquid/internal/set"
"blitiri.com.ar/go/chasquid/internal/testlib"
"blitiri.com.ar/go/chasquid/internal/trace"
)
func allUsersExist(tr *trace.Trace, user, domain string) (bool, error) {
return true, nil
}
func TestBasic(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
dir := t.ArtifactDir()
localC := testlib.NewTestCourier()
remoteC := testlib.NewTestCourier()
q, _ := New(dir, set.NewString("loco"),
aliases.NewResolver(allUsersExist),
localC, remoteC)
tr := trace.New("test", "TestBasic")
defer tr.Finish()
localC.Expect(2)
remoteC.Expect(1)
id, err := q.Put(tr, "from", []string{"am@loco", "x@remote", "nodomain"}, []byte("data"))
if err != nil {
t.Fatalf("Put: %v", err)
}
if len(id) < 6 {
t.Errorf("short ID: %v", id)
}
localC.Wait()
remoteC.Wait()
// Wait for the send loop to finish, and make sure the delivered items
// leave the queue.
synctest.Wait()
if q.Len() != 0 {
t.Fatalf("%d items not removed from the queue after delivery", q.Len())
}
cases := []struct {
courier *testlib.TestCourier
expectedTo string
}{
{localC, "nodomain"},
{localC, "am@loco"},
{remoteC, "x@remote"},
}
for _, c := range cases {
req := c.courier.ReqFor[c.expectedTo]
if req == nil {
t.Errorf("missing request for %q", c.expectedTo)
continue
}
if req.From != "from" || req.To != c.expectedTo ||
!bytes.Equal(req.Data, []byte("data")) {
t.Errorf("wrong request for %q: %v", c.expectedTo, req)
}
}
})
}
func TestDSNOnTimeout(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
localC := testlib.NewTestCourier()
remoteC := testlib.NewTestCourier()
dir := t.ArtifactDir()
q, _ := New(dir, set.NewString("loco"),
aliases.NewResolver(allUsersExist),
localC, remoteC)
// Insert an expired item in the queue.
item := &Item{
Message: Message{
ID: <-newID,
From: "from@loco",
Rcpt: []*Recipient{
mkR("to@to", Recipient_EMAIL, Recipient_PENDING, "err", "to@to")},
Data: []byte("data"),
},
CreatedAt: time.Now().Add(-24 * time.Hour),
}
q.q[item.ID] = item
err := item.WriteTo(q.path)
if err != nil {
t.Errorf("failed to write item: %v", err)
}
// Exercise DumpString while at it.
q.DumpString()
// Launch the sending loop, expect 1 local delivery (the DSN).
localC.Expect(1)
go item.SendLoop(q)
localC.Wait()
req := localC.ReqFor["from@loco"]
if req == nil {
t.Fatal("missing DSN")
}
if req.From != "<>" || req.To != "from@loco" ||
!strings.Contains(string(req.Data), "X-Failed-Recipients: to@to,") {
t.Errorf("wrong DSN: %q", string(req.Data))
}
// Wait for the send loops to finish, so they don't race with the test
// directory cleanup.
synctest.Wait()
})
}
func TestAliases(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
localC := testlib.NewTestCourier()
remoteC := testlib.NewTestCourier()
dir := t.ArtifactDir()
q, _ := New(dir, set.NewString("loco"),
aliases.NewResolver(allUsersExist),
localC, remoteC)
tr := trace.New("test", "TestAliases")
defer tr.Finish()
q.aliases.AddDomain("loco")
q.aliases.AddAliasForTesting("ab@loco", "pq@loco", nil, aliases.EMAIL)
q.aliases.AddAliasForTesting("ab@loco", "rs@loco", nil, aliases.EMAIL)
q.aliases.AddAliasForTesting("cd@loco", "ata@hualpa", nil, aliases.EMAIL)
q.aliases.AddAliasForTesting(
"fwd@loco", "fwd@loco", []string{"server"}, aliases.FORWARD)
q.aliases.AddAliasForTesting(
"remote@loco", "remote@rana", []string{"server"}, aliases.FORWARD)
// Note the pipe aliases are tested below, as they don't use the couriers
// and it can be quite inconvenient to test them in this way.
localC.Expect(2)
remoteC.Expect(3)
// One email from a local domain: from@loco -> ab@loco, cd@loco, fwd@loco.
_, err := q.Put(tr, "from@loco",
[]string{"ab@loco", "cd@loco", "fwd@loco"},
[]byte("data"))
if err != nil {
t.Fatalf("Put: %v", err)
}
// And another from a remote domain: from@rana -> remote@loco
_, err = q.Put(tr, "from@rana",
[]string{"remote@loco"},
[]byte("data"))
if err != nil {
t.Fatalf("Put: %v", err)
}
localC.Wait()
remoteC.Wait()
cases := []struct {
courier *testlib.TestCourier
expectedFrom string
expectedTo string
}{
// From the local domain: from@loco
{localC, "from@loco", "pq@loco"},
{localC, "from@loco", "rs@loco"},
{remoteC, "from@loco", "ata@hualpa"},
{remoteC, "from@loco", "fwd@loco"},
// From the remote domain: from@rana.
// Note the SRS in the remoteC.
{remoteC, "remote+fwd_from=from=rana@loco", "remote@rana"},
}
for _, c := range cases {
req := c.courier.ReqFor[c.expectedTo]
if req == nil {
t.Errorf("missing request for %q", c.expectedTo)
continue
}
if req.From != c.expectedFrom || req.To != c.expectedTo ||
!bytes.Equal(req.Data, []byte("data")) {
t.Errorf("wrong request for %q: %v", c.expectedTo, *req)
}
}
// Wait for the send loops to finish, so they don't race with the test
// directory cleanup.
synctest.Wait()
})
}
func TestRewriteSender(t *testing.T) {
cases := []struct {
from, originalAddr, expected string
}{
{"from@rana", "remote@loco", "remote+fwd_from=from=rana@loco"},
// The domain of the original address is converted to ASCII.
{"from@rana", "remote@ñoño", "remote+fwd_from=from=rana@xn--oo-yjab"},
// If the conversion fails, the domain is left as-is.
{"from@rana", "remote@xn---", "remote+fwd_from=from=rana@xn---"},
}
for _, c := range cases {
got := rewriteSender(c.from, c.originalAddr)
if got != c.expected {
t.Errorf("rewriteSender(%q, %q) = %q, expected %q",
c.from, c.originalAddr, got, c.expected)
}
}
}
func TestFullQueue(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
dir := t.ArtifactDir()
q, _ := New(dir, set.NewString(),
aliases.NewResolver(allUsersExist),
testlib.DumbCourier, testlib.DumbCourier)
tr := trace.New("test", "TestFullQueue")
defer tr.Finish()
// Force-insert as many items in the queue as it supports.
oneID := ""
for i := 0; i < q.MaxItems; i++ {
item := &Item{
Message: Message{
ID: <-newID,
From: fmt.Sprintf("from-%d", i),
Rcpt: []*Recipient{
mkR("to", Recipient_EMAIL, Recipient_PENDING, "", "")},
Data: []byte("data"),
},
CreatedAt: time.Now(),
}
q.q[item.ID] = item
oneID = item.ID
}
// This one should fail due to the queue being too big.
id, err := q.Put(tr, "from", []string{"to"}, []byte("data-qf"))
if err != errQueueFull {
t.Errorf("Not failed as expected: %v - %v", id, err)
}
// Remove one, and try again: it should succeed.
// Write it first so we don't get complaints about the file not existing
// (as we did not all the items properly).
q.q[oneID].WriteTo(q.path)
q.Remove(oneID)
id, err = q.Put(tr, "from", []string{"to"}, []byte("data"))
if err != nil {
t.Errorf("Put: %v", err)
}
q.Remove(id)
// Wait for the send loops to finish, so they don't race with the test
// directory cleanup.
synctest.Wait()
})
}
func TestPipes(t *testing.T) {
dir := t.ArtifactDir()
q, _ := New(dir, set.NewString("loco"),
aliases.NewResolver(allUsersExist),
testlib.DumbCourier, testlib.DumbCourier)
item := &Item{
Message: Message{
ID: <-newID,
From: "from",
Rcpt: []*Recipient{
mkR("true", Recipient_PIPE, Recipient_PENDING, "", "")},
Data: []byte("data"),
},
CreatedAt: time.Now(),
}
if err, _ := item.deliver(q, item.Rcpt[0]); err != nil {
t.Errorf("pipe delivery failed: %v", err)
}
}
func TestBadPath(t *testing.T) {
// A new queue will attempt to os.MkdirAll the path.
// We expect this path to fail.
_, err := New("/proc/doesnotexist", set.NewString("loco"),
aliases.NewResolver(allUsersExist),
testlib.DumbCourier, testlib.DumbCourier)
if err == nil {
t.Errorf("could create queue, expected permission denied")
}
}
func TestNextDelay(t *testing.T) {
cases := []struct{ since, min time.Duration }{
{10 * time.Second, 1 * time.Minute},
{3 * time.Minute, 5 * time.Minute},
{7 * time.Minute, 10 * time.Minute},
{15 * time.Minute, 20 * time.Minute},
{30 * time.Minute, 20 * time.Minute},
}
for _, c := range cases {
// Repeat each case a few times to exercise the perturbation a bit.
for range 10 {
delay := nextDelay(time.Now().Add(-c.since))
max := c.min + 1*time.Minute
if delay < c.min || delay > max {
t.Errorf("since:%v expected [%v, %v], got %v",
c.since, c.min, max, delay)
}
}
}
}
func TestSerialization(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
dir := t.ArtifactDir()
// Save an item in the queue directory.
item := &Item{
Message: Message{
ID: <-newID,
From: "from@loco",
Rcpt: []*Recipient{
mkR("to@to", Recipient_EMAIL, Recipient_PENDING, "err", "to@to")},
Data: []byte("data"),
},
CreatedAt: time.Now().Add(-1 * time.Hour),
}
err := item.WriteTo(dir)
if err != nil {
t.Errorf("failed to write item: %v", err)
}
// Create the queue; should load the
remoteC := testlib.NewTestCourier()
remoteC.Expect(1)
q, _ := New(dir, set.NewString("loco"),
aliases.NewResolver(allUsersExist),
testlib.DumbCourier, remoteC)
q.Load()
// Launch the sending loop, expect 1 remote delivery for the item we saved.
remoteC.Wait()
req := remoteC.ReqFor["to@to"]
if req == nil {
t.Fatal("email not delivered")
}
if req.From != "from@loco" || req.To != "to@to" {
t.Errorf("wrong email: %v", req)
}
// Wait for the send loops to finish, so they don't race with the test
// directory cleanup.
synctest.Wait()
})
}
func mkR(a string, t Recipient_Type, s Recipient_Status, m, o string) *Recipient {
return &Recipient{
Address: a,
Type: t,
Status: s,
LastFailureMessage: m,
OriginalAddress: o,
}
}