e987f24903
Все проверки выполнены успешно
CI / test (push) Successful in 2m32s
Docker / Build and publish worker image (push) Successful in 18m17s
- reconnect safely after token rotation and retry leased results - reject malformed tasks and remove production cluster debug mutation - validate environment files and require immutable container images BREAKING CHANGE: Docker install, deploy, and Compose now require an immutable repository@sha256 image reference.
408 строки
15 KiB
Go
408 строки
15 KiB
Go
package distworker
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"sync"
|
|
"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)
|
|
}
|
|
|
|
func TestDelegatedNotificationAuditCoexistsWithSelfcheckWithoutSecrets(t *testing.T) {
|
|
const (
|
|
recipient = "ops-private@example.com"
|
|
renderedBody = "private rendered body"
|
|
credentialSecret = "smtp-password-secret"
|
|
hookURL = "https://hooks.example.com/private"
|
|
authorization = "Bearer private-authorization"
|
|
providerResponse = "private provider response"
|
|
)
|
|
r := runnerWithCreds(&wire.NotificationCredentials{
|
|
SMTP: []wire.SMTPCredential{{Password: credentialSecret}},
|
|
})
|
|
r.notifyResults = make(chan notifyResultEnvelope, 1)
|
|
r.RecordNotification(&NotificationRow{
|
|
Kind: "email", Channel: notificationChannelSMTP, Subject: "selfcheck", Body: "main API down", OK: true,
|
|
})
|
|
r.executeAndForwardNotification(wire.NotificationTask{
|
|
JobID: "delegated-job", Method: "sms", Subject: authorization,
|
|
BodyText: renderedBody, BodyHTML: providerResponse,
|
|
Contact: wire.NotificationContact{Kind: "sms", Value: recipient, Name: hookURL},
|
|
})
|
|
|
|
rows := r.RecentNotifications(10)
|
|
require.Len(t, rows, 2)
|
|
assert.Equal(t, "selfcheck", rows[0].Subject)
|
|
delegated := rows[1]
|
|
assert.Equal(t, "delegated-job", delegated.JobID)
|
|
assert.Equal(t, "sms", delegated.Method)
|
|
assert.Equal(t, wire.NotificationResultPermanent, delegated.Status)
|
|
assert.Empty(t, delegated.Subject)
|
|
assert.Empty(t, delegated.Body)
|
|
assert.Empty(t, delegated.Error)
|
|
assert.False(t, delegated.At.IsZero())
|
|
|
|
stored, err := json.Marshal(delegated)
|
|
require.NoError(t, err)
|
|
for _, secret := range []string{recipient, renderedBody, credentialSecret, hookURL, authorization, providerResponse} {
|
|
assert.NotContains(t, string(stored), secret)
|
|
}
|
|
}
|
|
|
|
func TestDelegatedNotificationAuditRecordsOneTerminalRow(t *testing.T) {
|
|
const secret = "private-notification-secret"
|
|
validTask := func(jobID string) wire.NotificationTask {
|
|
return wire.NotificationTask{
|
|
JobID: jobID, LeaseToken: "lease", Method: "sms", Subject: secret, BodyText: secret,
|
|
Contact: wire.NotificationContact{Value: secret, Name: secret},
|
|
}
|
|
}
|
|
tests := []struct {
|
|
name string
|
|
invoke func(*Runner)
|
|
method string
|
|
}{
|
|
{
|
|
name: "normal result",
|
|
invoke: func(r *Runner) { r.executeAndForwardNotification(validTask("normal")) },
|
|
method: "sms",
|
|
},
|
|
{
|
|
name: "invalid deadline",
|
|
invoke: func(r *Runner) {
|
|
task := validTask("invalid-deadline")
|
|
deadline := "not-a-timestamp"
|
|
task.Deadline = &deadline
|
|
r.executeAndForwardNotification(task)
|
|
},
|
|
method: "sms",
|
|
},
|
|
{
|
|
name: "expired deadline",
|
|
invoke: func(r *Runner) {
|
|
task := validTask("expired-deadline")
|
|
deadline := time.Now().Add(-time.Second).Format(time.RFC3339Nano)
|
|
task.Deadline = &deadline
|
|
r.executeAndForwardNotification(task)
|
|
},
|
|
method: "sms",
|
|
},
|
|
{
|
|
name: "malformed payload",
|
|
invoke: func(r *Runner) {
|
|
r.ExecuteNotification(context.Background(), models.Task{JobID: "malformed-payload", Payload: []byte(`{"subject":"` + secret)})
|
|
},
|
|
},
|
|
{
|
|
name: "executor panic",
|
|
invoke: func(r *Runner) {
|
|
r.notificationExecutor = func(context.Context, models.Task) wire.NotificationResultReport {
|
|
panic(secret)
|
|
}
|
|
r.ExecuteNotification(context.Background(), models.Task{JobID: "panic", Payload: []byte(`{}`)})
|
|
},
|
|
},
|
|
{
|
|
name: "malformed envelope",
|
|
invoke: func(r *Runner) {
|
|
r.enqueueTaskMessage(wire.WorkerMessage{TaskEnvelope: &wire.TaskEnvelope{
|
|
Type: "invalid", JobID: "bad-envelope", Notify: ptrNotificationTask(validTask("bad-envelope")),
|
|
}})
|
|
},
|
|
method: "sms",
|
|
},
|
|
}
|
|
for _, tc := range tests {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
r := runnerWithCreds(&wire.NotificationCredentials{})
|
|
r.notifyResults = make(chan notifyResultEnvelope, 1)
|
|
tc.invoke(r)
|
|
|
|
rows := r.RecentNotifications(10)
|
|
require.Len(t, rows, 1)
|
|
assert.Equal(t, wire.NotificationResultPermanent, rows[0].Status)
|
|
assert.Equal(t, tc.method, rows[0].Method)
|
|
stored, err := json.Marshal(rows[0])
|
|
require.NoError(t, err)
|
|
assert.NotContains(t, string(stored), secret)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestDelegatedNotificationAuditConcurrentRowsAreSecretFree(t *testing.T) {
|
|
const attempts = 32
|
|
const secret = "concurrent-private-secret"
|
|
r := runnerWithCreds(&wire.NotificationCredentials{})
|
|
var wg sync.WaitGroup
|
|
for i := 0; i < attempts; i++ {
|
|
wg.Add(1)
|
|
go func(i int) {
|
|
defer wg.Done()
|
|
r.ExecuteNotification(context.Background(), models.Task{
|
|
JobID: fmt.Sprintf("job-%d", i),
|
|
Payload: []byte(`{"job_id":"job-` + fmt.Sprint(i) + `","method":"sms","body_text":"` + secret + `","contact":{"value":"` + secret + `"}}`),
|
|
})
|
|
}(i)
|
|
}
|
|
wg.Wait()
|
|
|
|
rows := r.RecentNotifications(attempts)
|
|
require.Len(t, rows, attempts)
|
|
for _, row := range rows {
|
|
stored, err := json.Marshal(row)
|
|
require.NoError(t, err)
|
|
assert.NotContains(t, string(stored), secret)
|
|
assert.Equal(t, wire.NotificationResultPermanent, row.Status)
|
|
}
|
|
}
|
|
|
|
func TestInvalidNotificationDeadlineIsNotExecuted(t *testing.T) {
|
|
r := runnerWithCreds(&wire.NotificationCredentials{})
|
|
r.notifyResults = make(chan notifyResultEnvelope, 1)
|
|
called := false
|
|
r.notificationExecutor = func(context.Context, models.Task) wire.NotificationResultReport {
|
|
called = true
|
|
return wire.NotificationResultReport{Status: wire.NotificationResultDelivered}
|
|
}
|
|
deadline := "not-a-timestamp"
|
|
r.executeAndForwardNotification(wire.NotificationTask{
|
|
JobID: "invalid-deadline", LeaseToken: "lease", Method: "sms", Deadline: &deadline,
|
|
})
|
|
|
|
assert.False(t, called)
|
|
env := <-r.notifyResults
|
|
assert.Equal(t, wire.NotificationResultPermanent, env.report.Status)
|
|
require.NotNil(t, env.report.Error)
|
|
assert.Equal(t, "invalid notification deadline", *env.report.Error)
|
|
require.Len(t, r.RecentNotifications(10), 1)
|
|
}
|
|
|
|
func ptrNotificationTask(task wire.NotificationTask) *wire.NotificationTask { return &task }
|
|
|
|
// guard against time import being unused if the above compile-time helpers
|
|
// are dropped in a future refactor.
|
|
var _ = time.Second
|