package main import ( "errors" "strings" "sync" "testing" "time" ) // 测试用的「开一个新任务」。全局任务只有一份,用例之间必须清干净: // 上一个没收尾的话 jobStart 会拒绝覆盖(这是刻意的防串台),断言就会读到别人的数据 func startJobForTest(t *testing.T, kind, outName string, inputBytes int64, totalSec float64) { t.Helper() jobFinish("") if !jobStart(kind, outName, inputBytes, totalSec) { t.Fatalf("jobStart 失败:还残留着未结束的任务 %q", jobBusyName()) } t.Cleanup(func() { jobFinish("") }) } // 阶段权重加起来必须正好是 1,否则总进度永远到不了 100% func TestJobPhaseWeightsSumToOne(t *testing.T) { if len(jobPhaseNames) != len(jobPhaseWeights) { t.Fatalf("阶段名有 %d 个,权重有 %d 个,对不上", len(jobPhaseNames), len(jobPhaseWeights)) } sum := 0.0 for _, w := range jobPhaseWeights { sum += w } if sum < 0.999999 || sum > 1.000001 { t.Errorf("阶段权重之和应为 1,实际 %v", sum) } } // 总进度:走过的阶段按权重满算,当前阶段按比例折算 func TestJobPercentByPhase(t *testing.T) { startJobForTest(t, "测试", "x.mp4", 0, 100) if got := jobSnapshot().Percent; got != 0 { t.Errorf("刚开始时总进度应为 0,实际 %v", got) } // 走到渲染阶段但一帧没渲染:前面三个阶段(准备 / 分段拼接 / 头像)已满算 jobSetPhase(jobPhaseRender) want := (jobPhaseWeights[0] + jobPhaseWeights[1] + jobPhaseWeights[2]) * 100 if got := jobSnapshot().Percent; got < want-0.001 || got > want+0.001 { t.Errorf("进入渲染阶段时总进度应为 %v,实际 %v", want, got) } // 渲染一半 jobRender(50, 1.5, 25, 0) want += jobPhaseWeights[jobPhaseRender] * 50 if got := jobSnapshot().Percent; got < want-0.001 || got > want+0.001 { t.Errorf("渲染一半时总进度应为 %v,实际 %v", want, got) } // 结束(成功)后必须正好 100% jobFinish("") if got := jobSnapshot().Percent; got != 100 { t.Errorf("完成后总进度应为 100,实际 %v", got) } } // 阶段序号只往前走,重复设置同一阶段不会把进度打回 0 func TestJobPhaseOnlyMovesForward(t *testing.T) { startJobForTest(t, "测试", "x.mp4", 0, 100) defer jobFinish("") jobSetPhase(jobPhaseRender) jobRender(60, 1, 25, 0) before := jobSnapshot().Percent jobSetPhase(jobPhaseAvatar) // 往回设,应该被忽略 if got := jobSnapshot().Percent; got < before { t.Errorf("阶段倒退把进度打回去了:%v -> %v", before, got) } if idx := jobSnapshot().PhaseIdx; idx != jobPhaseRender { t.Errorf("阶段应为 %d,实际 %d", jobPhaseRender, idx) } } // 渲染阶段的 ETA = 剩余时长 ÷ 倍速 func TestJobEtaFromSpeed(t *testing.T) { startJobForTest(t, "测试", "x.mp4", 0, 200) defer jobFinish("") jobSetPhase(jobPhaseRender) jobRender(50, 2.0, 25, 0) // 还剩 150 秒,2 倍速 → 75 秒 if got := jobSnapshot().ETASec; got < 74.5 || got > 75.5 { t.Errorf("ETA 应为约 75 秒,实际 %v", got) } } // 预计输出大小:先给按码率估的值,渲染过 5% 之后换成按已写出字节外推 func TestJobEstOutSize(t *testing.T) { startJobForTest(t, "测试", "x.mp4", 0, 100) defer jobFinish("") jobSetPhase(jobPhaseRender) jobPlanSize(5000) info := jobSnapshot() if info.EstOutBytes != 5000 || info.EstSource != "bitrate" { t.Fatalf("还没开始渲染时应按码率估 5000,实际 %d(%s)", info.EstOutBytes, info.EstSource) } // 渲染到 50%,写出 4000 字节 → 外推整片 8000 jobRender(50, 1, 25, 4000) info = jobSnapshot() if info.EstSource != "measured" { t.Fatalf("渲染过 5%% 后应改用实测外推,实际来源 %s", info.EstSource) } if info.EstOutBytes != 8000 { t.Errorf("外推应为 8000,实际 %d", info.EstOutBytes) } // 比例太小的时候不急着外推,免得开头抖动 startJobForTest(t, "测试", "x.mp4", 0, 100) jobSetPhase(jobPhaseRender) jobPlanSize(5000) jobRender(1, 1, 25, 100) if got := jobSnapshot(); got.EstSource != "bitrate" { t.Errorf("渲染不到 5%% 时不该改用外推,实际来源 %s", got.EstSource) } } // 速度折线:最多保留固定点数,且按秒采样 func TestJobSamples(t *testing.T) { startJobForTest(t, "测试", "x.mp4", 0, 100) defer jobFinish("") jobSetPhase(jobPhaseRender) // 同一秒内连着报多次只留一个点 jobRender(1, 1.0, 25, 0) jobRender(2, 1.1, 25, 0) jobRender(3, 1.2, 25, 0) if n := len(jobSnapshot().Samples); n != 1 { t.Errorf("同一秒内的多次上报应只留 1 个采样点,实际 %d", n) } // 攒够 1 秒之后再报就会多一个点 job.Lock() job.lastSample = time.Now().Add(-2 * time.Second) job.Unlock() jobRender(4, 1.3, 25, 0) samples := jobSnapshot().Samples if len(samples) != 2 { t.Fatalf("应有两个采样点,实际 %d", len(samples)) } if samples[1].Speed != 1.3 { t.Errorf("第二个采样点的速度应为 1.3,实际 %v", samples[1].Speed) } } // 暂停 / 恢复 / 停止的状态流转;停止后闸门必须让上层拿到 errJobStopped func TestJobPauseStopGate(t *testing.T) { startJobForTest(t, "测试", "x.mp4", 0, 100) defer jobFinish("") if err := jobPause(); err != nil { t.Fatalf("暂停失败:%v", err) } if !jobSnapshot().Paused { t.Fatal("暂停后状态里应为 paused") } // 暂停期间闸门会一直等,放到 goroutine 里,再恢复让它出来 done := make(chan error, 1) go func() { done <- jobGate() }() select { case err := <-done: t.Fatalf("暂停期间闸门不该放行,却返回了 %v", err) case <-time.After(200 * time.Millisecond): } if err := jobResume(); err != nil { t.Fatalf("恢复失败:%v", err) } select { case err := <-done: if err != nil { t.Fatalf("恢复后闸门应放行,实际返回 %v", err) } case <-time.After(time.Second): t.Fatal("恢复后闸门没有放行") } if err := jobStop(); err != nil { t.Fatalf("停止失败:%v", err) } if err := jobGate(); !errors.Is(err, errJobStopped) { t.Errorf("停止后闸门应返回 errJobStopped,实际 %v", err) } // 停止不是「失败」,界面要能区分开 jobFinish(errJobStopped.Error()) info := jobSnapshot() if info.Active { t.Error("结束后不该还是 active") } if info.Message != "已停止" || info.Err != errJobStopped.Error() { t.Errorf("停止后的提示应为「已停止」,实际 message=%q err=%q", info.Message, info.Err) } if jobErrText(nil) != "" || jobErrText(errNoDanmu) != "" { t.Error("正常结束和「没有弹幕」都不该算失败") } if jobErrText(errJobStopped) != errJobStopped.Error() { t.Error("停止的原因要原样传给界面") } if jobErrText(errors.New("炸了")) != "炸了" { t.Error("其它错误要原样传给界面") } } // 没有任务时控制接口要明确报错,而不是静默成功 func TestJobControlWithoutTask(t *testing.T) { jobFinish("") if err := jobPause(); err == nil { t.Error("没有任务时暂停应该报错") } if err := jobPause(); err == nil { t.Error("没有任务时恢复应该报错") } if err := jobStop(); err == nil { t.Error("没有任务时停止应该报错") } } // ffmpeg 的 speed 字段形如 "1.37x",暂停或还没开始时是 "N/A" func TestParseFFmpegSpeed(t *testing.T) { cases := map[string]float64{ "1.37x": 1.37, "2x": 2, "N/A": 0, "": 0, "-1x": 0, } for in, want := range cases { if got := parseFFmpegSpeed(in); got != want { t.Errorf("parseFFmpegSpeed(%q) 应为 %v,实际 %v", in, want, got) } } } // 解析 -progress pipe:1 的输出,只在 progress= 行回调一次 func TestScanFFmpegProgress(t *testing.T) { // out_time_us 和 out_time_ms 都是微秒(ffmpeg 的历史遗留) text := strings.Join([]string{ "frame=100", "fps=25.0", "total_size=1048576", "out_time_us=4000000", "out_time_ms=4000000", "out_time=00:00:04.000000", "speed=1.37x", "progress=continue", "frame=200", "out_time_us=8000000", "total_size=2097152", "speed=N/A", "progress=end", }, "\n") type call struct { sec, speed, fps float64 size int64 } var got []call scanFFmpegProgress(strings.NewReader(text), func(sec, speed, fps float64, size int64) { got = append(got, call{sec, speed, fps, size}) }) if len(got) != 2 { t.Fatalf("应该回调 2 次,实际 %d 次:%+v", len(got), got) } if got[0].sec != 4 || got[0].speed != 1.37 || got[0].fps != 25 || got[0].size != 1048576 { t.Errorf("第一次回调不对:%+v", got[0]) } if got[1].sec != 8 || got[1].size != 2097152 { t.Errorf("第二次回调不对:%+v", got[1]) } // 暂停时 speed 会是 N/A,这时不该把上一次的倍速顶掉 if got[1].speed != 0 { t.Errorf("N/A 应该解析成 0,实际 %v", got[1].speed) } } // 混着无关行和坏数据也不能崩 func TestScanFFmpegProgressTolerant(t *testing.T) { text := "乱七八糟\nspeed=abc\nprogress=continue\nout_time_us=不是数字\nprogress=end\n" n := 0 scanFFmpegProgress(strings.NewReader(text), func(sec, speed, fps float64, size int64) { n++ if sec != 0 || speed != 0 { t.Errorf("坏数据不该解析出值:sec=%v speed=%v", sec, speed) } }) if n != 2 { t.Errorf("progress= 行都应回调,期望 2 次,实际 %d", n) } } // 截图式地看一眼任务快照的字段都填上了 func TestJobSnapshotFields(t *testing.T) { startJobForTest(t, "手动合并", "成片.mp4", 123456, 3600) defer jobFinish("") jobSetPhase(jobPhaseAvatar) jobAvatars(3, 10) jobSetTotal(3600) info := jobSnapshot() if !info.Active || info.Kind != "手动合并" || info.OutName != "成片.mp4" { t.Fatalf("基本信息不对:%+v", info) } if info.InputBytes != 123456 || info.TotalSec != 3600 { t.Errorf("输入大小 / 总时长不对:%+v", info) } if info.AvatarDone != 3 || info.AvatarTotal != 10 { t.Errorf("头像进度不对:%+v", info) } if len(info.Phases) != len(jobPhaseNames) { t.Errorf("阶段列表长度应为 %d,实际 %d", len(jobPhaseNames), len(info.Phases)) } if info.PhasePercent < 29.9 || info.PhasePercent > 30.1 { t.Errorf("头像阶段进度应为 30%%,实际 %v", info.PhasePercent) } if !info.Pausable { t.Error("任务在跑的时候应该可暂停") } } // ==================== 任务串行化 ==================== // 全局任务状态只有一份:两个合并同时跑的话,后启动的会把前一个的进度和输出名清掉, // 点「停止」还会杀掉别家任务的 ffmpeg。下面几条例外情况锁住改串行之后的语义。 // jobStart 在已经有任务在跑时必须拒绝,并且一个字段都不能动 func TestJobStartRejectsWhenBusy(t *testing.T) { startJobForTest(t, "自动合并", "甲_弹幕合并.mp4", 1000, 100) jobSetPhase(jobPhaseRender) jobRender(30, 1.2, 25, 5000) if ok := jobStart("手动合并", "乙_弹幕合并.mp4", 2000, 200); ok { t.Fatal("已经有任务在跑,jobStart 应当返回 false") } info := jobSnapshot() if info.Kind != "自动合并" || info.OutName != "甲_弹幕合并.mp4" { t.Errorf("被拒之后不该动别人的任务信息:%+v", info) } if info.InputBytes != 1000 || info.TotalSec != 100 { t.Errorf("被拒之后输入大小和总时长也不该变:%+v", info) } if info.Percent < 25 || info.Percent > 45 { t.Errorf("被拒之后进度必须还是原任务自己的(25~45),实际 %v", info.Percent) } if !info.Active { t.Error("拒绝新任务不代表把老任务结束掉") } } // 任务位被占着时手动合并要立刻知道(不能让页面干等),抢不到只能有一个赢 func TestMergeSlotTryLockExclusive(t *testing.T) { if !tryTakeMergeSlot() { t.Fatal("空闲时手动合并应该抢得到任务位") } if tryTakeMergeSlot() { t.Fatal("任务位已经被占着,第二次 TryLock 不该成功") } releaseMergeSlot() if !tryTakeMergeSlot() { t.Fatal("释放之后应该又能抢到") } releaseMergeSlot() } // 自动合并(十几个主播同时下播很常见)必须排队而不是并发 func TestMergeSlotSerializes(t *testing.T) { if !tryTakeMergeSlot() { t.Fatal("前置条件:任务位本该是空的") } const n = 4 order := make([]int, 0, n) var mu sync.Mutex done := make(chan struct{}, n) for i := 0; i < n; i++ { idx := i go func() { defer func() { done <- struct{}{} }() // 和真实链路一致:自动合并会一直等到轮到自己 _ = runMergeExclusive(func() error { mu.Lock() order = append(order, idx) mu.Unlock() time.Sleep(20 * time.Millisecond) return nil }) }() } time.Sleep(60 * time.Millisecond) // 让它们先堆在门外 releaseMergeSlot() for i := 0; i < n; i++ { select { case <-done: case <-time.After(5 * time.Second): t.Fatalf("第 %d 个任务没跑完,队列卡住了", i) } } if len(order) != n { t.Fatalf("应当 %d 个任务都跑完,实际 %d 个", n, len(order)) } } // 任务内部 panic 时也必须释放任务位、并且能翻成错误 // —— 否则卡片会永远停在「进行中」,队列也被锁死 func TestMergeExclusiveReleasesOnPanic(t *testing.T) { err := runMergeExclusive(func() error { panic("模拟内部出错") }) if err == nil { t.Fatal("内部 panic 应当被翻成错误返回") } if !strings.Contains(err.Error(), "模拟内部出错") { t.Errorf("错误信息里应带上原因,实际:%v", err) } if !tryTakeMergeSlot() { t.Fatal("panic 之后任务位必须已经释放") } releaseMergeSlot() }