restic-manager/internal/agent/scheduler/scheduler_test.go

package scheduler

import (
	"sync"
	"testing"
	"time"

	"gitea.dcglab.co.uk/steve/restic-manager/internal/api"
)

// recSender is a Sender that records every envelope it gets. Tests
// inspect it after a tick to assert the right messages were emitted.
type recSender struct {
	mu   sync.Mutex
	envs []api.Envelope
}

func (r *recSender) Send(env api.Envelope) error {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.envs = append(r.envs, env)
	return nil
}

func (r *recSender) snapshot() []api.Envelope {
	r.mu.Lock()
	defer r.mu.Unlock()
	out := make([]api.Envelope, len(r.envs))
	copy(out, r.envs)
	return out
}

func TestApplyEmitsAck(t *testing.T) {
	t.Parallel()
	tx := &recSender{}
	s := New()
	defer s.Stop()

	s.Apply(api.ScheduleSetPayload{
		Version: 7,
		Schedules: []api.Schedule{
			{ID: "s1", CronExpr: "@hourly", Enabled: true},
		},
	}, tx)

	if got := s.Version(); got != 7 {
		t.Fatalf("Version: got %d, want 7", got)
	}

	envs := tx.snapshot()
	if len(envs) != 1 {
		t.Fatalf("expected 1 envelope (ack), got %d", len(envs))
	}
	if envs[0].Type != api.MsgScheduleAck {
		t.Fatalf("envelope type: got %s, want %s", envs[0].Type, api.MsgScheduleAck)
	}
	var ack api.ScheduleAckPayload
	_ = envs[0].UnmarshalPayload(&ack)
	if ack.Version != 7 {
		t.Fatalf("ack version: got %d", ack.Version)
	}
}

func TestApplyTickFiresScheduleFire(t *testing.T) {
	t.Parallel()
	tx := &recSender{}
	s := New()
	defer s.Stop()

	// Cron expression that fires roughly every second; close enough
	// to be reliable in CI without making the test slow.
	s.Apply(api.ScheduleSetPayload{
		Version: 1,
		Schedules: []api.Schedule{
			{ID: "every-second", CronExpr: "@every 1s", Enabled: true},
		},
	}, tx)

	deadline := time.Now().Add(3 * time.Second)
	for time.Now().Before(deadline) {
		envs := tx.snapshot()
		for _, e := range envs {
			if e.Type == api.MsgScheduleFire {
				var p api.ScheduleFirePayload
				_ = e.UnmarshalPayload(&p)
				if p.ScheduleID == "every-second" {
					return
				}
			}
		}
		time.Sleep(50 * time.Millisecond)
	}
	t.Fatal("schedule.fire did not arrive within 3s")
}

func TestApplyDisabledEntriesSkipped(t *testing.T) {
	t.Parallel()
	tx := &recSender{}
	s := New()
	defer s.Stop()

	s.Apply(api.ScheduleSetPayload{
		Version: 1,
		Schedules: []api.Schedule{
			{ID: "off", CronExpr: "@every 1s", Enabled: false},
		},
	}, tx)

	// A disabled schedule must never fire — give the cron a couple
	// of ticks to confirm it's silent.
	time.Sleep(2200 * time.Millisecond)
	for _, e := range tx.snapshot() {
		if e.Type == api.MsgScheduleFire {
			t.Fatalf("disabled schedule fired: %+v", e)
		}
	}
}

func TestApplyReplacesPriorState(t *testing.T) {
	t.Parallel()
	tx := &recSender{}
	s := New()
	defer s.Stop()

	s.Apply(api.ScheduleSetPayload{
		Version: 1,
		Schedules: []api.Schedule{
			{ID: "old", CronExpr: "@every 1s", Enabled: true},
		},
	}, tx)

	// Wait long enough for the first version to fire at least once.
	time.Sleep(1500 * time.Millisecond)

	// Now replace with version 2 that doesn't include "old".
	s.Apply(api.ScheduleSetPayload{
		Version:   2,
		Schedules: []api.Schedule{},
	}, tx)

	// Snapshot count *after* the replacement.
	before := 0
	for _, e := range tx.snapshot() {
		if e.Type == api.MsgScheduleFire {
			before++
		}
	}
	time.Sleep(2 * time.Second)
	after := 0
	for _, e := range tx.snapshot() {
		if e.Type == api.MsgScheduleFire {
			after++
		}
	}
	if after != before {
		t.Fatalf("schedule.fire count grew after replacement (before=%d after=%d) — old cron still firing",
			before, after)
	}
}