package distworker import ( "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "rocketgit.ru/rsmon/worker/internal/wire" ) func TestEnqueueTaskMessagePrefersEnvelopeOverLegacyCheck(t *testing.T) { r := &Runner{jobQueue: make(chan wire.CheckJob, 2), stopCh: make(chan struct{})} message := wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{Type: wire.TaskTypeCheck, JobID: "v2", Job: &wire.CheckJob{JobID: "v2", LeaseToken: "lease-2", Kind: "http"}}, Task: &wire.CheckJob{JobID: "v1"}} require.True(t, r.enqueueTaskMessage(message)) job := <-r.jobQueue assert.Equal(t, "v2", job.JobID) assert.Empty(t, r.jobQueue, "the sibling legacy field must not run a second check") } func TestEnqueueTaskMessageRejectsMismatchedEnvelopeJobIDs(t *testing.T) { r := &Runner{jobQueue: make(chan wire.CheckJob, 1), results: make(chan resultEnvelope, 1), stopCh: make(chan struct{})} message := wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{ Type: wire.TaskTypeCheck, JobID: "outer-job", Job: &wire.CheckJob{JobID: "inner-job", LeaseToken: "lease-1", Kind: "http"}, }} require.True(t, r.enqueueTaskMessage(message)) assert.Empty(t, r.jobQueue) assert.Empty(t, r.results, "a task without an unambiguous job ID cannot be reported safely") } func TestEnqueueTaskMessageRejectsMissingEnvelopeJobID(t *testing.T) { r := &Runner{jobQueue: make(chan wire.CheckJob, 1), results: make(chan resultEnvelope, 1), stopCh: make(chan struct{})} message := wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{ Type: wire.TaskTypeCheck, Job: &wire.CheckJob{JobID: "inner-job", LeaseToken: "lease-1", Kind: "http"}, }} require.True(t, r.enqueueTaskMessage(message)) assert.Empty(t, r.jobQueue) assert.Empty(t, r.results, "a missing outer ID cannot be reported safely") } func TestEnqueueTaskMessageReportsMalformedCheckEnvelopeWithMatchingJobID(t *testing.T) { r := &Runner{jobQueue: make(chan wire.CheckJob, 1), results: make(chan resultEnvelope, 1), stopCh: make(chan struct{})} message := wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{ Type: wire.TaskTypeNotification, JobID: "job-1", Job: &wire.CheckJob{JobID: "job-1", LeaseToken: "lease-1", CheckID: 4, MonitorID: 5, Kind: "http"}, }} require.True(t, r.enqueueTaskMessage(message)) assert.Empty(t, r.jobQueue) env := <-r.results require.Len(t, env.reports, 1) assert.Equal(t, "job-1", env.reports[0].JobID) assert.Equal(t, "FAIL", env.reports[0].State) require.NotNil(t, env.reports[0].Error) assert.Equal(t, malformedTaskEnvelopeError, *env.reports[0].Error) assert.Empty(t, r.results) } func TestEnqueueTaskMessageRejectsAmbiguousEnvelope(t *testing.T) { r := &Runner{ jobQueue: make(chan wire.CheckJob, 1), results: make(chan resultEnvelope, 1), notifyQueue: make(chan wire.NotificationTask, 1), notifyResults: make(chan notifyResultEnvelope, 1), stopCh: make(chan struct{}), } message := wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{ Type: wire.TaskTypeCheck, JobID: "job-1", Job: &wire.CheckJob{JobID: "job-1", LeaseToken: "lease-1", Kind: "http"}, Notify: &wire.NotificationTask{JobID: "job-1", LeaseToken: "lease-1"}, }} require.True(t, r.enqueueTaskMessage(message)) assert.Empty(t, r.jobQueue) assert.Empty(t, r.results) assert.Empty(t, r.notifyQueue) assert.Empty(t, r.notifyResults) } func TestEnqueueTaskMessageReportsUnsupportedCheckKind(t *testing.T) { r := &Runner{jobQueue: make(chan wire.CheckJob, 1), results: make(chan resultEnvelope, 1), stopCh: make(chan struct{})} message := wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{ Type: wire.TaskTypeCheck, JobID: "job-1", Job: &wire.CheckJob{JobID: "job-1", LeaseToken: "lease-1", CheckID: 4, MonitorID: 5, Kind: "rkn"}, }} require.True(t, r.enqueueTaskMessage(message)) assert.Empty(t, r.jobQueue, "unsupported work must never reach the executor") env := <-r.results require.Len(t, env.reports, 1) report := env.reports[0] assert.Equal(t, "job-1", report.JobID) assert.Equal(t, int64(4), report.CheckID) assert.Equal(t, int64(5), report.MonitorID) assert.Equal(t, "FAIL", report.State) require.NotNil(t, report.Error) assert.Equal(t, "unsupported_kind: rkn", *report.Error) assert.Empty(t, r.results, "each rejected task must generate one terminal result") } func TestEnqueueTaskMessageReportsMalformedNotificationForInvalidType(t *testing.T) { r := &Runner{notifyQueue: make(chan wire.NotificationTask, 1), notifyResults: make(chan notifyResultEnvelope, 1), stopCh: make(chan struct{})} message := wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{ Type: "unknown", JobID: "job-1", Notify: &wire.NotificationTask{JobID: "job-1", LeaseToken: "lease-1", MessageID: 9, Method: "email"}, }} require.True(t, r.enqueueTaskMessage(message)) assert.Empty(t, r.notifyQueue) env := <-r.notifyResults assert.Equal(t, "job-1", env.report.JobID) assert.Equal(t, "lease-1", env.report.LeaseToken) assert.Equal(t, int64(9), env.report.MessageID) assert.Equal(t, wire.NotificationResultPermanent, env.report.Status) require.NotNil(t, env.report.Error) assert.Equal(t, malformedTaskEnvelopeError, *env.report.Error) assert.Empty(t, r.notifyResults) } func TestEnqueueTaskMessageRejectsEmptyLeaseWithoutSideEffects(t *testing.T) { cases := []struct { name string message wire.WorkerMessage }{ { name: "envelope check", message: wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{ Type: wire.TaskTypeCheck, JobID: "job-1", Job: &wire.CheckJob{JobID: "job-1", Kind: "http"}, }}, }, { name: "envelope notification", message: wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{ Type: wire.TaskTypeNotification, JobID: "job-1", Notify: &wire.NotificationTask{JobID: "job-1", Method: "email"}, }}, }, { name: "legacy check", message: wire.WorkerMessage{Kind: "task", Task: &wire.CheckJob{JobID: "job-1", Kind: "http"}}, }, { name: "legacy notification", message: wire.WorkerMessage{Kind: "task", NotificationTask: &wire.NotificationTask{JobID: "job-1", Method: "email"}}, }, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { r := &Runner{ jobQueue: make(chan wire.CheckJob, 1), results: make(chan resultEnvelope, 1), notifyQueue: make(chan wire.NotificationTask, 1), notifyResults: make(chan notifyResultEnvelope, 1), stopCh: make(chan struct{}), } require.True(t, r.enqueueTaskMessage(tc.message)) assert.Empty(t, r.jobQueue) assert.Empty(t, r.results) assert.Empty(t, r.notifyQueue) assert.Empty(t, r.notifyResults) }) } }