package core // 双层并发 + 扫尾 的集成测试:用 httptest 模拟 EH 服务器, // 记录每个时刻有多少请求在飞,验证「页并发」真的生效。 import ( "context" "fmt" "net/http" "net/http/httptest" "os" "path/filepath" "strconv" "strings" "sync" "sync/atomic" "testing" "time" "eh-downloader-wails/internal/eh" ) // mockEH 一个可控的 EH 模拟器 type mockEH struct { srv *httptest.Server mu sync.Mutex // 记录并发峰值 inFlight int32 peakFlight int32 totalReqs int64 // 每页失败次数(第一次失败 -> 成功,模拟可重试错误) failOnce map[int]int32 // 哪些页彻底失败 alwaysFail map[int]bool // 慢响应(ms),用于观察并发 delayMs int32 } func newMockEH(totalPages int) *mockEH { m := &mockEH{ failOnce: map[int]int32{}, alwaysFail: map[int]bool{}, delayMs: 5, } mux := http.NewServeMux() // 画廊页(一次性给全,避免分页逻辑干扰测试) mux.HandleFunc("/g/", func(w http.ResponseWriter, r *http.Request) { m.track(func() { var sb strings.Builder sb.WriteString(`
| artist: |
`, base, page)))
})
})
// 图片本体
mux.HandleFunc("/img/", func(w http.ResponseWriter, r *http.Request) {
m.track(func() {
page := 0
name := r.URL.Path[strings.LastIndex(r.URL.Path, "/")+1:]
fmt.Sscanf(name, "%03d.png", &page)
m.mu.Lock()
always := m.alwaysFail[page]
n := m.failOnce[page]
if !always && n > 0 {
m.failOnce[page] = n - 1
}
m.mu.Unlock()
if always {
w.WriteHeader(http.StatusServiceUnavailable)
return
}
if n > 0 {
w.WriteHeader(http.StatusServiceUnavailable)
return
}
w.Write([]byte("image-data-" + name))
})
})
m.srv = httptest.NewServer(mux)
return m
}
// track 统计并发峰值
func (m *mockEH) track(fn func()) {
cur := atomic.AddInt32(&m.inFlight, 1)
atomic.AddInt64(&m.totalReqs, 1)
for {
old := atomic.LoadInt32(&m.peakFlight)
if cur <= old || atomic.CompareAndSwapInt32(&m.peakFlight, old, cur) {
break
}
}
time.Sleep(time.Duration(m.delayMs) * time.Millisecond)
defer atomic.AddInt32(&m.inFlight, -1)
fn()
}
func (m *mockEH) URL() string { return m.srv.URL }
func (m *mockEH) Close() { m.srv.Close() }
func (m *mockEH) peak() int32 { return atomic.LoadInt32(&m.peakFlight) }
// 把 hath URL 指向 mock 服务器
func (m *mockEH) installBase() func() {
old := eh.EHBase
eh.EHBase = m.srv.URL
return func() { eh.EHBase = old }
}
func newEngineFor(t *testing.T, root string, g, p int) (*Engine, *Store) {
t.Helper()
s := NewStore(filepath.Join(t.TempDir(), "state.json"))
s.UpdateSettings(map[string]interface{}{
"downloadRoot": root,
"dirTemplate": "{gid}",
"galleryConcurrency": g,
"pageConcurrency": p,
"proxy": "",
"proxyMode": "off",
"timeout": 10000,
})
e := NewEngine(s, &noopSink{}, s.Settings())
return e, s
}
func addOne(t *testing.T, s *Store, gid, token string) {
t.Helper()
s.AddTasks([]eh.GalleryRef{{GID: gid, Token: token}})
}
// 页并发真的生效:单画廊 8 并发时,服务器侧应看到并发 > 1
func TestPageConcurrencyWorks(t *testing.T) {
const total = 20
m := newMockEH(total)
defer m.Close()
restore := m.installBase()
defer restore()
root := t.TempDir()
e, s := newEngineFor(t, root, 1, 8)
addOne(t, s, "999", "tok")
e.Enqueue(s.Tasks())
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
e.Start(ctx)
task := s.FindTask("999", "tok")
if task.Status != "done" {
t.Fatalf("应全部下完,实际状态 %s (done=%d failed=%d)", task.Status, task.Done, task.Failed)
}
if task.Done != total {
t.Errorf("应下 %d 张,实际 %d", total, task.Done)
}
// 页并发 8 => 峰值并发应该 > 1
if m.peak() < 2 {
t.Errorf("页并发 8 时服务器侧并发峰值只有 %d,说明页级并发没生效", m.peak())
}
t.Logf("页并发峰值 = %d(期望接近 8)", m.peak())
// 文件都要在
for i := 1; i <= total; i++ {
name := filepath.Join(root, "999", fmt.Sprintf("%03d.png", i))
if st, err := os.Stat(name); err != nil || st.Size() == 0 {
t.Errorf("第 %d 张图没下好", i)
}
}
// 清单也要写
mf, err := ReadManifest(filepath.Join(root, "999"))
if err != nil || mf == nil {
t.Fatalf("画廊清单没写: %v", err)
}
if mf.TotalPages != total || len(mf.DonePages) != total {
t.Errorf("清单内容不对: total=%d done=%d", mf.TotalPages, len(mf.DonePages))
}
}
// 页并发=1 时应退化成串行
func TestPageConcurrencyOneIsSerial(t *testing.T) {
const total = 8
m := newMockEH(total)
defer m.Close()
restore := m.installBase()
defer restore()
root := t.TempDir()
e, s := newEngineFor(t, root, 1, 1)
addOne(t, s, "888", "tok")
e.Enqueue(s.Tasks())
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
e.Start(ctx)
if s.FindTask("888", "tok").Status != "done" {
t.Error("应全部下完")
}
if m.peak() > 1 {
t.Errorf("页并发 1 时不该有并发请求,峰值 %d", m.peak())
}
}
// 重试:首轮失败(503)后应重试成功,不该进扫尾失败
func TestRetryThenSucceed(t *testing.T) {
const total = 4
m := newMockEH(total)
defer m.Close()
restore := m.installBase()
defer restore()
// 每页都先失败一次
m.mu.Lock()
for i := 1; i <= total; i++ {
m.failOnce[i] = 1
}
m.mu.Unlock()
root := t.TempDir()
e, s := newEngineFor(t, root, 1, 4)
addOne(t, s, "777", "tok")
e.Enqueue(s.Tasks())
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
defer cancel()
e.Start(ctx)
task := s.FindTask("777", "tok")
if task.Status != "done" {
t.Errorf("重试后应全部成功,实际 %s (failed=%d)", task.Status, task.Failed)
}
if task.Failed != 0 {
t.Errorf("不该有失败,实际 %d", task.Failed)
}
}
// 整轮扫尾:首轮必定失败(4 次重试全挂),扫尾再试仍失败 -> 标记 partial
// 但如果扫尾时恢复成功,应该是 done
func TestSweepRetrySucceeds(t *testing.T) {
const total = 3
m := newMockEH(total)
defer m.Close()
restore := m.installBase()
defer restore()
// 每页失败次数 = 首轮重试次数(4) + 1,确保首轮 4 次全挂,
// 扫尾时能成功
m.mu.Lock()
for i := 1; i <= total; i++ {
m.failOnce[i] = 5
}
m.mu.Unlock()
root := t.TempDir()
e, s := newEngineFor(t, root, 1, 3)
addOne(t, s, "666", "tok")
e.Enqueue(s.Tasks())
ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
defer cancel()
e.Start(ctx)
task := s.FindTask("666", "tok")
if task.Status != "done" {
t.Errorf("扫尾后应全部成功,实际 %s (done=%d failed=%d)", task.Status, task.Done, task.Failed)
}
if task.Failed != 0 {
t.Errorf("扫尾后不该有失败,实际 %d", task.Failed)
}
for i := 1; i <= total; i++ {
name := filepath.Join(root, "666", fmt.Sprintf("%03d.png", i))
if st, err := os.Stat(name); err != nil || st.Size() == 0 {
t.Errorf("第 %d 张没下好", i)
}
}
}
// 彻底失败(一直 503)-> partial,且扫尾也救不回来
func TestPermanentFailureMarksPartial(t *testing.T) {
const total = 2
m := newMockEH(total)
defer m.Close()
restore := m.installBase()
defer restore()
m.mu.Lock()
for i := 1; i <= total; i++ {
m.alwaysFail[i] = true
}
m.mu.Unlock()
root := t.TempDir()
e, s := newEngineFor(t, root, 1, 2)
addOne(t, s, "555", "tok")
e.Enqueue(s.Tasks())
ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
defer cancel()
e.Start(ctx)
task := s.FindTask("555", "tok")
if task.Status != "partial" {
t.Errorf("全失败应标记 partial,实际 %s", task.Status)
}
if task.Failed != total {
t.Errorf("失败数应为 %d,实际 %d", total, task.Failed)
}
if len(task.FailedList) != total {
t.Errorf("失败清单应有 %d 条,实际 %d", total, len(task.FailedList))
}
}
// 断点续传:已有文件不重下
func TestResumeWithExistingFiles(t *testing.T) {
const total = 6
m := newMockEH(total)
defer m.Close()
restore := m.installBase()
defer restore()
root := t.TempDir()
dir := filepath.Join(root, "444")
os.MkdirAll(dir, 0o755)
// 预先放好 4 张
for i := 1; i <= 4; i++ {
os.WriteFile(filepath.Join(dir, fmt.Sprintf("%03d.png", i)), []byte("image-data"), 0o644)
}
e, s := newEngineFor(t, root, 1, 4)
addOne(t, s, "444", "tok")
before := m.totalReqs
e.Enqueue(s.Tasks())
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
e.Start(ctx)
task := s.FindTask("444", "tok")
if task.Status != "done" {
t.Errorf("应下完,实际 %s (done=%d)", task.Status, task.Done)
}
if task.Done != total {
t.Errorf("完成数应为 %d,实际 %d", total, task.Done)
}
// 只该请求剩下的 2 页(每页 2 个请求 = 4 次)
used := m.totalReqs - before
if used > 8 {
t.Errorf("已有 4 张却发了 %d 次请求 —— 断点续传没生效", used)
}
t.Logf("续传只用了 %d 次请求(4 张已存在,应只用 4 次)", used)
}
// 多个画廊并行:画廊并发生效
func TestGalleryConcurrencyWorks(t *testing.T) {
const (
galleries = 5
pages = 4
)
m := newMockEH(pages)
defer m.Close()
restore := m.installBase()
defer restore()
root := t.TempDir()
e, s := newEngineFor(t, root, galleries, 1)
for i := 1; i <= galleries; i++ {
gid := strconv.Itoa(300 + i)
addOne(t, s, gid, "tok")
}
e.Enqueue(s.Tasks())
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
defer cancel()
e.Start(ctx)
for i := 1; i <= galleries; i++ {
gid := strconv.Itoa(300 + i)
task := s.FindTask(gid, "tok")
if task == nil || task.Status != "done" {
t.Errorf("画廊 %s 没下完: %v", gid, task)
}
}
// 画廊并发 5 => 峰值并发应该 > 1
if m.peak() < 2 {
t.Errorf("画廊并发 5 时峰值只有 %d,并发没生效", m.peak())
}
t.Logf("画廊并发峰值 = %d", m.peak())
}
// 全局限流在引擎里生效
func TestEngineRespectsRateLimit(t *testing.T) {
const total = 6
m := newMockEH(total)
defer m.Close()
restore := m.installBase()
defer restore()
root := t.TempDir()
e, s := newEngineFor(t, root, 1, 4)
e.SetRequestInterval(80) // 80ms
addOne(t, s, "333", "tok")
e.Enqueue(s.Tasks())
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
defer cancel()
start := time.Now()
e.Start(ctx)
elapsed := time.Since(start)
task := s.FindTask("333", "tok")
if task.Status != "done" {
t.Fatalf("应下完,实际 %s", task.Status)
}
// 6 页 = 6 次 /s/ + 6 次图片 = 12 次请求,至少 11 个间隔 = 880ms
wantMin := 11 * 80 * time.Millisecond
if elapsed < wantMin {
t.Errorf("限流 80ms 时 %d 次请求应至少耗时 %v,实际 %v", 12, wantMin, elapsed)
}
t.Logf("12 次请求限流 80ms,实际耗时 %v", elapsed)
}
// 取消:页级并发下点停止要能真的中断
func TestCancelWithPageConcurrency(t *testing.T) {
const total = 30
m := newMockEH(total)
defer m.Close()
m.delayMs = 50 // 放慢点,方便中途取消
restore := m.installBase()
defer restore()
root := t.TempDir()
e, s := newEngineFor(t, root, 1, 8)
addOne(t, s, "222", "tok")
e.Enqueue(s.Tasks())
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
done := make(chan struct{})
go func() { e.Start(ctx); close(done) }()
time.Sleep(300 * time.Millisecond)
e.Stop() // 相当于点「停止」
select {
case <-done:
case <-time.After(5 * time.Second):
t.Fatal("取消后应该在 5 秒内返回")
}
task := s.FindTask("222", "tok")
if task.Status != "queued" {
t.Errorf("取消后任务应退回 queued 等待续传,实际 %s", task.Status)
}
// 已下的文件还在(断点续传的基础)
dir := filepath.Join(root, "222")
files, _ := os.ReadDir(dir)
if len(files) > 0 {
t.Logf("取消前已下 %d 个文件,续传时会跳过", len(files))
}
}