acfunlive-remake/progress_test.go

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()
}