talk/internal/queue/queue_test.go

package queue

import (
	"context"
	"os"
	"path/filepath"
	"slices"
	"testing"
	"time"
)

// gesteuerteSpielerin ist ein testbarer Speaker: Jede Ansage meldet ihren
// Start auf started (Kanal-Puffer: der Worker blockiert nie) und wartet
// dann, bis der Test sie per step freigibt — so ist die Abarbeitung
// vollständig deterministisch steuerbar. Ein abgebrochener ctx beendet das
// Warten (Cancel-/Close-Pfade). Alle Felder werden nur vom Test gelesen;
// started-Reihenfolge == Start-Reihenfolge.
type gesteuerteSpielerin struct {
	started  chan Item
	step     chan struct{}
	consumed int // von warteBis konsumierte Starts (nur Test-Goroutine)
}

func neueSpielerin() *gesteuerteSpielerin {
	return &gesteuerteSpielerin{started: make(chan Item, 16), step: make(chan struct{})}
}

func (g *gesteuerteSpielerin) speak(ctx context.Context, item Item) error {
	g.started <- item
	select {
	case <-g.step:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

// warteBis wartet, bis insgesamt n Ansagen gestartet sind, und liefert die
// seit dem letzten warteBis-Aufruf neu gestarteten.
func (g *gesteuerteSpielerin) warteBis(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("nur %d von %d Ansagen gestartet", g.consumed, n)
		}
	}
	return items
}

// keineWeitere prüft, dass keine weitere Ansage startet (Poll-Fenster).
func (g *gesteuerteSpielerin) keineWeitere(t *testing.T) {
	t.Helper()
	select {
	case it := <-g.started:
		g.consumed++
		t.Fatalf("unerwartete Ansage %q gestartet", it.Text)
	case <-time.After(150 * time.Millisecond):
	}
}

// TestAbarbeitungsReihenfolge: now vor high vor normal vor low, FIFO je
// Level. Der erste Blocker besetzt den Worker; danach werden alle Level
// gleichzeitig eingereiht — die Abarbeitung muss der Klasse folgen.
func TestAbarbeitungsReihenfolge(t *testing.T) {
	g := neueSpielerin()
	q := New(g.speak, nil)
	defer q.Close()

	erster, err := q.Enqueue(Item{Text: "blocker", Priority: PriorityNormal})
	if err != nil {
		t.Fatal(err)
	}
	if erster == "" {
		t.Fatal("Enqueue vergibt keine ID")
	}
	g.warteBis(t, 1) // Blocker läuft, Worker gebunden.

	for _, it := range []Item{
		{Text: "niedrig", Priority: PriorityLow},
		{Text: "zweit-niedrig", Priority: PriorityLow},
		{Text: "mittel", Priority: PriorityNormal},
		{Text: "dringend", Priority: PriorityHigh},
		{Text: "sofort", Now: true},
	} {
		if _, err := q.Enqueue(it); err != nil {
			t.Fatal(err)
		}
	}
	g.keineWeitere(t) // Serialisierung: Worker spielt noch den Blocker.

	// Freigeben: alles Übrige läuft in Klassen-Reihenfolge durch.
	close(g.step)
	rest := g.warteBis(t, 6)

	var texts []string
	for _, it := range rest {
		texts = append(texts, it.Text)
	}
	want := []string{"sofort", "dringend", "mittel", "niedrig", "zweit-niedrig"}
	if !slices.Equal(texts, want) {
		t.Errorf("Abarbeitung nach dem Blocker = %v, want %v", texts, want)
	}
}

// TestCurrentUndPending: Status-Sicht auf laufendes Item + Warteschlange.
func TestCurrentUndPending(t *testing.T) {
	g := neueSpielerin()
	q := New(g.speak, nil)
	defer q.Close()

	if _, ok := q.Current(); ok {
		t.Fatal("Current vor erster Ansage: want idle")
	}

	q.Enqueue(Item{Text: "blocker"})
	g.warteBis(t, 1)

	q.Enqueue(Item{Text: "hoch", Priority: PriorityHigh})
	q.Enqueue(Item{Text: "niedrig", Priority: PriorityLow})

	if cur, ok := q.Current(); !ok || cur.Text != "blocker" {
		t.Errorf("Current = %+v (ok=%v), will Blocker", cur, ok)
	}
	var pendingTexts []string
	for _, it := range q.Pending() {
		pendingTexts = append(pendingTexts, it.Text)
	}
	if !slices.Equal(pendingTexts, []string{"hoch", "niedrig"}) {
		t.Errorf("Pending = %v, want [hoch niedrig]", pendingTexts)
	}

	close(g.step)
	g.warteBis(t, 3)
}

// TestCancel: laufende Ansage wird über ctx abgebrochen, Queue geleert,
// Gesamtzahl zurückgemeldet; danach ist die Queue wieder aufnahme- und
// spielbereit.
func TestCancel(t *testing.T) {
	g := neueSpielerin()
	q := New(g.speak, nil)
	defer q.Close()

	q.Enqueue(Item{Text: "blocker"})
	g.warteBis(t, 1)

	q.Enqueue(Item{Text: "wartet-eins"})
	q.Enqueue(Item{Text: "wartet-zwei"})

	if n := q.Cancel(); n != 3 {
		t.Errorf("Cancel = %d, want 3 (laufende + 2 wartende)", n)
	}

	g.keineWeitere(t) // Blocker durch ctx beendet, wartende Items sind weg.
	waitIdle(t, q)

	if n := q.Cancel(); n != 0 {
		t.Errorf("Cancel auf leerer Queue = %d, want 0", n)
	}

	// Queue nimmt wieder an und spielt:
	id, err := q.Enqueue(Item{Text: "danach"})
	if err != nil {
		t.Fatal(err)
	}
	g.step <- struct{}{} // gibt die wartende Ansage im Speaker frei
	gestartet := g.warteBis(t, 2)
	if len(gestartet) != 1 || gestartet[0].ID != id {
		t.Errorf("neue Ansage nicht gestartet: %+v", gestartet)
	}
}

// TestClose: wartende Items werden verworfen, die laufende Ansage über den
// abgebrochenen ctx beendet, Close wartet auf den Worker-Exit, danach ist
// Enqueue ein Fehler (und Close idempotent).
func TestClose(t *testing.T) {
	g := neueSpielerin()
	q := New(g.speak, nil)

	q.Enqueue(Item{Text: "blocker"})
	g.warteBis(t, 1)
	q.Enqueue(Item{Text: "wartet"})

	q.Close()

	g.keineWeitere(t)
	if _, ok := q.Current(); ok {
		t.Error("Current nach Close: want idle")
	}
	if _, err := q.Enqueue(Item{Text: "zu spät"}); err == nil {
		t.Error("Enqueue nach Close: want Fehler")
	}
	q.Close() // idempotent
}

// TestEnqueueVergibtIDs: IDs sind kurz und eindeutig.
func TestEnqueueVergibtIDs(t *testing.T) {
	g := neueSpielerin()
	q := New(g.speak, nil)
	defer q.Close()

	seen := map[string]bool{}
	for i := 0; i < 20; i++ {
		id, err := q.Enqueue(Item{Text: "x"})
		if err != nil {
			t.Fatal(err)
		}
		if len(id) < 4 || seen[id] {
			t.Errorf("ID %q zu kurz oder doppelt", id)
		}
		seen[id] = true
	}
}

// waitIdle pollt, bis kein Item mehr läuft (Cancel-/Close-Pfade geben das
// nach dem Worker-Exit synchron frei; das Pollen deckt die Übergangsphase
// ab, ohne auf Timing zu trauen).
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 wird nicht idle")
		}
		time.Sleep(5 * time.Millisecond)
	}
}

// TestMuteStore: Datei-Roundtrip, Fehlend-Heißt-Nicht-Stumm, atomares Set.
func TestMuteStore(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, MuteFileName)

	var s MuteStore
	if s.Muted() {
		t.Error("MuteStore ohne Pfad: Muted() = true, want false")
	}
	if err := s.Set(true); err == nil {
		t.Error("Set ohne Pfad: want Fehler")
	}

	s = MuteStore{Path: path}
	if s.Muted() {
		t.Error("ohne Datei: Muted() = true, want false (nie still aus Versehen)")
	}

	if err := s.Set(true); err != nil {
		t.Fatal(err)
	}
	if !s.Muted() {
		t.Error("nach Set(true): Muted() = false")
	}
	if b, _ := os.ReadFile(path); string(b) != "true\n" {
		t.Errorf("Datei-Inhalt = %q, want %q", b, "true\n")
	}

	if err := s.Set(false); err != nil {
		t.Fatal(err)
	}
	if s.Muted() {
		t.Error("nach Set(false): Muted() = true")
	}

	// Müll in der Datei heißt: nicht stumm (nur „true" ist stumm).
	if err := os.WriteFile(path, []byte("vielleicht\n"), 0o644); err != nil {
		t.Fatal(err)
	}
	if s.Muted() {
		t.Error("Datei-Inhalt „vielleicht“: Muted() = true, want false")
	}
}

// TestRateLimiter: Struktur-Semantik (Task 05 verdrahtet sie in den Loop).
func TestRateLimiter(t *testing.T) {
	// Unbegrenzt (Task-04-Default): Interval 0 erlaubt immer.
	unlimited := &RateLimiter{}
	for i := 0; i < 3; i++ {
		if !unlimited.Allow() {
			t.Fatal("Interval 0: Allow = false, want true (unbegrenzt)")
		}
		unlimited.Reserve()
	}

	// Begrenzt: nach Reserve ist erst nach Interval wieder Allow.
	l := &RateLimiter{Interval: 30 * time.Millisecond}
	if !l.Allow() {
		t.Fatal("frisches Limit: Allow = false")
	}
	l.Reserve()
	if l.Allow() {
		t.Fatal("sofort nach Reserve: Allow = true, want false")
	}
	if next := l.Next(); time.Now().After(next) {
		t.Errorf("Next() = %v liegt in der Vergangenheit", next)
	}
	time.Sleep(40 * time.Millisecond)
	if !l.Allow() {
		t.Fatal("nach Interval: Allow = false, want true")
	}
}