// Tests for `fluxrec serve`: the poll loop against an httptest fake // Miniflux (multi-page, mid-run crash), batch assembly split/kinds, and // the refresh-time computation including the midnight wrap. Helpers // (testEntry, makeEntries, entriesJSON) come from export_test.go. package cmds import ( "encoding/json" "fmt" "math/rand" "net/http" "net/http/httptest" "os" "path/filepath" "strconv" "strings" "testing" "time" "fluxrec/core" "fluxrec/miniflux" ) // testModel is a minimal hand-built model: one unigram ("title"), so every // makeEntries title scores identically and batch ties break by entry ID. func testModel() *core.ModelEnvelope { return &core.ModelEnvelope{ CreatedAt: time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC), Meta: map[string]any{}, Vectorizer: map[string]float64{"title": 1.0}, OrderedVocab: []string{"title"}, Weights: []float64{2.0, 0.1}, } } // idServer pages its entry pool by after_entry_id ascending, like the // serve poll expects, and can be told to die on the Nth request. type idServer struct { entries []testEntry // ascending IDs queries []string calls int failOn int // 1-based request number to 500 on; 0 = never } func (s *idServer) handler() http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { s.calls++ s.queries = append(s.queries, r.URL.RawQuery) if s.failOn == s.calls { http.Error(w, "boom", http.StatusInternalServerError) return } q := r.URL.Query() after, _ := strconv.ParseInt(q.Get("after_entry_id"), 10, 64) limit, _ := strconv.Atoi(q.Get("limit")) if limit <= 0 { limit = len(s.entries) } var page []testEntry for _, e := range s.entries { if e.ID > after { page = append(page, e) } } if len(page) > limit { page = page[:limit] } w.Header().Set("Content-Type", "application/json") fmt.Fprint(w, entriesJSON(len(page), page)) } } func serveTestClient(t *testing.T, s http.Handler) *miniflux.Client { t.Helper() srv := httptest.NewServer(s) t.Cleanup(srv.Close) return miniflux.NewClientWithHTTP(miniflux.Config{BaseURL: srv.URL, Token: "t"}, srv.Client()) } func TestPollMultiPage(t *testing.T) { // 450 entries => three 200/200/50 pages off one ascending cursor. fake := &idServer{entries: makeEntries(450, 1, false, 1)} client := serveTestClient(t, fake.handler()) dir := t.TempDir() model := testModel() cmd := &ServeCommand{ Cursor: filepath.Join(dir, "cursor.json"), Runs: filepath.Join(dir, "runs.jsonl"), TopN: 15, ExploreFrac: 0.2, } var out strings.Builder rng := rand.New(rand.NewSource(1)) if err := cmd.pollOnce(client, core.CreateVectorizerFromModel(model), model, rng, &out); err != nil { t.Fatalf("pollOnce: %v", err) } if fake.calls != 3 { t.Fatalf("expected 3 paged requests, got %d", fake.calls) } if !strings.Contains(fake.queries[0], "direction=asc") || !strings.Contains(fake.queries[0], "limit=200") { t.Errorf("first query missing direction/limit: %s", fake.queries[0]) } if strings.Contains(fake.queries[0], "after_entry_id") { t.Errorf("first query should have no cursor: %s", fake.queries[0]) } if !strings.Contains(fake.queries[1], "after_entry_id=200") || !strings.Contains(fake.queries[2], "after_entry_id=400") { t.Errorf("cursor paging wrong: %q", fake.queries) } cursor, err := loadCursor(cmd.Cursor) if err != nil { t.Fatalf("loadCursor: %v", err) } if cursor != 450 { t.Errorf("cursor = %d, want 450", cursor) } rows, err := core.ReadRuns(cmd.Runs) if err != nil { t.Fatalf("ReadRuns: %v", err) } if len(rows) != 1 { t.Fatalf("expected 1 run row, got %d", len(rows)) } row := rows[0] if row.EntriesSeen != 450 { t.Errorf("entries_seen = %d, want 450", row.EntriesSeen) } if len(row.Kept) != 15 { t.Errorf("kept = %d, want 15", len(row.Kept)) } if !row.ModelCreatedAt.Equal(model.CreatedAt) { t.Errorf("model_created_at = %v, want %v", row.ModelCreatedAt, model.CreatedAt) } if row.ScoreMedian == nil { t.Error("score_median missing on run row (median over all scored entries)") } } func TestPollCrashLeavesNoState(t *testing.T) { fake := &idServer{entries: makeEntries(450, 1, false, 1), failOn: 2} client := serveTestClient(t, fake.handler()) dir := t.TempDir() model := testModel() cmd := &ServeCommand{ Cursor: filepath.Join(dir, "cursor.json"), Runs: filepath.Join(dir, "runs.jsonl"), TopN: 15, ExploreFrac: 0.2, } err := cmd.pollOnce(client, core.CreateVectorizerFromModel(model), model, rand.New(rand.NewSource(1)), &strings.Builder{}) if err == nil { t.Fatal("expected pollOnce to fail on the second page") } // Crash-safe ordering: the page-1 work must not be half-persisted. if _, statErr := os.Stat(cmd.Cursor); !os.IsNotExist(statErr) { t.Error("cursor file exists despite the failed poll") } if _, statErr := os.Stat(cmd.Runs); !os.IsNotExist(statErr) { t.Error("runs file exists despite the failed poll") } } func TestPollEmptyWritesNothing(t *testing.T) { fake := &idServer{} client := serveTestClient(t, fake.handler()) dir := t.TempDir() model := testModel() cmd := &ServeCommand{ Cursor: filepath.Join(dir, "cursor.json"), Runs: filepath.Join(dir, "runs.jsonl"), TopN: 15, ExploreFrac: 0.2, } if err := cmd.pollOnce(client, core.CreateVectorizerFromModel(model), model, rand.New(rand.NewSource(1)), &strings.Builder{}); err != nil { t.Fatalf("pollOnce: %v", err) } if rows, _ := core.ReadRuns(cmd.Runs); len(rows) != 0 { t.Errorf("empty poll wrote %d rows, want 0 (RSS tail starvation)", len(rows)) } } func TestPollSkipsRecFeed(t *testing.T) { fake := &idServer{ entries: append(makeEntries(10, 1, false, 1), makeEntries(5, 100, false, 99)...), } client := serveTestClient(t, fake.handler()) dir := t.TempDir() model := testModel() cmd := &ServeCommand{ Cursor: filepath.Join(dir, "cursor.json"), Runs: filepath.Join(dir, "runs.jsonl"), TopN: 30, ExploreFrac: 0, RecFeedID: 99, } if err := cmd.pollOnce(client, core.CreateVectorizerFromModel(model), model, rand.New(rand.NewSource(1)), &strings.Builder{}); err != nil { t.Fatalf("pollOnce: %v", err) } rows, _ := core.ReadRuns(cmd.Runs) if rows[0].EntriesSeen != 15 || rows[0].RecFeedSkipped != 5 { t.Errorf("seen=%d skipped=%d, want 15 / 5", rows[0].EntriesSeen, rows[0].RecFeedSkipped) } for _, k := range rows[0].Kept { if k.EntryID >= 100 { t.Errorf("rec-feed entry %d was kept; serve must not re-recommend its own output", k.EntryID) } } } func TestAssembleBatchSplitAndKinds(t *testing.T) { // 30 candidates, scores increasing with ID => deterministic ranks. entries := make([]scoredEntry, 30) for i := range entries { entries[i] = scoredEntry{ entry: miniflux.Entry{ID: int64(i + 1), Title: fmt.Sprintf("T%d", i+1)}, score: float64(i + 1), } } kept := assembleBatch(entries, 15, 0.2, rand.New(rand.NewSource(7))) if len(kept) != 15 { t.Fatalf("kept = %d, want 15", len(kept)) } counts := map[string]int{} var rankedIDs, runnerUpIDs []int64 random := map[int64]bool{} for _, k := range kept { counts[k.Kind]++ switch k.Kind { case core.KindRanked: rankedIDs = append(rankedIDs, k.EntryID) case core.KindRunnerUp: runnerUpIDs = append(runnerUpIDs, k.EntryID) case core.KindRandom: random[k.EntryID] = true } } if counts[core.KindRanked] != 12 || counts[core.KindRunnerUp] != 1 || counts[core.KindRandom] != 2 { t.Errorf("split = %v, want ranked=12 runnerup=1 random=2", counts) } // Ranked are the top 12 by score, in score order. for i, id := range rankedIDs { want := int64(30 - i) if id != want { t.Fatalf("ranked[%d] = %d, want %d", i, id, want) } } // The runner-up is the highest-scored entry below the cut. if len(runnerUpIDs) != 1 || runnerUpIDs[0] != 18 { t.Errorf("runner-up = %v, want [18]", runnerUpIDs) } // Random picks come from the tail below the runner-up, no duplicates. for id := range random { if id > 17 { t.Errorf("random pick %d not below the runner-up cut", id) } } } func TestAssembleBatchFewCandidates(t *testing.T) { entries := []scoredEntry{ {entry: miniflux.Entry{ID: 1}, score: 0.1}, {entry: miniflux.Entry{ID: 2}, score: 0.9}, {entry: miniflux.Entry{ID: 3}, score: 0.5}, } kept := assembleBatch(entries, 15, 0.2, rand.New(rand.NewSource(1))) if len(kept) != 3 { t.Fatalf("kept = %d, want all 3 as ranked", len(kept)) } for _, k := range kept { if k.Kind != core.KindRanked { t.Errorf("kind = %s, want ranked when candidates < cut", k.Kind) } } if kept[0].EntryID != 2 { // highest score first t.Errorf("first kept ID = %d, want 2 (highest score)", kept[0].EntryID) } } func TestMedianScore(t *testing.T) { if m := medianScore(nil); m != nil { t.Errorf("empty poll: median = %v, want nil (field omitted)", *m) } odd := []scoredEntry{{score: 0.1}, {score: 0.9}, {score: 0.5}} if m := medianScore(odd); m == nil || *m != 0.5 { t.Errorf("odd median = %v, want 0.5", m) } even := []scoredEntry{{score: 0.1}, {score: 0.5}, {score: 0.9}, {score: 0.2}} if m := medianScore(even); m == nil || *m != 0.35 { t.Errorf("even median = %v, want 0.35", m) } } func TestParseRefreshAt(t *testing.T) { mins, err := parseRefreshAt("18:00,06:00, 12:30") if err != nil { t.Fatalf("parseRefreshAt: %v", err) } want := []int{360, 750, 1080} if fmt.Sprint(mins) != fmt.Sprint(want) { t.Errorf("mins = %v, want %v (sorted)", mins, want) } for _, bad := range []string{"24:00", "abc", "12", "12:60", ""} { if _, err := parseRefreshAt(bad); err == nil { t.Errorf("parseRefreshAt(%q) unexpectedly succeeded", bad) } } } func TestNextRefresh(t *testing.T) { mins := []int{360, 720, 1080} // 06:00, 12:00, 18:00 at := func(day, h, m int) time.Time { return time.Date(2024, 3, day, h, m, 0, 0, time.UTC) } cases := []struct { now time.Time want time.Time }{ {at(1, 13, 0), at(1, 18, 0)}, // same-day next slot {at(1, 6, 0), at(1, 12, 0)}, // exactly on a slot: strictly after {at(1, 23, 30), at(2, 6, 0)}, // past last slot: midnight wrap {at(1, 0, 0), at(1, 6, 0)}, // before all slots } for _, tc := range cases { if got := nextRefresh(tc.now, mins); !got.Equal(tc.want) { t.Errorf("nextRefresh(%v) = %v, want %v", tc.now, got, tc.want) } } } func TestMinGapMinutes(t *testing.T) { if got := minGapMinutes([]int{360, 720, 1080}); got != 360 { t.Errorf("minGap = %d, want 360", got) } if got := minGapMinutes([]int{1380, 60}); got != 120 { t.Errorf("minGap across midnight = %d, want 120", got) } if got := minGapMinutes([]int{360}); got != 1440 { t.Errorf("single slot minGap = %d, want 1440", got) } } func TestCatchUpDue(t *testing.T) { mins := []int{360, 720, 1080} // min gap 6h => half = 3h now := time.Date(2024, 3, 2, 12, 0, 0, 0, time.UTC) dir := t.TempDir() runs := filepath.Join(dir, "runs.jsonl") // No log yet: first boot polls immediately. if due, err := catchUpDue(runs, mins, now); err != nil || !due { t.Errorf("no runs: due=%v err=%v, want true", due, err) } recent := core.RunRow{RanAt: now.Add(-time.Hour), EntriesSeen: 3} if err := core.AppendRun(runs, recent); err != nil { t.Fatalf("AppendRun: %v", err) } if due, _ := catchUpDue(runs, mins, now); due { t.Error("last run 1h ago: due=true, want false (next slot will cover it)") } old := core.RunRow{RanAt: now.Add(-5 * time.Hour), EntriesSeen: 3} if err := core.AppendRun(runs, old); err != nil { t.Fatalf("AppendRun: %v", err) } if due, _ := catchUpDue(runs, mins, now); !due { t.Error("last run 5h ago: due=false, want true (downtime across slots)") } } func TestServeRoutes(t *testing.T) { dir := t.TempDir() runsPath := filepath.Join(dir, "runs.jsonl") row := core.RunRow{ RanAt: time.Date(2024, 3, 1, 6, 0, 0, 0, time.UTC), EntriesSeen: 42, Kept: []core.KeptItem{ {EntryID: 101, URL: "https://example.com/a", Title: "Alpha", FeedTitle: "Feed A", PublishedAt: time.Date(2024, 3, 1, 5, 0, 0, 0, time.UTC), Content: "body a", Score: 0.9, Kind: core.KindRanked}, {EntryID: 102, URL: "https://example.com/b", Title: "Beta", FeedTitle: "Feed B", PublishedAt: time.Date(2024, 3, 1, 5, 30, 0, 0, time.UTC), Content: "body b", Score: 0.2, Kind: core.KindRandom}, }, } if err := core.AppendRun(runsPath, row); err != nil { t.Fatalf("AppendRun: %v", err) } cmd := &ServeCommand{Runs: runsPath, RSSBatches: 21} pollCalls := 0 pollStub := func() error { pollCalls++; return nil } handler := cmd.routes([]int{360, 720, 1080}, time.Now().Add(-48*time.Hour), pollStub) rec := httptest.NewRecorder() handler.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/recommendations.xml", nil)) if rec.Code != http.StatusOK { t.Fatalf("RSS status = %d", rec.Code) } if ct := rec.Header().Get("Content-Type"); !strings.Contains(ct, "application/rss+xml") { t.Errorf("RSS content-type = %q", ct) } body := rec.Body.String() if !strings.Contains(body, "mf:101") { t.Error("RSS missing mf:101 guid") } if strings.Count(body, "") != 2 { t.Errorf("RSS item count = %d, want 2", strings.Count(body, "")) } rec = httptest.NewRecorder() handler.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/api/status", nil)) if rec.Code != http.StatusOK { t.Fatalf("status = %d", rec.Code) } var status map[string]any if err := json.Unmarshal(rec.Body.Bytes(), &status); err != nil { t.Fatalf("status not JSON: %v", err) } last, ok := status["last_run"].(map[string]any) if !ok { t.Fatalf("status last_run missing: %v", status) } if last["entries_seen"].(float64) != 42 || last["kept"].(float64) != 2 { t.Errorf("last_run = %v, want entries_seen=42 kept=2", last) } if status["model_age_hours"].(float64) <= 0 { t.Errorf("model_age_hours = %v, want positive", status["model_age_hours"]) } } func TestManualPollRoute(t *testing.T) { dir := t.TempDir() runsPath := filepath.Join(dir, "runs.jsonl") // Last run hours ago: well outside the manual-poll cooldown. if err := core.AppendRun(runsPath, core.RunRow{ RanAt: time.Now().Add(-time.Hour), EntriesSeen: 3, }); err != nil { t.Fatalf("AppendRun: %v", err) } pollCalls := 0 pollStub := func() error { pollCalls++; return nil } cmd := &ServeCommand{Runs: runsPath} handler := cmd.routes([]int{360, 720, 1080}, time.Now(), pollStub) // GET is rejected; only POST triggers. rec := httptest.NewRecorder() handler.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/api/poll", nil)) if rec.Code != http.StatusMethodNotAllowed { t.Errorf("GET /api/poll = %d, want 405", rec.Code) } // POST runs the poll; the stub wrote no row, so no new run is reported. rec = httptest.NewRecorder() handler.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/api/poll", nil)) if rec.Code != http.StatusOK { t.Fatalf("POST /api/poll = %d, want 200", rec.Code) } if pollCalls != 1 { t.Errorf("poll called %d times, want 1", pollCalls) } var resp map[string]any if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { t.Fatalf("response not JSON: %v", err) } if resp["polled"] != true || resp["new_run"] != false { t.Errorf("response = %v, want polled=true new_run=false", resp) } // A fresh run row puts the next POST inside the cooldown window. if err := core.AppendRun(runsPath, core.RunRow{RanAt: time.Now(), EntriesSeen: 5}); err != nil { t.Fatalf("AppendRun: %v", err) } rec = httptest.NewRecorder() handler.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/api/poll", nil)) if rec.Code != http.StatusTooManyRequests { t.Errorf("POST inside cooldown = %d, want 429", rec.Code) } if pollCalls != 1 { t.Errorf("poll called %d times, want still 1 (cooldown)", pollCalls) } }