Repository: furiko-io/furiko
Pull request #39 (2022-04-10 17:29:09)

test(cron): Add unit tests for CronWorker
Adds unit tests for CronWorker.

codecov[bot]: # [Codecov](https://codecov.io/gh/furiko-io/furiko/pull/39?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=furiko-io) Report
> Merging [#39](https://codecov.io/gh/furiko-io/furiko/pull/39?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=furiko-io) (4f15320) into [main](https://codecov.io/gh/furiko-io/furiko/commit/7b79cd4db24f2fc81956d81fa04ac44898d75593?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=furiko-io) (7b79cd4) will **increase** coverage by `2.29%`.
> The diff coverage is `85.48%`.

```diff
@@            Coverage Diff             @@
##             main      #39      +/-   ##
==========================================
+ Coverage   49.73%   52.02%   +2.29%     
==========================================
  Files         161      163       +2     
  Lines        7991     8036      +45     
==========================================
+ Hits         3974     4181     +207     
+ Misses       3834     3657     -177     
- Partials      183      198      +15     
```


| [Impacted Files](https://codecov.io/gh/furiko-io/furiko/pull/39?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=furiko-io) | Coverage Δ | |
|---|---|---|
| [...execution/controllers/croncontroller/controller.go](https://codecov.io/gh/furiko-io/furiko/pull/39/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=furiko-io#diff-cGtnL2V4ZWN1dGlvbi9jb250cm9sbGVycy9jcm9uY29udHJvbGxlci9jb250cm9sbGVyLmdv) | `23.68% <16.66%> (-0.65%)` | :arrow_down: |
| [pkg/utils/testutils/klog.go](https://codecov.io/gh/furiko-io/furiko/pull/39/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=furiko-io#diff-cGtnL3V0aWxzL3Rlc3R1dGlscy9rbG9nLmdv) | `60.00% <60.00%> (ø)` | |
| [pkg/utils/testutils/workqueue.go](https://codecov.io/gh/furiko-io/furiko/pull/39/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=furiko-io#diff-cGtnL3V0aWxzL3Rlc3R1dGlscy93b3JrcXVldWUuZ28=) | `90.90% <90.90%> (ø)` | |
| [...xecution/controllers/croncontroller/cron\_worker.go](https://codecov.io/gh/furiko-io/furiko/pull/39/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=furiko-io#diff-cGtnL2V4ZWN1dGlvbi9jb250cm9sbGVycy9jcm9uY29udHJvbGxlci9jcm9uX3dvcmtlci5nbw==) | `75.00% <100.00%> (ø)` | |
| [...g/execution/controllers/croncontroller/informer.go](https://codecov.io/gh/furiko-io/furiko/pull/39/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=furiko-io#diff-cGtnL2V4ZWN1dGlvbi9jb250cm9sbGVycy9jcm9uY29udHJvbGxlci9pbmZvcm1lci5nbw==) | `68.08% <100.00%> (+68.08%)` | :arrow_up: |
| [pkg/execution/controllers/croncontroller/ticker.go](https://codecov.io/gh/furiko-io/furiko/pull/39/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=furiko-io#diff-cGtnL2V4ZWN1dGlvbi9jb250cm9sbGVycy9jcm9uY29udHJvbGxlci90aWNrZXIuZ28=) | `100.00% <100.00%> (ø)` | |
| [.../runtime/controllercontext/mock/context\_configs.go](https://codecov.io/gh/furiko-io/furiko/pull/39/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=furiko-io#diff-cGtnL3J1bnRpbWUvY29udHJvbGxlcmNvbnRleHQvbW9jay9jb250ZXh0X2NvbmZpZ3MuZ28=) | `81.39% <100.00%> (+1.90%)` | :arrow_up: |
| [pkg/execution/controllers/croncontroller/util.go](https://codecov.io/gh/furiko-io/furiko/pull/39/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=furiko-io#diff-cGtnL2V4ZWN1dGlvbi9jb250cm9sbGVycy9jcm9uY29udHJvbGxlci91dGlsLmdv) | `90.90% <0.00%> (+9.09%)` | :arrow_up: |
| [...ed/typed/execution/v1alpha1/fake/fake\_jobconfig.go](https://codecov.io/gh/furiko-io/furiko/pull/39/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=furiko-io#diff-cGtnL2dlbmVyYXRlZC9jbGllbnRzZXQvdmVyc2lvbmVkL3R5cGVkL2V4ZWN1dGlvbi92MWFscGhhMS9mYWtlL2Zha2Vfam9iY29uZmlnLmdv) | `42.25% <0.00%> (+14.08%)` | :arrow_up: |
| ... and [6 more](https://codecov.io/gh/furiko-io/furiko/pull/39/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=furiko-io) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/furiko-io/furiko/pull/39?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=furiko-io).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=furiko-io)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/furiko-io/furiko/pull/39?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=furiko-io). Last update [7b79cd4...4f15320](https://codecov.io/gh/furiko-io/furiko/pull/39?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=furiko-io). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=furiko-io).


Patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 171523c..b99fe50 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -66,9 +66,6 @@ jobs:
       - run: make lint-license
 
   go-test:
-    needs:
-      - golangci-lint
-      - generate-lint
     strategy:
       matrix:
         go-version: [ 1.17 ]
diff --git a/pkg/execution/controllers/croncontroller/controller.go b/pkg/execution/controllers/croncontroller/controller.go
index 2fc4083..3e38db5 100644
--- a/pkg/execution/controllers/croncontroller/controller.go
+++ b/pkg/execution/controllers/croncontroller/controller.go
@@ -109,8 +109,8 @@ func NewController(
 		terminate: cancel,
 	}
 
-	ctrl.cronWorker = NewCronWorker(ctrl.Context)
-	ctrl.informerWorker = NewInformerWorker(ctrl.Context)
+	ctrl.cronWorker = NewCronWorker(ctrl.Context, newEnqueueHandler(ctrl.Context))
+	ctrl.informerWorker = NewInformerWorker(ctrl.Context, NewUpdateHandler(ctrl.Context))
 
 	client := NewExecutionControl(
 		(&Reconciler{}).Name(),
@@ -131,13 +131,15 @@ func (c *Controller) Run(ctx context.Context) error {
 	defer utilruntime.HandleCrash()
 	klog.InfoS("croncontroller: starting controller")
 
+	c.informerWorker.Init()
+
 	if ok := cache.WaitForNamedCacheSync(controllerName, ctx.Done(), c.HasSynced...); !ok {
 		klog.Error("croncontroller: cache sync timeout")
 		return controllerutil.ErrWaitForCacheSyncTimeout
 	}
 
 	c.reconciler.Start(c.ctx)
-	c.cronWorker.Start(c.ctx.Done())
+	c.cronWorker.Start(c.ctx)
 
 	atomic.StoreUint64(&c.healthStatus, 1)
 	klog.InfoS("croncontroller: started controller")
diff --git a/pkg/execution/controllers/croncontroller/cron.go b/pkg/execution/controllers/croncontroller/cron_worker.go
similarity index 81%
rename from pkg/execution/controllers/croncontroller/cron.go
rename to pkg/execution/controllers/croncontroller/cron_worker.go
index d26cb9e..678c09e 100644
--- a/pkg/execution/controllers/croncontroller/cron.go
+++ b/pkg/execution/controllers/croncontroller/cron_worker.go
@@ -17,7 +17,9 @@
 package croncontroller
 
 import (
+	"context"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/pkg/errors"
@@ -45,13 +47,21 @@ const (
 // to be created.
 type CronWorker struct {
 	*Context
-	*Schedule
+	schedule *Schedule
+	handler  EnqueueHandler
+	mu       sync.Mutex
 }
 
-func NewCronWorker(ctrlContext *Context) *CronWorker {
+// EnqueueHandler knows how to enqueue a JobConfig to be created.
+type EnqueueHandler interface {
+	EnqueueJobConfig(jobConfig *execution.JobConfig, scheduleTime time.Time) error
+}
+
+func NewCronWorker(ctrlContext *Context, handler EnqueueHandler) *CronWorker {
 	return &CronWorker{
 		Context:  ctrlContext,
-		Schedule: NewSchedule(ctrlContext),
+		handler:  handler,
+		schedule: NewSchedule(ctrlContext),
 	}
 }
 
@@ -59,12 +69,15 @@ func (w *CronWorker) WorkerName() string {
 	return fmt.Sprintf("%v.CronWorker", controllerName)
 }
 
-func (w *CronWorker) Start(stopCh <-chan struct{}) {
-	go CronTimerUntil(instrumentWorkerMetrics(w.WorkerName(), w.work), CronWorkerInterval, stopCh)
+func (w *CronWorker) Start(ctx context.Context) {
+	go ClockTickUntil(instrumentWorkerMetrics(w.WorkerName(), w.Work), CronWorkerInterval, ctx.Done())
 }
 
-// work runs a single iteration of synchronizing JobConfigs.
-func (w *CronWorker) work() {
+// Work runs a single iteration of synchronizing all JobConfigs.
+func (w *CronWorker) Work() {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+
 	trace := utiltrace.New(
 		"cron_schedule_all",
 	)
@@ -123,7 +136,7 @@ func (w *CronWorker) flushKeys() {
 	for flushes < 1000 {
 		select {
 		case jobConfig := <-w.updatedConfigs:
-			w.FlushNextScheduleTime(jobConfig)
+			w.schedule.FlushNextScheduleTime(jobConfig)
 			flushes++
 		default:
 			// Nothing more to flush.
@@ -169,7 +182,14 @@ func (w *CronWorker) syncOne(
 	// Get next scheduled time repeatedly and enqueue for each scheduled time.
 	// This helps to prevent missed executions (maybe due to worker stuck).
 	for i := 0; i < maxMissedSchedules; i++ {
-		next := w.GetNextScheduleTime(jobConfig, now, expr)
+		next := w.schedule.GetNextScheduleTime(jobConfig, now, expr)
+
+		klog.V(6).InfoS("croncontroller: get next schedule for job config",
+			"namespace", jobConfig.Namespace,
+			"name", jobConfig.Name,
+			"now", now,
+			"next", next,
+		)
 
 		// There is no next schedule time.
 		if next.IsZero() {
@@ -182,12 +202,12 @@ func (w *CronWorker) syncOne(
 		}
 
 		// Enqueue the job.
-		if err := w.enqueueJob(jobConfig, next); err != nil {
+		if err := w.handler.EnqueueJobConfig(jobConfig, next); err != nil {
 			return errors.Wrapf(err, "cannot enqueue job for %v", next)
 		}
 
 		// Bump the next schedule time for the job config.
-		w.BumpNextScheduleTime(jobConfig, next, expr)
+		w.schedule.BumpNextScheduleTime(jobConfig, next, expr)
 
 		klog.V(2).InfoS("croncontroller: scheduled job by cron",
 			"worker", w.WorkerName(),
@@ -207,21 +227,7 @@ func (w *CronWorker) syncOne(
 		"namespace", jobConfig.GetNamespace(),
 		"name", jobConfig.GetName(),
 	)
-	w.BumpNextScheduleTime(jobConfig, now, expr)
-
-	return nil
-}
-
-// enqueueJob enqueues a single job to be scheduled.
-func (w *CronWorker) enqueueJob(jobConfig *execution.JobConfig, scheduleTime time.Time) error {
-	// Use our custom KeyFunc.
-	key, err := JobConfigKeyFunc(jobConfig, scheduleTime)
-	if err != nil {
-		return errors.Wrapf(err, "keyfunc error")
-	}
-
-	// Add the (JobConfig, ScheduleTime) to the workqueue.
-	w.queue.Add(key)
+	w.schedule.BumpNextScheduleTime(jobConfig, now, expr)
 
 	return nil
 }
@@ -241,3 +247,28 @@ func (w *CronWorker) getTimezone(cronSchedule *execution.CronSchedule, cfg *conf
 	// Fallback to controller default.
 	return defaultTimezone
 }
+
+type enqueueHandler struct {
+	*Context
+}
+
+var _ EnqueueHandler = (*enqueueHandler)(nil)
+
+func newEnqueueHandler(ctrlContext *Context) *enqueueHandler {
+	return &enqueueHandler{
+		Context: ctrlContext,
+	}
+}
+
+func (h *enqueueHandler) EnqueueJobConfig(jobConfig *execution.JobConfig, scheduleTime time.Time) error {
+	// Use our custom KeyFunc.
+	key, err := JobConfigKeyFunc(jobConfig, scheduleTime)
+	if err != nil {
+		return errors.Wrapf(err, "keyfunc error")
+	}
+
+	// Add the (JobConfig, ScheduleTime) to the workqueue.
+	h.queue.Add(key)
+
+	return nil
+}
diff --git a/pkg/execution/controllers/croncontroller/informer.go b/pkg/execution/controllers/croncontroller/informer.go
index 60a2aa1..9365ebb 100644
--- a/pkg/execution/controllers/croncontroller/informer.go
+++ b/pkg/execution/controllers/croncontroller/informer.go
@@ -23,6 +23,7 @@ import (
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/klog/v2"
 
+	execution "github.com/furiko-io/furiko/apis/execution/v1alpha1"
 	"github.com/furiko-io/furiko/pkg/utils/eventhandler"
 )
 
@@ -30,13 +31,28 @@ import (
 // for the controller.
 type InformerWorker struct {
 	*Context
+	handler UpdateHandler
 }
 
-func NewInformerWorker(ctrlContext *Context) *InformerWorker {
+// UpdateHandler knows how to handle updates for a JobConfig.
+type UpdateHandler interface {
+	OnUpdate(*execution.JobConfig)
+}
+
+func NewInformerWorker(ctrlContext *Context, handler UpdateHandler) *InformerWorker {
 	w := &InformerWorker{
 		Context: ctrlContext,
+		handler: handler,
 	}
 
+	return w
+}
+
+func (w *InformerWorker) WorkerName() string {
+	return fmt.Sprintf("%v.Informer", controllerName)
+}
+
+func (w *InformerWorker) Init() {
 	// Add event handler when we get JobConfig updates.
 	w.jobconfigInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
 		UpdateFunc: func(oldObj, newObj interface{}) {
@@ -44,12 +60,6 @@ func NewInformerWorker(ctrlContext *Context) *InformerWorker {
 		},
 		DeleteFunc: w.enqueueFlush,
 	})
-
-	return w
-}
-
-func (w *InformerWorker) WorkerName() string {
-	return fmt.Sprintf("%v.Informer", controllerName)
 }
 
 func (w *InformerWorker) handleUpdate(oldObj, newObj interface{}) {
@@ -89,6 +99,20 @@ func (w *InformerWorker) enqueueFlush(obj interface{}) {
 			"name", rjc.GetName(),
 			"schedule", spew.Sdump(rjc.Spec.Schedule),
 		)
-		w.updatedConfigs <- rjc
+		w.handler.OnUpdate(rjc)
 	}
 }
+
+type updateHandler struct {
+	updateChan chan *execution.JobConfig
+}
+
+var _ UpdateHandler = (*updateHandler)(nil)
+
+func NewUpdateHandler(ctrlContext *Context) UpdateHandler {
+	return &updateHandler{updateChan: ctrlContext.updatedConfigs}
+}
+
+func (d *updateHandler) OnUpdate(jobConfig *execution.JobConfig) {
+	d.updateChan <- jobConfig
+}
diff --git a/pkg/execution/controllers/croncontroller/cron_timer.go b/pkg/execution/controllers/croncontroller/ticker.go
similarity index 85%
rename from pkg/execution/controllers/croncontroller/cron_timer.go
rename to pkg/execution/controllers/croncontroller/ticker.go
index f6af87c..de098a0 100644
--- a/pkg/execution/controllers/croncontroller/cron_timer.go
+++ b/pkg/execution/controllers/croncontroller/ticker.go
@@ -19,13 +19,15 @@ package croncontroller
 import (
 	"math"
 	"time"
+
+	"k8s.io/klog/v2"
 )
 
-// CronTimerUntil runs work every timestep until stopCh is signaled.
+// ClockTickUntil runs work every timestep until stopCh is signaled.
 // The work function will first be called immediately on initial call, followed by
 // rounding up to the nearest timestep for every subsequent step.
 // The smallest timestep that is admissible is a second.
-func CronTimerUntil(work func(), timestep time.Duration, stopCh <-chan struct{}) {
+func ClockTickUntil(work func(), timestep time.Duration, stopCh <-chan struct{}) {
 	step := timestep.Seconds()
 	var nextInterval time.Duration
 
@@ -36,6 +38,7 @@ func CronTimerUntil(work func(), timestep time.Duration, stopCh <-chan struct{})
 		case <-Clock.After(nextInterval):
 		}
 
+		klog.V(6).InfoS("croncontroller: clock tick", "now", Clock.Now())
 		work()
 
 		// Reset timer to start of next timestep.
diff --git a/pkg/runtime/controllercontext/mock/context_configs.go b/pkg/runtime/controllercontext/mock/context_configs.go
index 7500557..42e9a56 100644
--- a/pkg/runtime/controllercontext/mock/context_configs.go
+++ b/pkg/runtime/controllercontext/mock/context_configs.go
@@ -33,7 +33,7 @@ type Configs struct {
 	configLoader *ConfigLoader
 }
 
-// NewConfigs returns a new dynamic Configs manager that supports overriding a
+// NewConfigs returns a new dynamic config manager that supports overriding a
 // defaults configloader with configs from memory.
 func NewConfigs() *Configs {
 	mgr := configloader.NewConfigManager()
@@ -48,6 +48,13 @@ func NewConfigs() *Configs {
 	}
 }
 
+// SetConfigs is a convenience method to set configs from a map.
+func (c *Configs) SetConfigs(cfgs map[configv1alpha1.ConfigName]runtime.Object) {
+	for configName, cfg := range cfgs {
+		c.MockConfigLoader().SetConfig(configName, cfg)
+	}
+}
+
 // MockConfigLoader returns the mock configloader.Loader.
 func (c *Configs) MockConfigLoader() *ConfigLoader {
 	return c.configLoader


Test patch:
diff --git a/pkg/execution/controllers/croncontroller/cron_worker_test.go b/pkg/execution/controllers/croncontroller/cron_worker_test.go
new file mode 100644
index 0000000..c477735
--- /dev/null
+++ b/pkg/execution/controllers/croncontroller/cron_worker_test.go
@@ -0,0 +1,515 @@
+/*
+ * Copyright 2022 The Furiko Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package croncontroller_test
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/clock"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/klog/v2"
+	"k8s.io/utils/pointer"
+
+	configv1alpha1 "github.com/furiko-io/furiko/apis/config/v1alpha1"
+	execution "github.com/furiko-io/furiko/apis/execution/v1alpha1"
+	"github.com/furiko-io/furiko/pkg/execution/controllers/croncontroller"
+	"github.com/furiko-io/furiko/pkg/runtime/controllercontext/mock"
+	"github.com/furiko-io/furiko/pkg/utils/testutils"
+)
+
+var (
+	cronWorkerJobConfig = &execution.JobConfig{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "job-config-test",
+		},
+		Spec: execution.JobConfigSpec{
+			Schedule: &execution.ScheduleSpec{
+				Cron: &execution.CronSchedule{
+					Expression: "* * * * *",
+				},
+			},
+		},
+	}
+
+	cronWorkerJobConfigUpdateToEvery15Sec = &execution.JobConfig{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "job-config-test",
+		},
+		Spec: execution.JobConfigSpec{
+			Schedule: &execution.ScheduleSpec{
+				Cron: &execution.CronSchedule{
+					Expression: "0/15 * * * * * *",
+				},
+			},
+		},
+	}
+
+	cronWorkerJobConfigDaily = &execution.JobConfig{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "job-config-daily",
+		},
+		Spec: execution.JobConfigSpec{
+			Schedule: &execution.ScheduleSpec{
+				Cron: &execution.CronSchedule{
+					Expression: "0 10 * * *",
+				},
+			},
+		},
+	}
+
+	cronWorkerJobConfigDailySingapore = &execution.JobConfig{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "job-config-daily-singapore",
+		},
+		Spec: execution.JobConfigSpec{
+			Schedule: &execution.ScheduleSpec{
+				Cron: &execution.CronSchedule{
+					Expression: "0 10 * * *",
+					Timezone:   "Asia/Singapore",
+				},
+			},
+		},
+	}
+)
+
+func TestCronWorker(t *testing.T) { // nolint:gocognit
+	type step struct {
+		Name        string
+		Time        time.Time
+		Update      *execution.JobConfig
+		Delete      *execution.JobConfig
+		WantEnqueue []string
+	}
+	tests := []struct {
+		name       string
+		jobConfigs []*execution.JobConfig
+		configs    map[configv1alpha1.ConfigName]runtime.Object
+		log        klog.Level
+		steps      []step
+	}{
+		{
+			name: "Scheduled every minute",
+			jobConfigs: []*execution.JobConfig{
+				cronWorkerJobConfig,
+			},
+			steps: []step{
+				{
+					Name: "Initial time",
+					Time: testutils.Mktime("2022-04-01T10:52:04Z"),
+				},
+				{
+					Name: "No enqueue until end of minute",
+					Time: testutils.Mktime("2022-04-01T10:52:59Z"),
+				},
+				{
+					Name: "Want enqueue at start of minute",
+					Time: testutils.Mktime("2022-04-01T10:53:00Z"),
+					WantEnqueue: []string{
+						keyFunc(cronWorkerJobConfig, testutils.Mktime("2022-04-01T10:53:00Z")),
+					},
+				},
+				{
+					Name: "No more enqueue at next second",
+					Time: testutils.Mktime("2022-04-01T10:53:01Z"),
+				},
+				{
+					Name: "Want enqueue at start of next minute",
+					Time: testutils.Mktime("2022-04-01T10:54:00Z"),
+					WantEnqueue: []string{
+						keyFunc(cronWorkerJobConfig, testutils.Mktime("2022-04-01T10:54:00Z")),
+					},
+				},
+				{
+					Name: "Multiple enqueue when jumping a few minutes",
+					Time: testutils.Mktime("2022-04-01T10:57:00Z"),
+					WantEnqueue: []string{
+						keyFunc(cronWorkerJobConfig, testutils.Mktime("2022-04-01T10:55:00Z")),
+						keyFunc(cronWorkerJobConfig, testutils.Mktime("2022-04-01T10:56:00Z")),
+						keyFunc(cronWorkerJobConfig, testutils.Mktime("2022-04-01T10:57:00Z")),
+					},
+				},
+			},
+		},
+		{
+			name: "Update JobConfig",
+			jobConfigs: []*execution.JobConfig{
+				cronWorkerJobConfig,
+			},
+			steps: []step{
+				{
+					Name: "Initial time",
+					Time: testutils.Mktime("2022-04-01T10:52:04Z"),
+				},
+				{
+					Name: "Enqueue at next minute",
+					Time: testutils.Mktime("2022-04-01T10:53:00Z"),
+					WantEnqueue: []string{
+						keyFunc(cronWorkerJobConfig, testutils.Mktime("2022-04-01T10:53:00Z")),
+					},
+				},
+				{
+					Name: "Update time",
+					Time: testutils.Mktime("2022-04-01T10:53:10Z"),
+				},
+				{
+					Name:   "Update JobConfig",
+					Update: cronWorkerJobConfigUpdateToEvery15Sec,
+				},
+				{
+					Name: "No enqueue yet",
+					Time: testutils.Mktime("2022-04-01T10:53:11Z"),
+				},
+				{
+					Name: "Want enqueue at 15 sec mark",
+					Time: testutils.Mktime("2022-04-01T10:53:15Z"),
+					WantEnqueue: []string{
+						keyFunc(cronWorkerJobConfig, testutils.Mktime("2022-04-01T10:53:15Z")),
+					},
+				},
+			},
+		},
+		{
+			name: "Delete JobConfig",
+			jobConfigs: []*execution.JobConfig{
+				cronWorkerJobConfig,
+			},
+			steps: []step{
+				{
+					Name: "Initial time",
+					Time: testutils.Mktime("2022-04-01T10:52:04Z"),
+				},
+				{
+					Name: "Enqueue at next minute",
+					Time: testutils.Mktime("2022-04-01T10:53:00Z"),
+					WantEnqueue: []string{
+						keyFunc(cronWorkerJobConfig, testutils.Mktime("2022-04-01T10:53:00Z")),
+					},
+				},
+				{
+					Name:   "Delete JobConfig",
+					Delete: cronWorkerJobConfig,
+				},
+				{
+					Name: "No enqueue 1 minute later",
+					Time: testutils.Mktime("2022-04-01T10:54:00Z"),
+				},
+				{
+					Name: "No enqueue 2 minutes later",
+					Time: testutils.Mktime("2022-04-01T10:55:00Z"),
+				},
+			},
+		},
+		{
+			name: "Scheduled daily",
+			jobConfigs: []*execution.JobConfig{
+				cronWorkerJobConfigDaily,
+			},
+			steps: []step{
+				{
+					Name: "Initial time",
+					Time: testutils.Mktime("2022-04-01T10:52:04Z"),
+				},
+				{
+					Name: "Enqueue next day",
+					Time: testutils.Mktime("2022-04-02T10:00:00Z"),
+					WantEnqueue: []string{
+						keyFunc(cronWorkerJobConfigDaily, testutils.Mktime("2022-04-02T10:00:00Z")),
+					},
+				},
+			},
+		},
+		{
+			name: "Scheduled daily with default configured timezone",
+			jobConfigs: []*execution.JobConfig{
+				cronWorkerJobConfigDaily,
+			},
+			configs: map[configv1alpha1.ConfigName]runtime.Object{
+				configv1alpha1.CronExecutionConfigName: &configv1alpha1.CronExecutionConfig{
+					DefaultTimezone: pointer.String("America/New_York"),
+				},
+			},
+			steps: []step{
+				{
+					Name: "Initial time",
+					Time: testutils.Mktime("2022-04-01T10:52:04Z"),
+				},
+				{
+					Name: "Enqueue at 14:00 UTC",
+					Time: testutils.Mktime("2022-04-01T14:00:00Z"),
+					WantEnqueue: []string{
+						keyFunc(cronWorkerJobConfigDaily, testutils.Mktime("2022-04-01T14:00:00Z")),
+					},
+				},
+			},
+		},
+		{
+			name: "Scheduled daily with job configured timezone",
+			jobConfigs: []*execution.JobConfig{
+				cronWorkerJobConfigDailySingapore,
+			},
+			configs: map[configv1alpha1.ConfigName]runtime.Object{
+				configv1alpha1.CronExecutionConfigName: &configv1alpha1.CronExecutionConfig{
+					DefaultTimezone: pointer.String("America/New_York"),
+				},
+			},
+			steps: []step{
+				{
+					Name: "Initial time",
+					Time: testutils.Mktime("2022-04-01T10:52:04Z"),
+				},
+				{
+					Name: "No enqueue at 14:00 UTC",
+					Time: testutils.Mktime("2022-04-01T14:00:00Z"),
+				},
+				{
+					Name: "Enqueue at 02:00 UTC next day",
+					Time: testutils.Mktime("2022-04-02T02:00:00Z"),
+					WantEnqueue: []string{
+						keyFunc(cronWorkerJobConfigDailySingapore, testutils.Mktime("2022-04-02T02:00:00Z")),
+					},
+				},
+			},
+		},
+		{
+			name: "Missed too many schedules",
+			jobConfigs: []*execution.JobConfig{
+				cronWorkerJobConfig,
+			},
+			configs: map[configv1alpha1.ConfigName]runtime.Object{
+				configv1alpha1.CronExecutionConfigName: &configv1alpha1.CronExecutionConfig{
+					MaxMissedSchedules: pointer.Int64(3),
+				},
+			},
+			steps: []step{
+				{
+					Name: "Initial time",
+					Time: testutils.Mktime("2022-04-01T10:52:04Z"),
+				},
+				{
+					Name: "Want enqueue at start of minute",
+					Time: testutils.Mktime("2022-04-01T10:53:00Z"),
+					WantEnqueue: []string{
+						keyFunc(cronWorkerJobConfig, testutils.Mktime("2022-04-01T10:53:00Z")),
+					},
+				},
+				{
+					Name: "Enqueue only 3 jobs",
+					Time: testutils.Mktime("2022-04-01T10:59:00Z"),
+					WantEnqueue: []string{
+						keyFunc(cronWorkerJobConfig, testutils.Mktime("2022-04-01T10:54:00Z")),
+						keyFunc(cronWorkerJobConfig, testutils.Mktime("2022-04-01T10:55:00Z")),
+						keyFunc(cronWorkerJobConfig, testutils.Mktime("2022-04-01T10:56:00Z")),
+					},
+				},
+				{
+					Name: "Enqueue once for next minute",
+					Time: testutils.Mktime("2022-04-01T11:00:00Z"),
+					WantEnqueue: []string{
+						keyFunc(cronWorkerJobConfig, testutils.Mktime("2022-04-01T11:00:00Z")),
+					},
+				},
+			},
+		},
+		{
+			name: "No future schedules",
+			jobConfigs: []*execution.JobConfig{
+				jobConfigPointInTime,
+			},
+			steps: []step{
+				{
+					Name: "Initial time",
+					Time: testutils.Mktime("2021-02-09T10:52:04Z"),
+				},
+				{
+					Name: "Enqueue job",
+					Time: testutils.Mktime("2021-02-09T12:00:00Z"),
+					WantEnqueue: []string{
+						keyFunc(jobConfigPointInTime, testutils.Mktime("2021-02-09T12:00:00Z")),
+					},
+				},
+				{
+					Name: "No more future schedules",
+					Time: testutils.Mktime("2099-12-31T23:59:59Z"),
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			testutils.SetLogLevel(tt.log)
+
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
+			var prevtime time.Time
+			fakeClock := clock.NewFakeClock(prevtime)
+			croncontroller.Clock = fakeClock
+
+			// Initialize contexts and worker
+			c := mock.NewContext()
+			c.MockConfigs().SetConfigs(tt.configs)
+			ctrlContext := croncontroller.NewContext(c)
+			queue := newEnqueueHandler()
+			worker := croncontroller.NewCronWorker(ctrlContext, queue)
+			executionClient := c.MockClientsets().Furiko().ExecutionV1alpha1()
+
+			// Use custom handler to intercept updates.
+			handler := newNotifyingUpdateHandler(croncontroller.NewUpdateHandler(ctrlContext))
+
+			// Initialize InformerWorker, so that we can receive events from clientset
+			// updates.
+			informer := croncontroller.NewInformerWorker(ctrlContext, handler)
+			informer.Init()
+
+			// Start context.
+			assert.NoError(t, c.Start(ctx))
+
+			// Initialize fixtures
+			for _, jobConfig := range tt.jobConfigs {
+				_, err := executionClient.JobConfigs(jobConfig.Namespace).Create(ctx, jobConfig, metav1.CreateOptions{})
+				assert.NoError(t, err)
+			}
+
+			// Wait for cache sync
+			if !cache.WaitForCacheSync(ctx.Done(), ctrlContext.HasSynced...) {
+				assert.FailNow(t, "caches not synced")
+			}
+
+			// Start CronWorker in background. This is mainly a no-op because we will be
+			// triggering Work manually in each step below.
+			worker.Start(ctx)
+
+			for _, step := range tt.steps {
+				msgAndArgs := []interface{}{
+					fmt.Sprintf("Error in step: %v", step.Name),
+				}
+
+				// Update clock, ensure to never move backwards in time.
+				if !step.Time.IsZero() {
+					if prevtime.After(step.Time) {
+						t.Errorf("cannot set now to %v, previous time was %v", step.Time, prevtime)
+						return
+					}
+					prevtime = step.Time
+					fakeClock.SetTime(step.Time)
+				}
+
+				// Perform update step.
+				if jobConfig := step.Update; jobConfig != nil {
+					_, err := executionClient.JobConfigs(jobConfig.Namespace).
+						Update(ctx, jobConfig, metav1.UpdateOptions{})
+					assert.NoError(t, err)
+					handler.Wait()
+				}
+
+				// Perform delete step.
+				if jobConfig := step.Delete; jobConfig != nil {
+					err := executionClient.JobConfigs(jobConfig.Namespace).
+						Delete(ctx, jobConfig.Name, metav1.DeleteOptions{})
+					assert.NoError(t, err)
+					handler.Wait()
+				}
+
+				// Trigger work manually.
+				worker.Work()
+
+				// Consume from queue.
+				for _, key := range step.WantEnqueue {
+					item, ok := queue.Get()
+					assert.True(t, ok, msgAndArgs...)
+					assert.Equal(t, item, key, msgAndArgs...)
+				}
+
+				// Queue should now be empty.
+				assert.Equal(t, 0, queue.Len(), msgAndArgs...)
+			}
+		})
+	}
+}
+
+type enqueueHandler struct {
+	queue []string
+	mu    sync.RWMutex
+}
+
+func newEnqueueHandler() *enqueueHandler {
+	return &enqueueHandler{}
+}
+
+var _ croncontroller.EnqueueHandler = (*enqueueHandler)(nil)
+
+func (h *enqueueHandler) EnqueueJobConfig(jobConfig *execution.JobConfig, scheduleTime time.Time) error {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.queue = append(h.queue, keyFunc(jobConfig, scheduleTime))
+	return nil
+}
+
+func (h *enqueueHandler) Get() (string, bool) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	if len(h.queue) == 0 {
+		return "", false
+	}
+	item := h.queue[0]
+	h.queue = h.queue[1:]
+	return item, true
+}
+
+func (h *enqueueHandler) Len() int {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+	return len(h.queue)
+}
+
+func keyFunc(jobConfig *execution.JobConfig, scheduleTime time.Time) string {
+	key, err := croncontroller.JobConfigKeyFunc(jobConfig, scheduleTime)
+	if err != nil {
+		panic(err)
+	}
+	return key
+}
+
+type notifyingUpdateHandler struct {
+	handler croncontroller.UpdateHandler
+	cond    chan struct{}
+}
+
+func newNotifyingUpdateHandler(handler croncontroller.UpdateHandler) *notifyingUpdateHandler {
+	return &notifyingUpdateHandler{
+		handler: handler,
+		cond:    make(chan struct{}),
+	}
+}
+
+var _ croncontroller.UpdateHandler = (*notifyingUpdateHandler)(nil)
+
+func (h *notifyingUpdateHandler) Wait() {
+	<-h.cond
+}
+
+func (h *notifyingUpdateHandler) OnUpdate(jobConfig *execution.JobConfig) {
+	h.handler.OnUpdate(jobConfig)
+	h.cond <- struct{}{}
+}
diff --git a/pkg/execution/controllers/croncontroller/cron_timer_test.go b/pkg/execution/controllers/croncontroller/ticker_test.go
similarity index 92%
rename from pkg/execution/controllers/croncontroller/cron_timer_test.go
rename to pkg/execution/controllers/croncontroller/ticker_test.go
index f07a8b0..ab019ec 100644
--- a/pkg/execution/controllers/croncontroller/cron_timer_test.go
+++ b/pkg/execution/controllers/croncontroller/ticker_test.go
@@ -28,12 +28,12 @@ import (
 )
 
 const (
-	realTimeProcessingDelay = time.Millisecond * 10
+	realTimeProcessingDelay = time.Millisecond * 20
 )
 
 var initialTime = time.Unix(1629701607, 5e8)
 
-func TestCronTimerUntil(t *testing.T) {
+func TestClockTickUntil(t *testing.T) {
 	var i int64
 	work := func() {
 		atomic.AddInt64(&i, 1)
@@ -44,7 +44,7 @@ func TestCronTimerUntil(t *testing.T) {
 	step := 5
 	timestep := time.Second * time.Duration(step)
 	stopCh := make(chan struct{}, 1)
-	go croncontroller.CronTimerUntil(work, timestep, stopCh)
+	go croncontroller.ClockTickUntil(work, timestep, stopCh)
 
 	set := func(t time.Time) {
 		fakeClock.SetTime(t)
@@ -53,7 +53,6 @@ func TestCronTimerUntil(t *testing.T) {
 
 	// Initialise with fake time.
 	set(initialTime)
-	assert.Equal(t, int64(0), i, "should be initialized to 0")
 
 	// Should be called immediately (i.e. 1ns later).
 	set(initialTime.Add(time.Nanosecond))
@@ -96,7 +95,7 @@ func TestCronTimerUntil_Seconds(t *testing.T) {
 
 	timestep := time.Second
 	stopCh := make(chan struct{}, 1)
-	go croncontroller.CronTimerUntil(work, timestep, stopCh)
+	go croncontroller.ClockTickUntil(work, timestep, stopCh)
 
 	set := func(t time.Time) {
 		fakeClock.SetTime(t)
@@ -105,7 +104,6 @@ func TestCronTimerUntil_Seconds(t *testing.T) {
 
 	// Initialise with fake time.
 	set(initialTime)
-	assert.Equal(t, int64(0), i, "should be initialized to 0")
 
 	// Should be called immediately (i.e. 1ns later).
 	set(initialTime.Add(time.Nanosecond))
@@ -132,7 +130,7 @@ func TestCronTimerUntil_Seconds(t *testing.T) {
 	assert.Equal(t, int64(4), i, "work() should be called immediately at start of next timestep")
 
 	// Jump a few timesteps.
-	jump := 300
+	jump := 10
 	for i := 1; i <= jump; i++ {
 		set(time.Unix(int64(1629701610+i), 0))
 	}
diff --git a/pkg/execution/controllers/jobcontroller/reconciler_test.go b/pkg/execution/controllers/jobcontroller/reconciler_test.go
index 57a6576..1e37c3a 100644
--- a/pkg/execution/controllers/jobcontroller/reconciler_test.go
+++ b/pkg/execution/controllers/jobcontroller/reconciler_test.go
@@ -381,9 +381,7 @@ func TestReconciler(t *testing.T) {
 			reconciler := jobcontroller.NewReconciler(ctrlCtx, &configv1alpha1.Concurrency{
 				Workers: 1,
 			})
-			for configName, cfg := range tt.configs {
-				c.MockConfigs().MockConfigLoader().SetConfig(configName, cfg)
-			}
+			c.MockConfigs().SetConfigs(tt.configs)
 
 			err := c.Start(ctx)
 			assert.NoError(t, err)
diff --git a/pkg/execution/mutation/mutation_test.go b/pkg/execution/mutation/mutation_test.go
index 1943d42..d9b65c4 100644
--- a/pkg/execution/mutation/mutation_test.go
+++ b/pkg/execution/mutation/mutation_test.go
@@ -1206,9 +1206,7 @@ func setup(t *testing.T, cfgs map[configv1alpha1.ConfigName]runtime.Object, rjcs
 	defer cancel()
 
 	ctrlContext := mock.NewContext()
-	for configName, cfg := range cfgs {
-		ctrlContext.MockConfigs().MockConfigLoader().SetConfig(configName, cfg)
-	}
+	ctrlContext.MockConfigs().SetConfigs(cfgs)
 	hasSynced := ctrlContext.Informers().Furiko().Execution().V1alpha1().JobConfigs().Informer().HasSynced
 	if err := ctrlContext.Start(ctx); err != nil {
 		t.Fatal(err)
diff --git a/pkg/execution/validation/validation_test.go b/pkg/execution/validation/validation_test.go
index 02e30de..cf55c72 100644
--- a/pkg/execution/validation/validation_test.go
+++ b/pkg/execution/validation/validation_test.go
@@ -402,9 +402,7 @@ func TestValidateJobConfig(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			original := tt.rjc.DeepCopy()
 			ctrlContext := mock.NewContext()
-			for configName, cfg := range tt.cfgs {
-				ctrlContext.MockConfigs().MockConfigLoader().SetConfig(configName, cfg)
-			}
+			ctrlContext.MockConfigs().SetConfigs(tt.cfgs)
 			if err := ctrlContext.Start(context.Background()); err != nil {
 				t.Fatal(err)
 			}
@@ -1043,9 +1041,7 @@ func setup(t *testing.T, cfgs map[configv1alpha1.ConfigName]runtime.Object, rjcs
 	defer cancel()
 
 	ctrlContext := mock.NewContext()
-	for configName, cfg := range cfgs {
-		ctrlContext.MockConfigs().MockConfigLoader().SetConfig(configName, cfg)
-	}
+	ctrlContext.MockConfigs().SetConfigs(cfgs)
 	hasSynced := ctrlContext.Informers().Furiko().Execution().V1alpha1().JobConfigs().Informer().HasSynced
 	if err := ctrlContext.Start(ctx); err != nil {
 		t.Fatal(err)
diff --git a/pkg/utils/testutils/klog.go b/pkg/utils/testutils/klog.go
new file mode 100644
index 0000000..5911324
--- /dev/null
+++ b/pkg/utils/testutils/klog.go
@@ -0,0 +1,33 @@
+/*
+ * Copyright 2022 The Furiko Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package testutils
+
+import (
+	"flag"
+	"fmt"
+
+	"k8s.io/klog/v2"
+)
+
+// SetLogLevel updates the default klog log level.
+func SetLogLevel(level klog.Level) {
+	flagSet := flag.NewFlagSet("test", flag.PanicOnError)
+	klog.InitFlags(flagSet)
+	if err := flagSet.Parse([]string{fmt.Sprintf("-v=%v", level)}); err != nil {
+		panic(err)
+	}
+}