2c884c5612
Некоторые проверки не удались
CI / test (push) Successful in 2m5s
Docker / Build and publish worker image (push) Failing after 31s
234 строки
8.8 KiB
Go
234 строки
8.8 KiB
Go
package distworker
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
|
|
"rocketgit.ru/rsmon/worker/app/models"
|
|
"rocketgit.ru/rsmon/worker/internal/wire"
|
|
)
|
|
|
|
// runnerWithCreds is the smallest fixture that yields a Runner with a
|
|
// credentials block already applied (no DB, no init websocket).
|
|
func runnerWithCreds(creds *wire.NotificationCredentials) *Runner {
|
|
r := NewRunner(&Config{MaxConcurrency: 4})
|
|
r.credentialsMu.Lock()
|
|
r.credentials = creds
|
|
r.credentialsMu.Unlock()
|
|
return r
|
|
}
|
|
|
|
// TestExecuteNotification_Email_NoCredentials verifies that the executor
|
|
// returns a permanent failure when no SMTP credential is pushed.
|
|
func TestExecuteNotification_Email_NoCredentials(t *testing.T) {
|
|
r := runnerWithCreds(nil)
|
|
report := r.ExecuteNotification(context.Background(), models.Task{
|
|
JobID: "job-1",
|
|
Payload: []byte(`{
|
|
"job_id":"job-1","message_id":1,"notification_id":1,
|
|
"method":"email","subject":"x","body_text":"x","body_html":"<p>x</p>",
|
|
"contact":{"id":1,"kind":"email","value":"ops@example.com","name":"ops"},
|
|
"message_kind":"down"
|
|
}`),
|
|
})
|
|
assert.Equal(t, wire.NotificationResultPermanent, report.Status)
|
|
assert.NotNil(t, report.Error)
|
|
assert.Equal(t, wire.NotificationResultPermanent, report.Status, "must be permanent so the producer does not loop")
|
|
}
|
|
|
|
func TestClassifyResult_NilEmailErrorDoesNotPanic(t *testing.T) {
|
|
status, _, text, retry, err := classifyResult(nil, "")
|
|
assert.Equal(t, wire.NotificationResultDelivered, status)
|
|
assert.Empty(t, text)
|
|
assert.Nil(t, retry)
|
|
assert.NoError(t, err)
|
|
}
|
|
|
|
// TestExecuteNotification_UnsupportedMethod covers sms/voice returning
|
|
// permanent + unsupported_method.
|
|
func TestExecuteNotification_UnsupportedMethod(t *testing.T) {
|
|
r := runnerWithCreds(&wire.NotificationCredentials{
|
|
SMTP: []wire.SMTPCredential{{Server: "smtp.example.com", Port: 587, Login: "u", Password: "p"}},
|
|
Telegram: []wire.TelegramCredential{{Name: "bot", Token: "123:abc"}},
|
|
})
|
|
for _, method := range []string{"sms", "voice"} {
|
|
t.Run(method, func(t *testing.T) {
|
|
report := r.ExecuteNotification(context.Background(), models.Task{
|
|
JobID: "job-" + method,
|
|
Payload: []byte(`{
|
|
"job_id":"job-` + method + `","message_id":1,"notification_id":1,
|
|
"method":"` + method + `","subject":"x","body_text":"x","body_html":"<p>x</p>",
|
|
"contact":{"id":1,"kind":"` + method + `","value":"x","name":"ops"},
|
|
"message_kind":"down"
|
|
}`),
|
|
})
|
|
assert.Equal(t, wire.NotificationResultPermanent, report.Status)
|
|
require.NotNil(t, report.Error)
|
|
assert.Contains(t, *report.Error, "unsupported_method")
|
|
})
|
|
}
|
|
}
|
|
|
|
// TestExecuteNotification_UnknownMethod classifies unknown methods as
|
|
// permanent so we never loop on bad producer output.
|
|
func TestExecuteNotification_UnknownMethod(t *testing.T) {
|
|
r := runnerWithCreds(&wire.NotificationCredentials{})
|
|
report := r.ExecuteNotification(context.Background(), models.Task{
|
|
JobID: "job-x",
|
|
Payload: []byte(`{
|
|
"job_id":"job-x","message_id":1,"notification_id":1,
|
|
"method":"pigeon","subject":"x","body_text":"x","body_html":"<p>x</p>",
|
|
"contact":{"id":1,"kind":"pigeon","value":"ops@example.com","name":"ops"},
|
|
"message_kind":"down"
|
|
}`),
|
|
})
|
|
assert.Equal(t, wire.NotificationResultPermanent, report.Status)
|
|
}
|
|
|
|
// TestExecuteNotification_EmptyPayloadPermanent: a malformed task must be
|
|
// rejected permanently so the operator can spot it on the admin page.
|
|
func TestExecuteNotification_EmptyPayloadPermanent(t *testing.T) {
|
|
r := runnerWithCreds(&wire.NotificationCredentials{
|
|
SMTP: []wire.SMTPCredential{{Server: "smtp.example.com", Port: 587, Login: "u", Password: "p"}},
|
|
})
|
|
report := r.ExecuteNotification(context.Background(), models.Task{JobID: "job-empty"})
|
|
assert.Equal(t, wire.NotificationResultPermanent, report.Status)
|
|
require.NotNil(t, report.Error)
|
|
}
|
|
|
|
func TestExecuteNotification_ExpiredDeadlineDoesNotDeliver(t *testing.T) {
|
|
r := runnerWithCreds(&wire.NotificationCredentials{})
|
|
expired := time.Now().Add(-time.Second)
|
|
report := r.ExecuteNotification(context.Background(), models.Task{JobID: "expired", Deadline: &expired, Payload: []byte(`{"job_id":"expired","message_id":1,"method":"email","contact":{"id":1,"kind":"email"}}`)})
|
|
require.NotNil(t, report.Error)
|
|
assert.Contains(t, *report.Error, "deadline expired")
|
|
assert.Equal(t, wire.NotificationResultPermanent, report.Status)
|
|
}
|
|
|
|
// TestExecuteNotification_DurationPositive: every report carries a non-zero
|
|
// duration_ms even when the work happens instantly. This is what the
|
|
// notification_deliveries audit row expects.
|
|
func TestExecuteNotification_DurationPositive(t *testing.T) {
|
|
r := runnerWithCreds(&wire.NotificationCredentials{})
|
|
report := r.ExecuteNotification(context.Background(), models.Task{
|
|
JobID: "job-d",
|
|
Payload: []byte(`{
|
|
"job_id":"job-d","message_id":1,"notification_id":1,
|
|
"method":"sms","subject":"x","body_text":"x","body_html":"<p>x</p>",
|
|
"contact":{"id":1,"kind":"sms","value":"x","name":"ops"},
|
|
"message_kind":"down"
|
|
}`),
|
|
})
|
|
assert.GreaterOrEqual(t, report.DurationMs, 0)
|
|
}
|
|
|
|
// TestClassifyResult verifies the error -> status mapping.
|
|
func TestClassifyResult(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
err error
|
|
wantStatus string
|
|
wantRetryAfter *int
|
|
}{
|
|
{"nil error", nil, wire.NotificationResultDelivered, nil},
|
|
{"permanent provider error", errors.New("550 mailbox not found"), wire.NotificationResultPermanent, nil},
|
|
{"smtp 421 retryable", errors.New("smtp: 421 try again later"), wire.NotificationResultRetryable, intPtr(30)},
|
|
{"smtp 452 retryable", errors.New("452 insufficient storage"), wire.NotificationResultRetryable, intPtr(30)},
|
|
{"network timeout retryable", errors.New("dial tcp: i/o timeout"), wire.NotificationResultRetryable, intPtr(15)},
|
|
{"tls handshake retryable", errors.New("tls: handshake failure"), wire.NotificationResultRetryable, intPtr(15)},
|
|
}
|
|
for _, tc := range tests {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
status, _, errStr, retry, _ := classifyResult(tc.err, "")
|
|
assert.Equal(t, tc.wantStatus, status)
|
|
if tc.err != nil {
|
|
assert.NotEmpty(t, errStr)
|
|
}
|
|
if tc.wantRetryAfter == nil {
|
|
assert.Nil(t, retry)
|
|
} else {
|
|
require.NotNil(t, retry)
|
|
assert.Equal(t, *tc.wantRetryAfter, *retry)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// TestWireTelegramToModel_RoundTrip sanity-checks the wire->DB credential
|
|
// translation used by the telegram branch.
|
|
func TestWireTelegramToModel_RoundTrip(t *testing.T) {
|
|
wire := &wire.TelegramCredential{
|
|
ID: 1,
|
|
Name: "main-bot",
|
|
BotName: "rsmon_bot",
|
|
Token: "123456:ABCDEFG",
|
|
APIURL: "https://api.telegram.org",
|
|
}
|
|
nc := wireTelegramToModel(wire)
|
|
require.NotNil(t, nc)
|
|
assert.Equal(t, models.CredentialKindTelegram, nc.Kind)
|
|
assert.Equal(t, "main-bot", nc.Name)
|
|
require.NotNil(t, nc.BotName)
|
|
assert.Equal(t, "rsmon_bot", *nc.BotName)
|
|
require.NotNil(t, nc.APIURL)
|
|
assert.Equal(t, "https://api.telegram.org", *nc.APIURL)
|
|
got, err := nc.GetSecret()
|
|
require.NoError(t, err)
|
|
assert.Equal(t, "123456:ABCDEFG", got)
|
|
}
|
|
|
|
func TestSelectNotificationCredentialByID(t *testing.T) {
|
|
id := int64(2)
|
|
smtp, ok := selectSMTPCredential([]wire.SMTPCredential{{ID: 1, Name: "first"}, {ID: 2, Name: "second"}}, &id)
|
|
require.True(t, ok)
|
|
assert.Equal(t, "second", smtp.Name)
|
|
|
|
tg, ok := selectTelegramCredential([]wire.TelegramCredential{{ID: 1, Name: "first"}, {ID: 2, Name: "second"}}, &id)
|
|
require.True(t, ok)
|
|
assert.Equal(t, "second", tg.Name)
|
|
|
|
missingID := int64(3)
|
|
_, ok = selectSMTPCredential([]wire.SMTPCredential{{ID: 1, Name: "first"}}, &missingID)
|
|
assert.False(t, ok)
|
|
}
|
|
|
|
// TestEnqueueNotification_BoundedQueue ensures the notifyQueue provides
|
|
// backpressure: the channel capacity is queueCapacity() and EnqueueNotification
|
|
// blocks once it is full.
|
|
func TestEnqueueNotification_BoundedQueue(t *testing.T) {
|
|
r := NewRunner(&Config{MaxConcurrency: 4})
|
|
r.notifyQueue = make(chan wire.NotificationTask, 4)
|
|
|
|
for i := 0; i < 4; i++ {
|
|
require.True(t, r.EnqueueNotification(wire.NotificationTask{JobID: "x"}))
|
|
}
|
|
// Channel is full; a non-blocking send must fail. We cannot truly verify
|
|
// the blocking case in unit tests, so we just assert the depth counter.
|
|
assert.Equal(t, int64(4), r.notifyDepth)
|
|
}
|
|
|
|
// TestExecuteNotification_ReportsJobIDFromPayload verifies the executor
|
|
// falls back to the task's JobID when the payload has none.
|
|
func TestExecuteNotification_ReportsJobIDFromPayload(t *testing.T) {
|
|
r := runnerWithCreds(nil)
|
|
report := r.ExecuteNotification(context.Background(), models.Task{
|
|
JobID: "outer-job",
|
|
Payload: []byte(`{
|
|
"job_id":"","message_id":1,"notification_id":1,
|
|
"method":"sms","subject":"x","body_text":"x","body_html":"<p>x</p>",
|
|
"contact":{"id":1,"kind":"sms","value":"x","name":"ops"},
|
|
"message_kind":"down"
|
|
}`),
|
|
})
|
|
assert.Equal(t, "outer-job", report.JobID)
|
|
}
|
|
|
|
// guard against time import being unused if the above compile-time helpers
|
|
// are dropped in a future refactor.
|
|
var _ = time.Second
|