package queue import ( "context" "os" "path/filepath" "slices" "strings" "sync" "testing" "time" ) // controlledSpeaker is a testable Speaker: every announcement reports its // start on started (buffered channel: the worker never blocks) and then // waits until the test releases it via step, so processing is fully // deterministic. A cancelled ctx ends the wait (Cancel/Close paths). All // fields are read only by the test; started order == start order. type controlledSpeaker struct { started chan Item step chan struct{} consumed int // starts consumed by waitUntil (test goroutine only) } func newSpeaker() *controlledSpeaker { return &controlledSpeaker{started: make(chan Item, 16), step: make(chan struct{})} } func (g *controlledSpeaker) speak(ctx context.Context, item Item) error { g.started <- item select { case <-g.step: return nil case <-ctx.Done(): return ctx.Err() } } // waitUntil waits until n announcements have started in total and returns // the ones started since the previous waitUntil call. func (g *controlledSpeaker) waitUntil(t *testing.T, n int) []Item { t.Helper() var items []Item deadline := time.After(2 * time.Second) for g.consumed < n { select { case it := <-g.started: g.consumed++ items = append(items, it) case <-deadline: t.Fatalf("only %d of %d announcements started", g.consumed, n) } } return items } // noFurther checks that no further announcement starts (poll window). func (g *controlledSpeaker) noFurther(t *testing.T) { t.Helper() select { case it := <-g.started: g.consumed++ t.Fatalf("unexpected announcement %q started", it.Text) case <-time.After(150 * time.Millisecond): } } // TestProcessingOrder: now before high before normal before low, FIFO per // level. The first blocker occupies the worker; then all levels are // enqueued at once and processing must follow the class. // (No laws: zero Laws = unlimited, no quiet hours.) func TestProcessingOrder(t *testing.T) { g := newSpeaker() q := New(Laws{}, g.speak, nil) defer q.Close() adm, err := q.Enqueue(Item{Text: "blocker", Priority: PriorityNormal}) if err != nil { t.Fatal(err) } if adm.ID == "" || adm.Discarded != "" { t.Fatalf("Enqueue = %+v, want admission with ID", adm) } g.waitUntil(t, 1) // blocker is playing, worker is busy for _, it := range []Item{ {Text: "low", Priority: PriorityLow}, {Text: "second-low", Priority: PriorityLow}, {Text: "normal", Priority: PriorityNormal}, {Text: "urgent", Priority: PriorityHigh}, {Text: "right-now", Now: true}, } { if _, err := q.Enqueue(it); err != nil { t.Fatal(err) } } g.noFurther(t) // serialization: the worker is still on the blocker // Release: everything else runs through in class order. close(g.step) rest := g.waitUntil(t, 6) var texts []string for _, it := range rest { texts = append(texts, it.Text) } want := []string{"right-now", "urgent", "normal", "low", "second-low"} if !slices.Equal(texts, want) { t.Errorf("processing after the blocker = %v, want %v", texts, want) } } // TestCurrentAndPending: status view of the playing item and the queue. func TestCurrentAndPending(t *testing.T) { g := newSpeaker() q := New(Laws{}, g.speak, nil) defer q.Close() if _, ok := q.Current(); ok { t.Fatal("Current before the first announcement: want idle") } q.Enqueue(Item{Text: "blocker"}) g.waitUntil(t, 1) q.Enqueue(Item{Text: "high", Priority: PriorityHigh}) q.Enqueue(Item{Text: "low", Priority: PriorityLow}) if cur, ok := q.Current(); !ok || cur.Text != "blocker" { t.Errorf("Current = %+v (ok=%v), want the blocker", cur, ok) } var pendingTexts []string for _, it := range q.Pending() { pendingTexts = append(pendingTexts, it.Text) } if !slices.Equal(pendingTexts, []string{"high", "low"}) { t.Errorf("Pending = %v, want [high low]", pendingTexts) } close(g.step) g.waitUntil(t, 3) } // TestCancel: the playing announcement is aborted via ctx, the queue // emptied, the total reported; afterwards the queue admits and plays again. func TestCancel(t *testing.T) { g := newSpeaker() q := New(Laws{}, g.speak, nil) defer q.Close() q.Enqueue(Item{Text: "blocker"}) g.waitUntil(t, 1) q.Enqueue(Item{Text: "waiting-one"}) q.Enqueue(Item{Text: "waiting-two"}) if n := q.Cancel(); n != 3 { t.Errorf("Cancel = %d, want 3 (playing + 2 waiting)", n) } g.noFurther(t) // blocker ended via ctx, waiting items are gone waitIdle(t, q) if n := q.Cancel(); n != 0 { t.Errorf("Cancel on empty queue = %d, want 0", n) } // The queue admits and plays again: adm, err := q.Enqueue(Item{Text: "afterwards"}) if err != nil { t.Fatal(err) } g.step <- struct{}{} // releases the waiting announcement in the speaker started := g.waitUntil(t, 2) if len(started) != 1 || started[0].ID != adm.ID { t.Errorf("new announcement not started: %+v", started) } } // TestClose: waiting items are dropped, the playing announcement ends via // the cancelled ctx, Close waits for the worker to exit, afterwards Enqueue // is an error (and Close idempotent). func TestClose(t *testing.T) { g := newSpeaker() q := New(Laws{}, g.speak, nil) q.Enqueue(Item{Text: "blocker"}) g.waitUntil(t, 1) q.Enqueue(Item{Text: "waiting"}) q.Close() g.noFurther(t) if _, ok := q.Current(); ok { t.Error("Current after Close: want idle") } if _, err := q.Enqueue(Item{Text: "too late"}); err == nil { t.Error("Enqueue after Close: want error") } q.Close() // idempotent } // TestEnqueueAssignsIDs: IDs are short and unique. func TestEnqueueAssignsIDs(t *testing.T) { g := newSpeaker() q := New(Laws{}, g.speak, nil) defer q.Close() seen := map[string]bool{} for i := 0; i < 20; i++ { adm, err := q.Enqueue(Item{Text: "x"}) if err != nil { t.Fatal(err) } if len(adm.ID) < 4 || seen[adm.ID] { t.Errorf("ID %q too short or duplicate", adm.ID) } seen[adm.ID] = true } } // waitIdle polls until no item is playing (Cancel/Close release that // synchronously after the worker exits; polling covers the transition // without trusting timing). func waitIdle(t *testing.T, q *Queue) { t.Helper() deadline := time.Now().Add(time.Second) for { if _, ok := q.Current(); !ok { return } if time.Now().After(deadline) { t.Fatal("queue does not become idle") } time.Sleep(5 * time.Millisecond) } } // TestMuteStore: file round trip, missing means not muted, atomic Set. func TestMuteStore(t *testing.T) { dir := t.TempDir() path := filepath.Join(dir, MuteFileName) var s MuteStore if s.Muted() { t.Error("MuteStore without path: Muted() = true, want false") } if err := s.Set(true); err == nil { t.Error("Set without path: want error") } s = MuteStore{Path: path} if s.Muted() { t.Error("without file: Muted() = true, want false (never silent by accident)") } if err := s.Set(true); err != nil { t.Fatal(err) } if !s.Muted() { t.Error("after Set(true): Muted() = false") } if b, _ := os.ReadFile(path); string(b) != "true\n" { t.Errorf("file content = %q, want %q", b, "true\n") } if err := s.Set(false); err != nil { t.Fatal(err) } if s.Muted() { t.Error("after Set(false): Muted() = true") } // Garbage in the file means not muted (only "true" is muted). if err := os.WriteFile(path, []byte("maybe\n"), 0o644); err != nil { t.Fatal(err) } if s.Muted() { t.Error("file content \"maybe\": Muted() = true, want false") } } // TestRateLimiter: the limiter's own semantics. func TestRateLimiter(t *testing.T) { // Unlimited: interval 0 always allows. unlimited := &RateLimiter{} for i := 0; i < 3; i++ { if !unlimited.Allow() { t.Fatal("interval 0: Allow = false, want true (unlimited)") } unlimited.Reserve() } // Limited: after Reserve, Allow only after the interval. l := &RateLimiter{Interval: 30 * time.Millisecond} if !l.Allow() { t.Fatal("fresh limiter: Allow = false") } l.Reserve() if l.Allow() { t.Fatal("right after Reserve: Allow = true, want false") } if next := l.Next(); time.Now().After(next) { t.Errorf("Next() = %v lies in the past", next) } time.Sleep(40 * time.Millisecond) if !l.Allow() { t.Fatal("after the interval: Allow = false, want true") } } // windowAround builds an "HH:MM-HH:MM" quiet window that safely contains t // (three minutes before to four minutes after; minute resolution). func windowAround(t time.Time) string { return t.Add(-3*time.Minute).Format("15:04") + "-" + t.Add(4*time.Minute).Format("15:04") } // waitForCond polls a condition (time laws are real; the controlled speaker // alone cannot see them). func waitForCond(t *testing.T, timeout time.Duration, cond func() bool, what string) { t.Helper() deadline := time.Now().Add(timeout) for !cond() { if time.Now().After(deadline) { t.Fatalf("%s did not happen within %s", what, timeout) } time.Sleep(5 * time.Millisecond) } } // TestRateLimitLowDiscardedNormalParks: with a short window the first // announcement plays at once; low over budget is discarded on admission // (Admission names the reason), normal parks and plays after the window. func TestRateLimitLowDiscardedNormalParks(t *testing.T) { g := newSpeaker() q := New(Laws{RateLimit: 400 * time.Millisecond}, g.speak, nil) defer q.Close() if adm, err := q.Enqueue(Item{Text: "first"}); err != nil || adm.ID == "" || adm.Discarded != "" { t.Fatalf("first Enqueue = %+v (%v), want admission", adm, err) } g.waitUntil(t, 1) // first is playing, the window is booked adm, err := q.Enqueue(Item{Text: "low-one", Priority: PriorityLow}) if err != nil { t.Fatal(err) } if adm.ID != "" || adm.Discarded != DiscardRateLimit || adm.Toast { t.Errorf("low over budget = %+v, want discard %q", adm, DiscardRateLimit) } if adm, err = q.Enqueue(Item{Text: "low-two", Priority: PriorityLow}); err != nil || adm.Discarded != DiscardRateLimit { t.Errorf("second low discard = %+v (%v)", adm, err) } adm, err = q.Enqueue(Item{Text: "second"}) if err != nil || adm.ID == "" || adm.Discarded != "" { t.Fatalf("normal = %+v (%v), want admission (parks until the slot)", adm, err) } // Status view while the first plays: second waits (normal, parked). waitForCond(t, time.Second, func() bool { for _, it := range q.Pending() { if it.Text == "second" { return true } } return false }, "parked normal announcement in Pending") close(g.step) // end the first; second must keep parking … g.noFurther(t) // … and start after the window elapses (deadline covers jitter easily). started := g.waitUntil(t, 2) if len(started) != 1 || started[0].Text != "second" { t.Errorf("after the window = %+v, want second", started) } // low-one/low-two never play: exactly 2 starts in total. g.noFurther(t) } // TestHighBurstOnAndOff: with burst rights high starts right after the // playing announcement although the window is full; without rights high // parks like normal (no violation). func TestHighBurstOnAndOff(t *testing.T) { g := newSpeaker() q := New(Laws{RateLimit: 10 * time.Second, AllowHighBurst: true}, g.speak, nil) q.Enqueue(Item{Text: "first"}) g.waitUntil(t, 1) if adm, err := q.Enqueue(Item{Text: "urgent", Priority: PriorityHigh}); err != nil || adm.ID == "" { t.Fatalf("high with burst rights = %+v (%v), want admission", adm, err) } g.step <- struct{}{} // end the first started := g.waitUntil(t, 2) // high starts despite the full window if len(started) != 1 || started[0].Text != "urgent" { t.Errorf("burst = %+v, want urgent right away", started) } q.Close() g2 := newSpeaker() q2 := New(Laws{RateLimit: 10 * time.Second, AllowHighBurst: false}, g2.speak, nil) defer q2.Close() q2.Enqueue(Item{Text: "first"}) g2.waitUntil(t, 1) // first is playing, window booked if adm, err := q2.Enqueue(Item{Text: "urgent", Priority: PriorityHigh}); err != nil || adm.ID == "" { t.Fatalf("high without burst rights = %+v (%v), want admission (parks)", adm, err) } g2.step <- struct{}{} // end the first g2.noFurther(t) // high waits for the slot: no violation without rights } // TestNowBypassesRateLimit: now starts at once despite a full window, the // documented escape (it still books a new window). func TestNowBypassesRateLimit(t *testing.T) { g := newSpeaker() q := New(Laws{RateLimit: 10 * time.Second}, g.speak, nil) defer q.Close() q.Enqueue(Item{Text: "first"}) g.waitUntil(t, 1) // window booked (10 s) if adm, err := q.Enqueue(Item{Text: "right-now", Now: true}); err != nil || adm.ID == "" || adm.Discarded != "" { t.Fatalf("now over a full window = %+v (%v), want admission", adm, err) } g.step <- struct{}{} started := g.waitUntil(t, 2) // long before the window elapses (10 s), else timeout if len(started) != 1 || started[0].Text != "right-now" { t.Errorf("now = %+v, want right-now", started) } // The now break-out booked a new window: normal parks now. if adm, err := q.Enqueue(Item{Text: "afterwards"}); err != nil || adm.ID == "" { t.Fatalf("normal after now = %+v (%v)", adm, err) } close(g.step) g.noFurther(t) // parks until the slot, not immediately } // TestQuietHoursDiscardAndToast: an active window discards normal and high // (burst is a rate right, not a quiet right); toast replaces instead; now // deliberately breaks out. func TestQuietHoursDiscardAndToast(t *testing.T) { window := windowAround(time.Now()) g := newSpeaker() q := New(Laws{Quiet: mustQuiet(t, window), QuietMode: QuietModeDiscard}, g.speak, nil) defer q.Close() adm, err := q.Enqueue(Item{Text: "normal"}) if err != nil || adm.ID != "" || adm.Discarded != DiscardQuietHours || adm.Toast { t.Errorf("normal in quiet hours = %+v (%v), want silent discard", adm, err) } if adm, err = q.Enqueue(Item{Text: "urgent", Priority: PriorityHigh}); err != nil || adm.Discarded != DiscardQuietHours { t.Errorf("high in quiet hours = %+v (%v), want discard too", adm, err) } adm, err = q.Enqueue(Item{Text: "right-now", Now: true}) if err != nil || adm.ID == "" || adm.Discarded != "" { t.Fatalf("now in quiet hours = %+v (%v), want break-out", adm, err) } started := g.waitUntil(t, 1) if len(started) != 1 || started[0].Text != "right-now" { t.Errorf("in quiet hours = %+v, want only right-now", started) } g.noFurther(t) g2 := newSpeaker() q2 := New(Laws{Quiet: mustQuiet(t, window), QuietMode: QuietModeToast}, g2.speak, nil) defer q2.Close() toasted := make(chan Item, 1) q2.OnToast(func(it Item) { toasted <- it }) adm, err = q2.Enqueue(Item{Text: "normal"}) if err != nil || adm.Discarded != DiscardQuietHours || !adm.Toast { t.Errorf("normal in quiet hours (toast) = %+v (%v), want toast replacement", adm, err) } select { case it := <-toasted: if it.Text != "normal" { t.Errorf("toast hook got %q", it.Text) } case <-time.After(time.Second): t.Error("toast hook not fired") } g2.noFurther(t) } // mustQuiet parses a quiet window or fails the test. func mustQuiet(t *testing.T, window string) QuietHours { t.Helper() q, err := ParseQuietHours(window) if err != nil { t.Fatal(err) } return q } // TestLawState: unlimited yields an open slot; after a booked window the // next slot lies in the future; the quiet window and limits are visible. func TestLawState(t *testing.T) { g := newSpeaker() q := New(Laws{}, g.speak, nil) st := q.LawState() if !st.NextSlot.IsZero() || st.RateLimit != 0 { t.Errorf("unlimited = %+v, want zero-time slot", st) } q.Close() // The quiet window must be inactive NOW, otherwise the announcement // below would be toasted instead of spoken. A fixed window would be a // time bomb (red at certain hours), hence dynamic: one hour starting two // hours from now. window := inactiveQuietWindow(time.Now()) quiet := mustQuiet(t, window) if quiet.Active(time.Now()) { t.Fatalf("test window %s is active, must not be", window) } q2 := New(Laws{RateLimit: time.Hour, AllowHighBurst: true, Quiet: quiet, QuietMode: QuietModeToast, MaxText: 400, MaxAge: 2 * time.Minute}, g.speak, nil) defer q2.Close() st2 := q2.LawState() if st2.RateLimit != time.Hour || !st2.AllowHighBurst || st2.MaxText != 400 || st2.MaxAge != 2*time.Minute { t.Errorf("LawState = %+v, want 1h window with burst rights and limits 400/2m", st2) } if st2.Quiet.String() != window || st2.QuietMode != QuietModeToast { t.Errorf("quiet hours in LawState = %q/%q, want %q/toast", st2.Quiet, st2.QuietMode, window) } if st2.QuietActive || !st2.QuietUntil.IsZero() { t.Errorf("quiet hours expected inactive, LawState reports active=%v until=%v", st2.QuietActive, st2.QuietUntil) } q2.Enqueue(Item{Text: "first"}) g.waitUntil(t, 1) if next := q2.LawState().NextSlot; !next.After(time.Now()) { t.Errorf("after start = %v, want slot in the future", next) } g.step <- struct{}{} } // inactiveQuietWindow returns a one-hour "HH:MM-HH:MM" window starting two // hours after now; it never contains now, whatever the time of day (also // across midnight). func inactiveQuietWindow(now time.Time) string { start := now.Add(2 * time.Hour) end := now.Add(3 * time.Hour) return start.Format("15:04") + "-" + end.Format("15:04") } // TestInactiveQuietWindow: the helper window must not be active at ANY time // of day, otherwise TestLawState would be a time bomb again. func TestInactiveQuietWindow(t *testing.T) { base := time.Date(2026, 10, 24, 0, 0, 0, 0, time.Local) for m := 0; m < 24*60; m += 7 { now := base.Add(time.Duration(m) * time.Minute) w := inactiveQuietWindow(now) q, err := ParseQuietHours(w) if err != nil { t.Fatalf("%s: %v", now.Format("15:04"), err) } if q.Active(now) { t.Errorf("at %s the window %s is active", now.Format("15:04"), w) } } } // syncLog is a concurrency-safe log buffer for worker lines. type syncLog struct { mu sync.Mutex b strings.Builder } func (l *syncLog) Write(p []byte) (int, error) { l.mu.Lock() defer l.mu.Unlock() return l.b.Write(p) } func (l *syncLog) String() string { l.mu.Lock() defer l.mu.Unlock() return l.b.String() } // TestLogFinishedNotCancelled: a normally finished announcement is logged as // "finished", not "cancelled"; the worker must not read its own cancel() as // a cancellation. A real Cancel does log "cancelled". func TestLogFinishedNotCancelled(t *testing.T) { logb := &syncLog{} g := newSpeaker() q := New(Laws{}, g.speak, logb) defer q.Close() q.Enqueue(Item{Text: "runs to the end"}) g.waitUntil(t, 1) g.step <- struct{}{} waitFor(t, func() bool { return strings.Contains(logb.String(), "finished after") }) if strings.Contains(logb.String(), "cancelled") { t.Errorf("finished announcement logged as cancelled:\n%s", logb.String()) } q.Enqueue(Item{Text: "gets cancelled"}) g.waitUntil(t, 2) q.Cancel() waitFor(t, func() bool { return strings.Contains(logb.String(), "cancelled") }) } // waitFor polls cond for up to two seconds. func waitFor(t *testing.T, cond func() bool) { t.Helper() deadline := time.Now().Add(2 * time.Second) for !cond() { if time.Now().After(deadline) { t.Fatal("condition not met within 2 s") } time.Sleep(5 * time.Millisecond) } }