package gocron import ( "context" "fmt" "sync" "testing" "time" "github.com/google/uuid" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "go.uber.org/goleak" ) func newTestScheduler(t *testing.T, options ...SchedulerOption) Scheduler { // default test options out := []SchedulerOption{ WithLogger(NewLogger(LogLevelDebug)), WithStopTimeout(time.Second), } // append any additional options 2nd to override defaults if needed out = append(out, options...) s, err := NewScheduler(out...) require.NoError(t, err) return s } func TestScheduler_OneSecond_NoOptions(t *testing.T) { defer goleak.VerifyNone(t) cronNoOptionsCh := make(chan struct{}, 10) durationNoOptionsCh := make(chan struct{}, 10) tests := []struct { name string ch chan struct{} jd JobDefinition tsk Task }{ { "cron", cronNoOptionsCh, CronJob( "* * * * * *", true, ), NewTask( func() { cronNoOptionsCh <- struct{}{} }, ), }, { "duration", durationNoOptionsCh, DurationJob( time.Second, ), NewTask( func() { durationNoOptionsCh <- struct{}{} }, ), }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { s := newTestScheduler(t) _, err := s.NewJob(tt.jd, tt.tsk) require.NoError(t, err) s.Start() startTime := time.Now() var runCount int for runCount < 1 { <-tt.ch runCount++ } require.NoError(t, s.Shutdown()) stopTime := time.Now() select { case <-tt.ch: t.Fatal("job ran after scheduler was stopped") case <-time.After(time.Millisecond * 50): } runDuration := stopTime.Sub(startTime) assert.GreaterOrEqual(t, runDuration, time.Millisecond) assert.LessOrEqual(t, runDuration, 1500*time.Millisecond) }) } } func TestScheduler_LongRunningJobs(t *testing.T) { defer goleak.VerifyNone(t) durationCh := make(chan struct{}, 10) durationSingletonCh := make(chan struct{}, 10) tests := []struct { name string ch chan struct{} jd JobDefinition tsk Task opts []JobOption options []SchedulerOption expectedRuns int }{ { "duration", durationCh, DurationJob( time.Millisecond * 500, ), NewTask( func() { time.Sleep(1 * time.Second) durationCh <- struct{}{} }, ), nil, []SchedulerOption{WithStopTimeout(time.Second * 2)}, 3, }, { "duration singleton", durationSingletonCh, DurationJob( time.Millisecond * 500, ), NewTask( func() { time.Sleep(1 * time.Second) durationSingletonCh <- struct{}{} }, ), []JobOption{WithSingletonMode(LimitModeWait)}, []SchedulerOption{WithStopTimeout(time.Second * 5)}, 2, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { s := newTestScheduler(t, tt.options...) _, err := s.NewJob(tt.jd, tt.tsk, tt.opts...) require.NoError(t, err) s.Start() time.Sleep(1600 * time.Millisecond) require.NoError(t, s.Shutdown()) var runCount int timeout := make(chan struct{}) go func() { time.Sleep(2 * time.Second) close(timeout) }() Outer: for { select { case <-tt.ch: runCount++ case <-timeout: break Outer } } assert.Equal(t, tt.expectedRuns, runCount) }) } } func TestScheduler_Update(t *testing.T) { defer goleak.VerifyNone(t) durationJobCh := make(chan struct{}) tests := []struct { name string initialJob JobDefinition updateJob JobDefinition tsk Task ch chan struct{} runCount int updateAfterCount int expectedMinTime time.Duration expectedMaxRunTime time.Duration }{ { "duration, updated to another duration", DurationJob( time.Millisecond * 500, ), DurationJob( time.Second, ), NewTask( func() { durationJobCh <- struct{}{} }, ), durationJobCh, 2, 1, time.Second * 1, time.Second * 2, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { s := newTestScheduler(t) j, err := s.NewJob(tt.initialJob, tt.tsk) require.NoError(t, err) startTime := time.Now() s.Start() var runCount int for runCount < tt.runCount { select { case <-tt.ch: runCount++ if runCount == tt.updateAfterCount { _, err = s.Update(j.ID(), tt.updateJob, tt.tsk) require.NoError(t, err) } default: } } require.NoError(t, s.Shutdown()) stopTime := time.Now() select { case <-tt.ch: t.Fatal("job ran after scheduler was stopped") case <-time.After(time.Millisecond * 50): } runDuration := stopTime.Sub(startTime) assert.GreaterOrEqual(t, runDuration, tt.expectedMinTime) assert.LessOrEqual(t, runDuration, tt.expectedMaxRunTime) }) } } func TestScheduler_StopTimeout(t *testing.T) { defer goleak.VerifyNone(t) tests := []struct { name string jd JobDefinition f any opts []JobOption }{ { "duration", DurationJob( time.Millisecond * 100, ), func(testDoneCtx context.Context) { select { case <-time.After(1 * time.Second): case <-testDoneCtx.Done(): } }, nil, }, { "duration singleton", DurationJob( time.Millisecond * 100, ), func(testDoneCtx context.Context) { select { case <-time.After(1 * time.Second): case <-testDoneCtx.Done(): } }, []JobOption{WithSingletonMode(LimitModeWait)}, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { testDoneCtx, cancel := context.WithCancel(context.Background()) s := newTestScheduler(t, WithStopTimeout(time.Millisecond*100), ) _, err := s.NewJob(tt.jd, NewTask(tt.f, testDoneCtx), tt.opts...) require.NoError(t, err) s.Start() assert.ErrorIs(t, err, s.Shutdown()) cancel() time.Sleep(2 * time.Second) }) } } func TestScheduler_Shutdown(t *testing.T) { goleak.VerifyNone(t) t.Run("start, stop, start, shutdown", func(t *testing.T) { s := newTestScheduler(t, WithStopTimeout(time.Second), ) _, err := s.NewJob( DurationJob( 50*time.Millisecond, ), NewTask( func() {}, ), WithStartAt( WithStartImmediately(), ), ) require.NoError(t, err) s.Start() require.NoError(t, s.StopJobs()) s.Start() require.NoError(t, s.Shutdown()) }) t.Run("calling Job methods after shutdown errors", func(t *testing.T) { s := newTestScheduler(t, WithStopTimeout(time.Second), ) j, err := s.NewJob( DurationJob( 100*time.Millisecond, ), NewTask( func() {}, ), WithStartAt( WithStartImmediately(), ), ) require.NoError(t, err) s.Start() require.NoError(t, s.Shutdown()) _, err = j.LastRun() assert.ErrorIs(t, err, ErrJobNotFound) _, err = j.NextRun() assert.ErrorIs(t, err, ErrJobNotFound) }) } func TestScheduler_NewJob(t *testing.T) { goleak.VerifyNone(t) tests := []struct { name string jd JobDefinition tsk Task opts []JobOption }{ { "cron with timezone", CronJob( "CRON_TZ=America/Chicago * * * * * *", true, ), NewTask( func() {}, ), nil, }, { "cron with timezone, no seconds", CronJob( "CRON_TZ=America/Chicago * * * * *", false, ), NewTask( func() {}, ), nil, }, { "random duration", DurationRandomJob( time.Second, time.Second*5, ), NewTask( func() {}, ), nil, }, { "daily", DailyJob( 1, NewAtTimes( NewAtTime(1, 0, 0), ), ), NewTask( func() {}, ), nil, }, { "weekly", WeeklyJob( 1, NewWeekdays(time.Monday), NewAtTimes( NewAtTime(1, 0, 0), ), ), NewTask( func() {}, ), nil, }, { "monthly", MonthlyJob( 1, NewDaysOfTheMonth(1, -1), NewAtTimes( NewAtTime(1, 0, 0), ), ), NewTask( func() {}, ), nil, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { s := newTestScheduler(t) _, err := s.NewJob(tt.jd, tt.tsk, tt.opts...) require.NoError(t, err) s.Start() require.NoError(t, s.Shutdown()) }) } } func TestScheduler_NewJobErrors(t *testing.T) { goleak.VerifyNone(t) tests := []struct { name string jd JobDefinition opts []JobOption err error }{ { "cron with timezone", CronJob( "bad cron", true, ), nil, ErrCronJobParse, }, { "random with bad min/max", DurationRandomJob( time.Second*5, time.Second, ), nil, ErrDurationRandomJobMinMax, }, { "daily job at times nil", DailyJob( 1, nil, ), nil, ErrDailyJobAtTimesNil, }, { "daily job at time nil", DailyJob( 1, NewAtTimes(nil), ), nil, ErrDailyJobAtTimeNil, }, { "daily job hours out of range", DailyJob( 1, NewAtTimes( NewAtTime(100, 0, 0), ), ), nil, ErrDailyJobHours, }, { "daily job minutes out of range", DailyJob( 1, NewAtTimes( NewAtTime(1, 100, 0), ), ), nil, ErrDailyJobMinutesSeconds, }, { "daily job seconds out of range", DailyJob( 1, NewAtTimes( NewAtTime(1, 0, 100), ), ), nil, ErrDailyJobMinutesSeconds, }, { "weekly job at times nil", WeeklyJob( 1, NewWeekdays(time.Monday), nil, ), nil, ErrWeeklyJobAtTimesNil, }, { "weekly job at time nil", WeeklyJob( 1, NewWeekdays(time.Monday), NewAtTimes(nil), ), nil, ErrWeeklyJobAtTimeNil, }, { "weekly job weekdays nil", WeeklyJob( 1, nil, NewAtTimes( NewAtTime(1, 0, 0), ), ), nil, ErrWeeklyJobDaysOfTheWeekNil, }, { "weekly job hours out of range", WeeklyJob( 1, NewWeekdays(time.Monday), NewAtTimes( NewAtTime(100, 0, 0), ), ), nil, ErrWeeklyJobHours, }, { "weekly job minutes out of range", WeeklyJob( 1, NewWeekdays(time.Monday), NewAtTimes( NewAtTime(1, 100, 0), ), ), nil, ErrWeeklyJobMinutesSeconds, }, { "weekly job seconds out of range", WeeklyJob( 1, NewWeekdays(time.Monday), NewAtTimes( NewAtTime(1, 0, 100), ), ), nil, ErrWeeklyJobMinutesSeconds, }, { "monthly job at times nil", MonthlyJob( 1, NewDaysOfTheMonth(1), nil, ), nil, ErrMonthlyJobAtTimesNil, }, { "monthly job at time nil", MonthlyJob( 1, NewDaysOfTheMonth(1), NewAtTimes(nil), ), nil, ErrMonthlyJobAtTimeNil, }, { "monthly job days out of range", MonthlyJob( 1, NewDaysOfTheMonth(0), NewAtTimes( NewAtTime(1, 0, 0), ), ), nil, ErrMonthlyJobDays, }, { "monthly job days out of range", MonthlyJob( 1, nil, NewAtTimes( NewAtTime(1, 0, 0), ), ), nil, ErrMonthlyJobDaysNil, }, { "monthly job hours out of range", MonthlyJob( 1, NewDaysOfTheMonth(1), NewAtTimes( NewAtTime(100, 0, 0), ), ), nil, ErrMonthlyJobHours, }, { "monthly job minutes out of range", MonthlyJob( 1, NewDaysOfTheMonth(1), NewAtTimes( NewAtTime(1, 100, 0), ), ), nil, ErrMonthlyJobMinutesSeconds, }, { "monthly job seconds out of range", MonthlyJob( 1, NewDaysOfTheMonth(1), NewAtTimes( NewAtTime(1, 0, 100), ), ), nil, ErrMonthlyJobMinutesSeconds, }, { "WithName no name", DurationJob( time.Second, ), []JobOption{WithName("")}, ErrWithNameEmpty, }, { "WithStartDateTime is zero", DurationJob( time.Second, ), []JobOption{WithStartAt(WithStartDateTime(time.Time{}))}, ErrWithStartDateTimePast, }, { "WithStartDateTime is in the past", DurationJob( time.Second, ), []JobOption{WithStartAt(WithStartDateTime(time.Now().Add(-time.Second)))}, ErrWithStartDateTimePast, }, { "oneTimeJob start at is zero", OneTimeJob(OneTimeJobStartDateTime(time.Time{})), nil, ErrOneTimeJobStartDateTimePast, }, { "oneTimeJob start at is in past", OneTimeJob(OneTimeJobStartDateTime(time.Now().Add(-time.Second))), nil, ErrOneTimeJobStartDateTimePast, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { s := newTestScheduler(t, WithStopTimeout(time.Millisecond*50), ) _, err := s.NewJob(tt.jd, NewTask(func() {}), tt.opts...) assert.ErrorIs(t, err, tt.err) require.NoError(t, s.Shutdown()) }) t.Run(tt.name+" global", func(t *testing.T) { s := newTestScheduler(t, WithStopTimeout(time.Millisecond*50), WithGlobalJobOptions(tt.opts...), ) _, err := s.NewJob(tt.jd, NewTask(func() {})) assert.ErrorIs(t, err, tt.err) require.NoError(t, s.Shutdown()) }) } } func TestScheduler_NewJobTask(t *testing.T) { goleak.VerifyNone(t) testFuncPtr := func() {} testFuncWithParams := func(one, two string) {} testStruct := struct{}{} tests := []struct { name string tsk Task err error }{ { "task nil", nil, ErrNewJobTaskNil, }, { "task not func - nil", NewTask(nil), ErrNewJobTaskNotFunc, }, { "task not func - string", NewTask("not a func"), ErrNewJobTaskNotFunc, }, { "task func is pointer", NewTask(&testFuncPtr), nil, }, { "parameter number does not match", NewTask(testFuncWithParams, "one"), ErrNewJobWrongNumberOfParameters, }, { "parameter type does not match", NewTask(testFuncWithParams, "one", 2), ErrNewJobWrongTypeOfParameters, }, { "parameter number does not match - ptr", NewTask(&testFuncWithParams, "one"), ErrNewJobWrongNumberOfParameters, }, { "parameter type does not match - ptr", NewTask(&testFuncWithParams, "one", 2), ErrNewJobWrongTypeOfParameters, }, { "all good struct", NewTask(func(one struct{}) {}, struct{}{}), nil, }, { "all good interface", NewTask(func(one interface{}) {}, struct{}{}), nil, }, { "all good any", NewTask(func(one any) {}, struct{}{}), nil, }, { "all good slice", NewTask(func(one []struct{}) {}, []struct{}{}), nil, }, { "all good chan", NewTask(func(one chan struct{}) {}, make(chan struct{})), nil, }, { "all good pointer", NewTask(func(one *struct{}) {}, &testStruct), nil, }, { "all good map", NewTask(func(one map[string]struct{}) {}, make(map[string]struct{})), nil, }, { "all good", NewTask(&testFuncWithParams, "one", "two"), nil, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { s := newTestScheduler(t) _, err := s.NewJob(DurationJob(time.Second), tt.tsk) assert.ErrorIs(t, err, tt.err) require.NoError(t, s.Shutdown()) }) } } func TestScheduler_WithOptionsErrors(t *testing.T) { goleak.VerifyNone(t) tests := []struct { name string opt SchedulerOption err error }{ { "WithClock nil", WithClock(nil), ErrWithClockNil, }, { "WithDistributedElector nil", WithDistributedElector(nil), ErrWithDistributedElectorNil, }, { "WithDistributedLocker nil", WithDistributedLocker(nil), ErrWithDistributedLockerNil, }, { "WithLimitConcurrentJobs limit 0", WithLimitConcurrentJobs(0, LimitModeWait), ErrWithLimitConcurrentJobsZero, }, { "WithLocation nil", WithLocation(nil), ErrWithLocationNil, }, { "WithLogger nil", WithLogger(nil), ErrWithLoggerNil, }, { "WithStopTimeout 0", WithStopTimeout(0), ErrWithStopTimeoutZeroOrNegative, }, { "WithStopTimeout -1", WithStopTimeout(-1), ErrWithStopTimeoutZeroOrNegative, }, { "WithMonitorer nil", WithMonitor(nil), ErrWithMonitorNil, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { _, err := NewScheduler(tt.opt) assert.ErrorIs(t, err, tt.err) }) } } func TestScheduler_Singleton(t *testing.T) { goleak.VerifyNone(t) tests := []struct { name string duration time.Duration limitMode LimitMode runCount int expectedMin time.Duration expectedMax time.Duration }{ { "singleton mode reschedule", time.Millisecond * 100, LimitModeReschedule, 3, time.Millisecond * 600, time.Millisecond * 1100, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { jobRanCh := make(chan struct{}, 10) s := newTestScheduler(t, WithStopTimeout(1*time.Second), WithLocation(time.Local), ) _, err := s.NewJob( DurationJob( tt.duration, ), NewTask(func() { time.Sleep(tt.duration * 2) jobRanCh <- struct{}{} }), WithSingletonMode(tt.limitMode), ) require.NoError(t, err) start := time.Now() s.Start() var runCount int for runCount < tt.runCount { select { case <-jobRanCh: runCount++ case <-time.After(time.Second): t.Fatalf("timed out waiting for jobs to run") } } stop := time.Now() require.NoError(t, s.Shutdown()) assert.GreaterOrEqual(t, stop.Sub(start), tt.expectedMin) assert.LessOrEqual(t, stop.Sub(start), tt.expectedMax) }) } } func TestScheduler_LimitMode(t *testing.T) { goleak.VerifyNone(t) tests := []struct { name string numJobs int limit uint limitMode LimitMode duration time.Duration expectedMin time.Duration expectedMax time.Duration }{ { "limit mode reschedule", 10, 2, LimitModeReschedule, time.Millisecond * 100, time.Millisecond * 400, time.Millisecond * 700, }, { "limit mode wait", 10, 2, LimitModeWait, time.Millisecond * 100, time.Millisecond * 200, time.Millisecond * 500, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { s := newTestScheduler(t, WithLimitConcurrentJobs(tt.limit, tt.limitMode), WithStopTimeout(2*time.Second), ) jobRanCh := make(chan struct{}, 20) for i := 0; i < tt.numJobs; i++ { _, err := s.NewJob( DurationJob(tt.duration), NewTask(func() { time.Sleep(tt.duration / 2) jobRanCh <- struct{}{} }), ) require.NoError(t, err) } start := time.Now() s.Start() var runCount int for runCount < tt.numJobs { select { case <-jobRanCh: runCount++ case <-time.After(time.Second): t.Fatalf("timed out waiting for jobs to run") } } stop := time.Now() require.NoError(t, s.Shutdown()) assert.GreaterOrEqual(t, stop.Sub(start), tt.expectedMin) assert.LessOrEqual(t, stop.Sub(start), tt.expectedMax) }) } } func TestScheduler_LimitModeAndSingleton(t *testing.T) { goleak.VerifyNone(t) tests := []struct { name string numJobs int limit uint limitMode LimitMode singletonMode LimitMode duration time.Duration expectedMin time.Duration expectedMax time.Duration }{ { "limit mode reschedule", 10, 2, LimitModeReschedule, LimitModeReschedule, time.Millisecond * 100, time.Millisecond * 400, time.Millisecond * 700, }, { "limit mode wait", 10, 2, LimitModeWait, LimitModeWait, time.Millisecond * 100, time.Millisecond * 200, time.Millisecond * 500, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { s := newTestScheduler(t, WithLimitConcurrentJobs(tt.limit, tt.limitMode), WithStopTimeout(2*time.Second), ) jobRanCh := make(chan int, 20) for i := 0; i < tt.numJobs; i++ { jobNum := i _, err := s.NewJob( DurationJob(tt.duration), NewTask(func() { time.Sleep(tt.duration / 2) jobRanCh <- jobNum }), WithSingletonMode(tt.singletonMode), ) require.NoError(t, err) } start := time.Now() s.Start() jobsRan := make(map[int]int) var runCount int for runCount < tt.numJobs { select { case jobNum := <-jobRanCh: runCount++ jobsRan[jobNum]++ case <-time.After(time.Second): t.Fatalf("timed out waiting for jobs to run") } } stop := time.Now() require.NoError(t, s.Shutdown()) assert.GreaterOrEqual(t, stop.Sub(start), tt.expectedMin) assert.LessOrEqual(t, stop.Sub(start), tt.expectedMax) for _, count := range jobsRan { if tt.singletonMode == LimitModeWait { assert.Equal(t, 1, count) } else { assert.LessOrEqual(t, count, 5) } } }) } } var _ Elector = (*testElector)(nil) type testElector struct { mu sync.Mutex leaderElected bool notLeader chan struct{} } func (t *testElector) IsLeader(ctx context.Context) error { select { case <-ctx.Done(): return fmt.Errorf("done") default: } t.mu.Lock() defer t.mu.Unlock() if t.leaderElected { t.notLeader <- struct{}{} return fmt.Errorf("already elected leader") } t.leaderElected = true return nil } var _ Locker = (*testLocker)(nil) type testLocker struct { mu sync.Mutex jobLocked bool notLocked chan struct{} } func (t *testLocker) Lock(_ context.Context, _ string) (Lock, error) { t.mu.Lock() defer t.mu.Unlock() if t.jobLocked { t.notLocked <- struct{}{} return nil, fmt.Errorf("job already locked") } t.jobLocked = true return &testLock{}, nil } var _ Lock = (*testLock)(nil) type testLock struct{} func (t testLock) Unlock(_ context.Context) error { return nil } func TestScheduler_WithDistributed(t *testing.T) { notLocked := make(chan struct{}, 10) notLeader := make(chan struct{}, 10) goleak.VerifyNone(t) tests := []struct { name string count int opt SchedulerOption assertions func(*testing.T) }{ { "3 schedulers with elector", 3, WithDistributedElector(&testElector{ notLeader: notLeader, }), func(t *testing.T) { timeout := time.Now().Add(1 * time.Second) var notLeaderCount int for { if time.Now().After(timeout) { break } select { case <-notLeader: notLeaderCount++ default: } } assert.Equal(t, 2, notLeaderCount) }, }, { "3 schedulers with locker", 3, WithDistributedLocker(&testLocker{ notLocked: notLocked, }), func(t *testing.T) { timeout := time.Now().Add(1 * time.Second) var notLockedCount int for { if time.Now().After(timeout) { break } select { case <-notLocked: notLockedCount++ default: } } }, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { jobsRan := make(chan struct{}, 20) ctx, cancel := context.WithCancel(context.Background()) schedulersDone := make(chan struct{}, tt.count) for i := tt.count; i > 0; i-- { s := newTestScheduler(t, tt.opt, ) go func() { s.Start() _, err := s.NewJob( DurationJob( time.Second, ), NewTask( func() { time.Sleep(100 * time.Millisecond) jobsRan <- struct{}{} }, ), WithStartAt( WithStartImmediately(), ), ) require.NoError(t, err) <-ctx.Done() err = s.Shutdown() require.NoError(t, err) schedulersDone <- struct{}{} }() } var runCount int select { case <-jobsRan: cancel() runCount++ case <-time.After(time.Second): cancel() t.Error("timed out waiting for job to run") } var doneCount int timeout := time.Now().Add(3 * time.Second) for doneCount < tt.count && time.Now().After(timeout) { select { case <-schedulersDone: doneCount++ default: } } close(jobsRan) for range jobsRan { runCount++ } assert.Equal(t, 1, runCount) time.Sleep(time.Second) tt.assertions(t) }) } } func TestScheduler_RemoveJob(t *testing.T) { goleak.VerifyNone(t) tests := []struct { name string addJob bool err error }{ { "success", true, nil, }, { "job not found", false, ErrJobNotFound, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { s := newTestScheduler(t) var id uuid.UUID if tt.addJob { j, err := s.NewJob(DurationJob(time.Second), NewTask(func() {})) require.NoError(t, err) id = j.ID() } else { id = uuid.New() } err := s.RemoveJob(id) assert.ErrorIs(t, err, err) require.NoError(t, s.Shutdown()) }) } } func TestScheduler_WithEventListeners(t *testing.T) { goleak.VerifyNone(t) listenerRunCh := make(chan error, 1) testErr := fmt.Errorf("test error") tests := []struct { name string tsk Task el EventListener expectRun bool expectErr error }{ { "AfterJobRuns", NewTask(func() {}), AfterJobRuns(func(_ uuid.UUID, _ string) { listenerRunCh <- nil }), true, nil, }, { "AfterJobRunsWithError - error", NewTask(func() error { return testErr }), AfterJobRunsWithError(func(_ uuid.UUID, _ string, err error) { listenerRunCh <- err }), true, testErr, }, { "AfterJobRunsWithError - multiple return values, including error", NewTask(func() (bool, error) { return false, testErr }), AfterJobRunsWithError(func(_ uuid.UUID, _ string, err error) { listenerRunCh <- err }), true, testErr, }, { "AfterJobRunsWithError - no error", NewTask(func() error { return nil }), AfterJobRunsWithError(func(_ uuid.UUID, _ string, err error) { listenerRunCh <- err }), false, nil, }, { "BeforeJobRuns", NewTask(func() {}), BeforeJobRuns(func(_ uuid.UUID, _ string) { listenerRunCh <- nil }), true, nil, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { s := newTestScheduler(t) _, err := s.NewJob( DurationJob(time.Minute*10), tt.tsk, WithStartAt( WithStartImmediately(), ), WithEventListeners(tt.el), WithLimitedRuns(1), ) require.NoError(t, err) s.Start() if tt.expectRun { select { case err = <-listenerRunCh: assert.ErrorIs(t, err, tt.expectErr) case <-time.After(time.Second): t.Fatal("timed out waiting for listener to run") } } else { select { case <-listenerRunCh: t.Fatal("listener ran when it shouldn't have") case <-time.After(time.Millisecond * 100): } } require.NoError(t, s.Shutdown()) }) } } func TestScheduler_ManyJobs(t *testing.T) { s := newTestScheduler(t) jobsRan := make(chan struct{}, 20000) for i := 1; i <= 1000; i++ { _, err := s.NewJob( DurationJob( time.Millisecond*100, ), NewTask( func() { jobsRan <- struct{}{} }, ), WithStartAt(WithStartImmediately()), ) require.NoError(t, err) } s.Start() time.Sleep(1 * time.Second) require.NoError(t, s.Shutdown()) close(jobsRan) var count int for range jobsRan { count++ } assert.GreaterOrEqual(t, count, 9900) assert.LessOrEqual(t, count, 11000) } func TestScheduler_RunJobNow(t *testing.T) { chDuration := make(chan struct{}, 10) chMonthly := make(chan struct{}, 10) chDurationImmediate := make(chan struct{}, 10) chDurationSingleton := make(chan struct{}, 10) chOneTime := make(chan struct{}, 10) tests := []struct { name string ch chan struct{} j JobDefinition fun any opts []JobOption expectedDiff func() time.Duration expectedRuns int }{ { "duration job", chDuration, DurationJob(time.Second * 10), func() { chDuration <- struct{}{} }, nil, func() time.Duration { return 0 }, 1, }, { "monthly job", chMonthly, MonthlyJob(1, NewDaysOfTheMonth(1), NewAtTimes(NewAtTime(0, 0, 0))), func() { chMonthly <- struct{}{} }, nil, func() time.Duration { return 0 }, 1, }, { "duration job - start immediately", chDurationImmediate, DurationJob(time.Second * 5), func() { chDurationImmediate <- struct{}{} }, []JobOption{ WithStartAt( WithStartImmediately(), ), }, func() time.Duration { return 5 * time.Second }, 2, }, { "duration job - singleton", chDurationSingleton, DurationJob(time.Second * 10), func() { chDurationSingleton <- struct{}{} time.Sleep(200 * time.Millisecond) }, []JobOption{ WithStartAt( WithStartImmediately(), ), WithSingletonMode(LimitModeReschedule), }, func() time.Duration { return 20 * time.Second }, 1, }, { "one time job", chOneTime, OneTimeJob(OneTimeJobStartImmediately()), func() { chOneTime <- struct{}{} }, nil, nil, 2, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { s := newTestScheduler(t) _, err := s.NewJob(tt.j, NewTask(tt.fun), tt.opts...) require.NoError(t, err) j := s.Jobs()[0] s.Start() var nextRunBefore time.Time if tt.expectedDiff != nil { for ; nextRunBefore.IsZero() || err != nil; nextRunBefore, err = j.NextRun() { //nolint:revive } } assert.NoError(t, err) time.Sleep(100 * time.Millisecond) require.NoError(t, j.RunNow()) var runCount int select { case <-tt.ch: runCount++ case <-time.After(time.Second): t.Fatal("timed out waiting for job to run") } timeout := time.Now().Add(time.Second) for time.Now().Before(timeout) { select { case <-tt.ch: runCount++ default: } } assert.Equal(t, tt.expectedRuns, runCount) nextRunAfter, err := j.NextRun() if tt.expectedDiff != nil && tt.expectedDiff() > 0 { for ; nextRunBefore.IsZero() || nextRunAfter.Equal(nextRunBefore); nextRunAfter, err = j.NextRun() { //nolint:revive time.Sleep(100 * time.Millisecond) } } assert.NoError(t, err) assert.NoError(t, s.Shutdown()) if tt.expectedDiff != nil { assert.Equal(t, tt.expectedDiff(), nextRunAfter.Sub(nextRunBefore)) } }) } } func TestScheduler_OneTimeJob(t *testing.T) { tests := []struct { name string startAt func() OneTimeJobStartAtOption }{ { "start now", func() OneTimeJobStartAtOption { return OneTimeJobStartImmediately() }, }, { "start in 100 ms", func() OneTimeJobStartAtOption { return OneTimeJobStartDateTime(time.Now().Add(100 * time.Millisecond)) }, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { jobRan := make(chan struct{}, 2) s := newTestScheduler(t) j, err := s.NewJob( OneTimeJob(tt.startAt()), NewTask(func() { jobRan <- struct{}{} }), ) require.NoError(t, err) s.Start() select { case <-jobRan: case <-time.After(500 * time.Millisecond): t.Fatal("timed out waiting for job to run") } var nextRun time.Time for ; nextRun.IsZero(); nextRun, err = j.NextRun() { //nolint:revive } assert.NoError(t, err) assert.True(t, nextRun.Before(time.Now())) assert.NoError(t, s.Shutdown()) }) } } func TestScheduler_Jobs(t *testing.T) { tests := []struct { name string }{ { "order is equal", }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { s := newTestScheduler(t) for i := 0; i <= 20; i++ { _, err := s.NewJob( DurationJob(time.Second), NewTask(func() {}), ) require.NoError(t, err) } jobsFirst := s.Jobs() jobsSecond := s.Jobs() assert.Equal(t, jobsFirst, jobsSecond) assert.NoError(t, s.Shutdown()) }) } } type testMonitor struct { mu sync.Mutex counter map[string]int time map[string][]time.Duration } func newTestMonitor() *testMonitor { return &testMonitor{ counter: make(map[string]int), time: make(map[string][]time.Duration), } } func (t *testMonitor) IncrementJob(_ uuid.UUID, name string, _ []string, _ JobStatus) { t.mu.Lock() defer t.mu.Unlock() _, ok := t.counter[name] if !ok { t.counter[name] = 0 } t.counter[name]++ } func (t *testMonitor) RecordJobTiming(startTime, endTime time.Time, _ uuid.UUID, name string, _ []string) { t.mu.Lock() defer t.mu.Unlock() _, ok := t.time[name] if !ok { t.time[name] = make([]time.Duration, 0) } t.time[name] = append(t.time[name], endTime.Sub(startTime)) } func TestScheduler_WithMonitor(t *testing.T) { goleak.VerifyNone(t) tests := []struct { name string jd JobDefinition jobName string }{ { "scheduler with monitorer", DurationJob(time.Millisecond * 50), "job", }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { ch := make(chan struct{}, 20) monitor := newTestMonitor() s := newTestScheduler(t, WithMonitor(monitor)) opt := []JobOption{ WithName(tt.jobName), WithStartAt( WithStartImmediately(), ), } _, err := s.NewJob( tt.jd, NewTask(func() { ch <- struct{}{} }), opt..., ) require.NoError(t, err) s.Start() time.Sleep(150 * time.Millisecond) require.NoError(t, s.Shutdown()) close(ch) expectedCount := 0 for range ch { expectedCount++ } got := monitor.counter[tt.jobName] if got != expectedCount { t.Fatalf("job %q counter expected %d, got %d", tt.jobName, expectedCount, got) } }) } }