vidarchive

vidarchive/internal/worker/pool_test.go
pool_test.go
⎇
Raw
1package worker
2
3import (
4 "os"
5 "path/filepath"
6 "testing"
7 "time"
8
9 "vidarchive/internal/config"
10 "vidarchive/internal/database"
11 "vidarchive/internal/models"
12 "vidarchive/internal/repository"
13 "vidarchive/internal/service"
14)
15
16// newTestPool builds a pool over a real service and database, with yt-dlp
17// replaced by a script that reports it has started and then blocks. scratch is
18// where that script writes its marker.
19func newTestPool(t *testing.T) (pool *Pool, repo *repository.DownloadRepository, scratch string) {
20 t.Helper()
21
22 root := t.TempDir()
23 cfg := &config.Config{
24 DBPath: ":memory:",
25 LibraryDir: filepath.Join(root, "library"),
26 TempDir: filepath.Join(root, "temp"),
27 FFmpegPath: "ffmpeg",
28 FFprobePath: "ffprobe",
29 }
30 for _, dir := range []string{cfg.LibraryDir, cfg.TempDir} {
31 if err := os.MkdirAll(dir, 0o755); err != nil {
32 t.Fatal(err)
33 }
34 }
35
36 // The script must not exit on its own: the point of these tests is that Stop
37 // interrupts a download rather than waiting one out.
38 cfg.YTDLPPath = filepath.Join(root, "yt-dlp")
39 script := "#!/bin/sh\ntouch " + filepath.Join(root, "started") + "\nsleep 300\n"
40 if err := os.WriteFile(cfg.YTDLPPath, []byte(script), 0o755); err != nil {
41 t.Fatal(err)
42 }
43
44 db, err := database.New(cfg)
45 if err != nil {
46 t.Fatalf("init db: %v", err)
47 }
48 t.Cleanup(func() { db.Close() })
49
50 repo = repository.NewDownloadRepository(db)
51 downloadSvc := service.NewDownloadService(
52 repo,
53 service.NewLibraryService(cfg.LibraryDir, cfg.FFmpegPath, cfg.FFprobePath),
54 service.NewPresetService(repository.NewPresetRepository(db)),
55 service.NewSettingsService(repository.NewSettingsRepository(db)),
56 service.NewSubscriptionService(repository.NewSubscriptionRepository(db), cfg),
57 cfg,
58 )
59
60 return New(downloadSvc, 2), repo, root
61}
62
63func waitForFile(t *testing.T, path string) {
64 t.Helper()
65 deadline := time.Now().Add(10 * time.Second)
66 for time.Now().Before(deadline) {
67 if _, err := os.Stat(path); err == nil {
68 return
69 }
70 time.Sleep(5 * time.Millisecond)
71 }
72 t.Fatalf("timed out waiting for %s", path)
73}
74
75// Stop must return while a download is still running. If it waited for yt-dlp
76// to finish on its own, a shutdown would hang for as long as the download takes.
77func TestStopReturnsWhileDownloadRunning(t *testing.T) {
78 pool, repo, scratch := newTestPool(t)
79
80 d := &models.Download{URL: "https://example.com/v", Status: "queued"}
81 if err := repo.Create(d); err != nil {
82 t.Fatal(err)
83 }
84
85 pool.Start()
86 pool.Submit(d)
87 waitForFile(t, filepath.Join(scratch, "started"))
88
89 stopped := make(chan struct{})
90 go func() {
91 pool.Stop()
92 close(stopped)
93 }()
94
95 select {
96 case <-stopped:
97 case <-time.After(30 * time.Second):
98 t.Fatal("Stop did not return while a download was running")
99 }
100
101 // The row stays "downloading" after a shutdown so the next start resumes it.
102 got, err := repo.GetByID(d.ID)
103 if err != nil {
104 t.Fatal(err)
105 }
106 if got.Status != "downloading" {
107 t.Errorf("status = %q, want downloading", got.Status)
108 }
109}
110
111// Stop is called from the shutdown path and may race with an already-stopped
112// pool; calling it twice must not panic or block.
113func TestStopIsIdempotent(t *testing.T) {
114 pool, _, _ := newTestPool(t)
115 pool.Start()
116
117 done := make(chan struct{})
118 go func() {
119 pool.Stop()
120 pool.Stop()
121 close(done)
122 }()
123
124 select {
125 case <-done:
126 case <-time.After(30 * time.Second):
127 t.Fatal("second Stop blocked")
128 }
129}
130
131// A download submitted after shutdown must be dropped rather than sent to a
132// channel nobody is reading.
133func TestSubmitAfterStopIsNoOp(t *testing.T) {
134 pool, repo, _ := newTestPool(t)
135 pool.Start()
136 pool.Stop()
137
138 d := &models.Download{URL: "https://example.com/v", Status: "queued"}
139 if err := repo.Create(d); err != nil {
140 t.Fatal(err)
141 }
142
143 done := make(chan struct{})
144 go func() {
145 pool.Submit(d)
146 close(done)
147 }()
148
149 select {
150 case <-done:
151 case <-time.After(10 * time.Second):
152 t.Fatal("Submit blocked after Stop")
153 }
154 if len(pool.queue) != 0 {
155 t.Errorf("queue holds %d items, want 0 after Stop", len(pool.queue))
156 }
157}
158
159// Submit is called from an HTTP handler, so a full buffer must never block it:
160// the row stays queued in the database and the queue checker picks it up later.
161func TestSubmitDoesNotBlockWhenBufferFull(t *testing.T) {
162 pool, _, _ := newTestPool(t)
163 // Deliberately no Start: nothing drains the buffer.
164
165 done := make(chan struct{})
166 go func() {
167 for i := 0; i < cap(pool.queue)*2; i++ {
168 pool.Submit(&models.Download{ID: int64(i + 1), URL: "https://example.com/v"})
169 }
170 close(done)
171 }()
172
173 select {
174 case <-done:
175 case <-time.After(10 * time.Second):
176 t.Fatal("Submit blocked on a full buffer")
177 }
178 if len(pool.queue) != cap(pool.queue) {
179 t.Errorf("queue holds %d items, want the buffer to be full at %d", len(pool.queue), cap(pool.queue))
180 }
181}
182