679 lines
22 KiB
Go
679 lines
22 KiB
Go
package worker
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"io"
|
|
"log"
|
|
"os"
|
|
"path/filepath"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"gitea.kazu634.com/kazu634/tver/internal/media"
|
|
"gitea.kazu634.com/kazu634/tver/internal/store"
|
|
"github.com/jackc/pgx/v5"
|
|
)
|
|
|
|
type fakeStore struct {
|
|
mu sync.Mutex
|
|
job store.Job
|
|
progress []store.Progress
|
|
heartbeats int
|
|
completed bool
|
|
completePath string
|
|
completeSize int64
|
|
failedKind, failedMessage string
|
|
requeued bool
|
|
claimed bool
|
|
}
|
|
|
|
type queueStore struct {
|
|
mu sync.Mutex
|
|
jobs []store.Job
|
|
claimCounts map[int64]int
|
|
active int
|
|
maxActive int
|
|
}
|
|
|
|
func (f *queueStore) ClaimNext(context.Context) (store.Job, error) {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
for i := range f.jobs {
|
|
if f.jobs[i].Status != "queued" {
|
|
continue
|
|
}
|
|
f.jobs[i].Status = "running"
|
|
f.claimCounts[f.jobs[i].ID]++
|
|
f.active++
|
|
if f.active > f.maxActive {
|
|
f.maxActive = f.active
|
|
}
|
|
return f.jobs[i], nil
|
|
}
|
|
return store.Job{}, pgx.ErrNoRows
|
|
}
|
|
|
|
func (f *queueStore) UpdateProgress(_ context.Context, id int64, _ store.Progress) error {
|
|
return f.requireRunning(id)
|
|
}
|
|
|
|
func (f *queueStore) Heartbeat(_ context.Context, id int64) error {
|
|
return f.requireRunning(id)
|
|
}
|
|
|
|
func (f *queueStore) requireRunning(id int64) error {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
for i := range f.jobs {
|
|
if f.jobs[i].ID == id && f.jobs[i].Status == "running" {
|
|
return nil
|
|
}
|
|
}
|
|
return store.ErrInvalidTransition
|
|
}
|
|
|
|
func (f *queueStore) GetJob(_ context.Context, id int64) (store.Job, error) {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
for _, job := range f.jobs {
|
|
if job.ID == id {
|
|
return job, nil
|
|
}
|
|
}
|
|
return store.Job{}, pgx.ErrNoRows
|
|
}
|
|
|
|
func (f *queueStore) Complete(_ context.Context, id int64, _ string, _ int64) error {
|
|
return f.finish(id, "succeeded")
|
|
}
|
|
|
|
func (f *queueStore) Fail(_ context.Context, id int64, _, _ string) error {
|
|
return f.finish(id, "failed")
|
|
}
|
|
|
|
func (f *queueStore) RequeueRunning(_ context.Context, id int64) error {
|
|
return f.finish(id, "queued")
|
|
}
|
|
|
|
func (f *queueStore) finish(id int64, status string) error {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
for i := range f.jobs {
|
|
if f.jobs[i].ID == id && f.jobs[i].Status == "running" {
|
|
f.jobs[i].Status = status
|
|
f.active--
|
|
return nil
|
|
}
|
|
}
|
|
return store.ErrInvalidTransition
|
|
}
|
|
|
|
func (f *queueStore) state() ([]store.Job, map[int64]int, int) {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
jobs := append([]store.Job(nil), f.jobs...)
|
|
counts := make(map[int64]int, len(f.claimCounts))
|
|
for id, count := range f.claimCounts {
|
|
counts[id] = count
|
|
}
|
|
return jobs, counts, f.maxActive
|
|
}
|
|
|
|
func (f *fakeStore) ClaimNext(context.Context) (store.Job, error) {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
if f.claimed || f.job.Status != "queued" {
|
|
return store.Job{}, pgx.ErrNoRows
|
|
}
|
|
f.claimed = true
|
|
f.job.Status = "running"
|
|
return f.job, nil
|
|
}
|
|
|
|
func (f *fakeStore) UpdateProgress(_ context.Context, _ int64, p store.Progress) error {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
if f.job.Status != "running" {
|
|
return store.ErrInvalidTransition
|
|
}
|
|
f.progress = append(f.progress, p)
|
|
return nil
|
|
}
|
|
|
|
func (f *fakeStore) Heartbeat(context.Context, int64) error {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
if f.job.Status != "running" {
|
|
return store.ErrInvalidTransition
|
|
}
|
|
f.heartbeats++
|
|
return nil
|
|
}
|
|
|
|
func (f *fakeStore) GetJob(context.Context, int64) (store.Job, error) {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
return f.job, nil
|
|
}
|
|
|
|
func (f *fakeStore) Complete(_ context.Context, _ int64, path string, size int64) error {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
if f.job.Status != "running" {
|
|
return store.ErrInvalidTransition
|
|
}
|
|
f.job.Status = "succeeded"
|
|
f.completed, f.completePath, f.completeSize = true, path, size
|
|
return nil
|
|
}
|
|
|
|
func (f *fakeStore) Fail(_ context.Context, _ int64, kind, message string) error {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
if f.job.Status != "running" {
|
|
return store.ErrInvalidTransition
|
|
}
|
|
f.job.Status = "failed"
|
|
f.failedKind, f.failedMessage = kind, message
|
|
return nil
|
|
}
|
|
|
|
func (f *fakeStore) RequeueRunning(context.Context, int64) error {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
if f.job.Status != "running" {
|
|
return store.ErrInvalidTransition
|
|
}
|
|
f.job.Status = "queued"
|
|
f.requeued = true
|
|
return nil
|
|
}
|
|
|
|
func (f *fakeStore) setStatus(status string) {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
f.job.Status = status
|
|
}
|
|
|
|
func (f *fakeStore) heartbeatCount() int {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
return f.heartbeats
|
|
}
|
|
|
|
func testRunner(t *testing.T, jobs *fakeStore, videoDir string) *Runner {
|
|
t.Helper()
|
|
ytdlp := filepath.Join("testdata", "fake-yt-dlp.sh")
|
|
ffprobe := filepath.Join("testdata", "fake-ffprobe.sh")
|
|
for _, path := range []string{ytdlp, ffprobe} {
|
|
info, err := os.Stat(path)
|
|
if err != nil || info.Mode().Perm()&0o111 == 0 {
|
|
t.Fatalf("偽コマンドが実行可能ではありません: %s: %v", path, err)
|
|
}
|
|
}
|
|
runner, err := New(jobs, Config{VideoDir: videoDir, YTDLPPath: ytdlp, FFprobePath: ffprobe,
|
|
Concurrency: 1, PollInterval: 5 * time.Millisecond, ProgressInterval: time.Millisecond,
|
|
HeartbeatInterval: 10 * time.Millisecond, CancelInterval: 5 * time.Millisecond}, log.New(io.Discard, "", 0))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return runner
|
|
}
|
|
|
|
func runningJob() store.Job {
|
|
return store.Job{ID: 7, EpisodeID: "ep123", SeriesTitle: "番組名", Title: "第1話", Status: "running"}
|
|
}
|
|
|
|
func relativeVideoPath(t *testing.T, seriesTitle, episodeTitle, episodeID string) string {
|
|
t.Helper()
|
|
path, err := media.RelativeVideoPath(seriesTitle, episodeTitle, episodeID)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return path
|
|
}
|
|
|
|
func TestProcessJobDownloadsProbesMovesAndCleans(t *testing.T) {
|
|
t.Setenv("FAKE_YTDLP_MODE", "success")
|
|
t.Setenv("FAKE_FFPROBE_MODE", "valid")
|
|
dir := t.TempDir()
|
|
argsFile := filepath.Join(dir, "args")
|
|
t.Setenv("FAKE_ARGS_FILE", argsFile)
|
|
jobs := &fakeStore{job: runningJob()}
|
|
testRunner(t, jobs, dir).ProcessJob(context.Background(), jobs.job)
|
|
want := filepath.ToSlash(relativeVideoPath(t, "番組名", "第1話", "ep123"))
|
|
if !jobs.completed || jobs.completePath != want || jobs.completeSize == 0 {
|
|
t.Fatalf("complete: done=%v path=%q size=%d", jobs.completed, jobs.completePath, jobs.completeSize)
|
|
}
|
|
if _, err := os.Stat(filepath.Join(dir, filepath.FromSlash(want))); err != nil {
|
|
t.Fatalf("保存先: %v", err)
|
|
}
|
|
if _, err := os.Stat(filepath.Join(dir, ".incoming", "7")); !errors.Is(err, os.ErrNotExist) {
|
|
t.Fatalf("作業ディレクトリが残った: %v", err)
|
|
}
|
|
args, err := os.ReadFile(argsFile)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
argumentText := string(args)
|
|
for _, want := range []string{"--newline", "--no-playlist", "--no-simulate", "--quiet", "--progress", "--continue", "--merge-output-format", "bv*[vcodec^=avc1]+ba/b", `download:{"event":"download","format_id":%(info.format_id)j,"progress":%(progress)j}`, `postprocess:{"event":"postprocess","progress":%(progress)j}`, `after_move:{"event":"after_move","filepath":%(filepath)j}`, filepath.Join(dir, ".incoming", "7", "%(id)s.%(ext)s")} {
|
|
if !strings.Contains(argumentText, want) {
|
|
t.Errorf("yt-dlp 引数に %q がない: %s", want, argumentText)
|
|
}
|
|
}
|
|
if strings.Contains(argumentText, "番組名") || strings.Contains(argumentText, "第1話") {
|
|
t.Errorf("出力テンプレートに番組情報が入った: %s", argumentText)
|
|
}
|
|
phases := make(map[string]bool)
|
|
for _, progress := range jobs.progress {
|
|
phases[progress.Phase] = true
|
|
}
|
|
for _, phase := range []string{"video", "audio", "merging", "moving"} {
|
|
if !phases[phase] {
|
|
t.Errorf("進捗段階 %s がない: %+v", phase, jobs.progress)
|
|
}
|
|
}
|
|
if jobs.heartbeats < 2 {
|
|
t.Errorf("heartbeat が少ない: %d", jobs.heartbeats)
|
|
}
|
|
}
|
|
|
|
func TestProcessJobRejectsLongEpisodeIDAndCleans(t *testing.T) {
|
|
dir := t.TempDir()
|
|
argsFile := filepath.Join(dir, "yt-dlp-args")
|
|
t.Setenv("FAKE_ARGS_FILE", argsFile)
|
|
workDir := filepath.Join(dir, ".incoming", "7")
|
|
if err := os.MkdirAll(workDir, 0o755); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := os.WriteFile(filepath.Join(workDir, "stale"), []byte("stale"), 0o644); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
job := runningJob()
|
|
job.EpisodeID = strings.Repeat("i", 65)
|
|
jobs := &fakeStore{job: job}
|
|
testRunner(t, jobs, dir).ProcessJob(context.Background(), job)
|
|
if jobs.completed || jobs.failedKind != "invalid_output" || !strings.Contains(jobs.failedMessage, "64バイト以下") {
|
|
t.Fatalf("長いIDの失敗: kind=%q message=%q completed=%v", jobs.failedKind, jobs.failedMessage, jobs.completed)
|
|
}
|
|
if _, err := os.Stat(argsFile); !errors.Is(err, os.ErrNotExist) {
|
|
t.Fatalf("yt-dlp が起動された: %v", err)
|
|
}
|
|
if _, err := os.Stat(workDir); !errors.Is(err, os.ErrNotExist) {
|
|
t.Fatalf("作業ディレクトリが残った: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestProcessJobKeepsPercentNamesOutOfYTDLPTemplate(t *testing.T) {
|
|
t.Setenv("FAKE_YTDLP_MODE", "success")
|
|
t.Setenv("FAKE_FFPROBE_MODE", "valid")
|
|
dir := t.TempDir()
|
|
argsFile := filepath.Join(dir, "args")
|
|
t.Setenv("FAKE_ARGS_FILE", argsFile)
|
|
job := runningJob()
|
|
job.SeriesTitle = "100%番組"
|
|
job.Title = "第%1話"
|
|
jobs := &fakeStore{job: job}
|
|
testRunner(t, jobs, dir).ProcessJob(context.Background(), job)
|
|
if !jobs.completed {
|
|
t.Fatalf("%% を含む番組を完了できなかった: %+v", jobs)
|
|
}
|
|
args, err := os.ReadFile(argsFile)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if strings.Contains(string(args), job.SeriesTitle) || strings.Contains(string(args), job.Title) {
|
|
t.Fatalf("番組情報が yt-dlp 引数に入った: %s", args)
|
|
}
|
|
want := filepath.ToSlash(relativeVideoPath(t, job.SeriesTitle, job.Title, job.EpisodeID))
|
|
if jobs.completePath != want {
|
|
t.Fatalf("保存先=%q want=%q", jobs.completePath, want)
|
|
}
|
|
}
|
|
|
|
func TestProcessJobReplacesExistingDestination(t *testing.T) {
|
|
t.Setenv("FAKE_YTDLP_MODE", "success")
|
|
t.Setenv("FAKE_FFPROBE_MODE", "valid")
|
|
dir := t.TempDir()
|
|
job := runningJob()
|
|
relative := relativeVideoPath(t, job.SeriesTitle, job.Title, job.EpisodeID)
|
|
destination := filepath.Join(dir, relative)
|
|
if err := os.MkdirAll(filepath.Dir(destination), 0o755); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := os.WriteFile(destination, []byte("stale file from previous attempt"), 0o644); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
jobs := &fakeStore{job: job}
|
|
testRunner(t, jobs, dir).ProcessJob(context.Background(), job)
|
|
got, err := os.ReadFile(destination)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !jobs.completed || string(got) != "fake mp4" || jobs.completeSize != int64(len(got)) {
|
|
t.Fatalf("既存ファイルの置換: completed=%v content=%q size=%d", jobs.completed, got, jobs.completeSize)
|
|
}
|
|
}
|
|
|
|
func TestProcessJobRejectsAfterMoveSymlinkOutsideWorkDirectory(t *testing.T) {
|
|
t.Setenv("FAKE_YTDLP_MODE", "symlink")
|
|
t.Setenv("FAKE_FFPROBE_MODE", "valid")
|
|
dir := t.TempDir()
|
|
outside := filepath.Join(t.TempDir(), "outside.mp4")
|
|
if err := os.WriteFile(outside, []byte("outside video"), 0o644); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Setenv("FAKE_OUTSIDE_FILE", outside)
|
|
jobs := &fakeStore{job: runningJob()}
|
|
testRunner(t, jobs, dir).ProcessJob(context.Background(), jobs.job)
|
|
if jobs.completed || jobs.failedKind != "invalid_output" {
|
|
t.Fatalf("作業ディレクトリ外を指すシンボリックリンクを受理した: %+v", jobs)
|
|
}
|
|
if got, err := os.ReadFile(outside); err != nil || string(got) != "outside video" {
|
|
t.Fatalf("作業ディレクトリ外のファイルを変更した: %q %v", got, err)
|
|
}
|
|
}
|
|
|
|
func TestPathInsideSymlinkDerivedCases(t *testing.T) {
|
|
root := t.TempDir()
|
|
insideDir := filepath.Join(root, "real")
|
|
if err := os.Mkdir(insideDir, 0o755); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
insideFile := filepath.Join(insideDir, "video.mp4")
|
|
if err := os.WriteFile(insideFile, []byte("video"), 0o644); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
outsideDir := t.TempDir()
|
|
outsideFile := filepath.Join(outsideDir, "outside.mp4")
|
|
if err := os.WriteFile(outsideFile, []byte("outside"), 0o644); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
internalFileLink := filepath.Join(root, "internal-file-link.mp4")
|
|
internalDirLink := filepath.Join(root, "internal-dir-link")
|
|
externalDirLink := filepath.Join(root, "external-dir-link")
|
|
rootLink := filepath.Join(t.TempDir(), "root-link")
|
|
for link, target := range map[string]string{
|
|
internalFileLink: insideFile,
|
|
internalDirLink: insideDir,
|
|
externalDirLink: outsideDir,
|
|
rootLink: root,
|
|
} {
|
|
if err := os.Symlink(target, link); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
for _, tt := range []struct {
|
|
name, testRoot, candidate string
|
|
wantOK bool
|
|
}{
|
|
{"通常ファイル", root, insideFile, true},
|
|
{"作業ディレクトリ内を指すファイルリンク", root, internalFileLink, false},
|
|
{"after_moveがディレクトリ", root, insideDir, false},
|
|
{"作業ディレクトリ内を指すディレクトリリンク配下", root, filepath.Join(internalDirLink, "video.mp4"), true},
|
|
{"作業ディレクトリ外を指すディレクトリリンク配下", root, filepath.Join(externalDirLink, "outside.mp4"), false},
|
|
{"作業ディレクトリ自体がリンク", rootLink, filepath.Join(rootLink, "real", "video.mp4"), true},
|
|
} {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
got, err := pathInside(tt.testRoot, tt.candidate)
|
|
if tt.wantOK {
|
|
want, absErr := filepath.Abs(tt.candidate)
|
|
if absErr != nil {
|
|
t.Fatal(absErr)
|
|
}
|
|
if err != nil || got != want {
|
|
t.Fatalf("pathInside()=(%q, %v) want=(%q, nil)", got, err, want)
|
|
}
|
|
return
|
|
}
|
|
if !errors.Is(err, errInvalidOutput) || got != "" {
|
|
t.Fatalf("pathInside()=(%q, %v), invalid_output を期待", got, err)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestProcessJobDestinationDirectoryFailureCleansIncoming(t *testing.T) {
|
|
t.Setenv("FAKE_YTDLP_MODE", "success")
|
|
t.Setenv("FAKE_FFPROBE_MODE", "valid")
|
|
dir := t.TempDir()
|
|
job := runningJob()
|
|
seriesDir := filepath.Join(dir, filepath.Dir(relativeVideoPath(t, job.SeriesTitle, job.Title, job.EpisodeID)))
|
|
if err := os.WriteFile(seriesDir, []byte("blocks directory"), 0o644); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
jobs := &fakeStore{job: job}
|
|
testRunner(t, jobs, dir).ProcessJob(context.Background(), job)
|
|
if jobs.completed || jobs.failedKind != "unknown" {
|
|
t.Fatalf("保存先作成失敗: %+v", jobs)
|
|
}
|
|
if _, err := os.Stat(filepath.Join(dir, ".incoming", "7")); !errors.Is(err, os.ErrNotExist) {
|
|
t.Fatalf("作業ディレクトリが残った: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestFailureClassificationAndRedaction(t *testing.T) {
|
|
for _, tt := range []struct{ mode, kind string }{
|
|
{"drm", "drm"}, {"geo", "geo"}, {"network", "network"},
|
|
{"unavailable", "unavailable"}, {"extract", "extract"}, {"unknown", "unknown"},
|
|
} {
|
|
t.Run(tt.mode, func(t *testing.T) {
|
|
t.Setenv("FAKE_YTDLP_MODE", tt.mode)
|
|
dir := t.TempDir()
|
|
jobs := &fakeStore{job: runningJob()}
|
|
testRunner(t, jobs, dir).ProcessJob(context.Background(), jobs.job)
|
|
if jobs.failedKind != tt.kind || jobs.completed || strings.Contains(jobs.failedMessage, "secret-value") {
|
|
t.Fatalf("failure: kind=%q message=%q completed=%v", jobs.failedKind, jobs.failedMessage, jobs.completed)
|
|
}
|
|
if _, err := os.Stat(filepath.Join(dir, ".incoming", "7")); !errors.Is(err, os.ErrNotExist) {
|
|
t.Fatalf("作業ディレクトリが残った: %v", err)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestInvalidProbeFailsAndCleans(t *testing.T) {
|
|
t.Setenv("FAKE_YTDLP_MODE", "success")
|
|
for _, mode := range []string{"bad-container", "bad-video", "bad-audio", "non-json", "failure"} {
|
|
t.Run(mode, func(t *testing.T) {
|
|
t.Setenv("FAKE_FFPROBE_MODE", mode)
|
|
dir := t.TempDir()
|
|
jobs := &fakeStore{job: runningJob()}
|
|
testRunner(t, jobs, dir).ProcessJob(context.Background(), jobs.job)
|
|
if jobs.failedKind != "invalid_output" || jobs.completed {
|
|
t.Fatalf("ffprobe 不合格: kind=%q complete=%v", jobs.failedKind, jobs.completed)
|
|
}
|
|
if _, err := os.Stat(filepath.Join(dir, ".incoming", "7")); !errors.Is(err, os.ErrNotExist) {
|
|
t.Fatalf("作業ディレクトリが残った: %v", err)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestProgressIsThrottledButHeartbeatContinues(t *testing.T) {
|
|
t.Setenv("FAKE_YTDLP_MODE", "burst")
|
|
t.Setenv("FAKE_FFPROBE_MODE", "valid")
|
|
dir := t.TempDir()
|
|
jobs := &fakeStore{job: runningJob()}
|
|
runner := testRunner(t, jobs, dir)
|
|
runner.config.ProgressInterval = 50 * time.Millisecond
|
|
runner.ProcessJob(context.Background(), jobs.job)
|
|
downloadUpdates := 0
|
|
for _, progress := range jobs.progress {
|
|
if progress.Phase == "video" {
|
|
downloadUpdates++
|
|
}
|
|
}
|
|
if downloadUpdates > 2 || downloadUpdates == 0 {
|
|
t.Fatalf("進捗の間引き: %d updates (%+v)", downloadUpdates, jobs.progress)
|
|
}
|
|
if jobs.heartbeats < 5 {
|
|
t.Fatalf("進捗が止まった間の heartbeat: %d", jobs.heartbeats)
|
|
}
|
|
}
|
|
|
|
func TestRunConcurrencyTwoDoesNotDuplicateAndIsolatesFailure(t *testing.T) {
|
|
t.Setenv("FAKE_YTDLP_MODE", "mixed")
|
|
t.Setenv("FAKE_FFPROBE_MODE", "valid")
|
|
dir := t.TempDir()
|
|
jobs := &queueStore{
|
|
jobs: []store.Job{
|
|
{ID: 7, EpisodeID: "ep-ok", SeriesTitle: "成功番組", Title: "成功回", Status: "queued"},
|
|
{ID: 8, EpisodeID: "ep-fail", SeriesTitle: "失敗番組", Title: "失敗回", Status: "queued"},
|
|
},
|
|
claimCounts: make(map[int64]int),
|
|
}
|
|
runner, err := New(jobs, Config{VideoDir: dir,
|
|
YTDLPPath: filepath.Join("testdata", "fake-yt-dlp.sh"), FFprobePath: filepath.Join("testdata", "fake-ffprobe.sh"),
|
|
Concurrency: 2, PollInterval: 5 * time.Millisecond, ProgressInterval: time.Millisecond,
|
|
HeartbeatInterval: 10 * time.Millisecond, CancelInterval: 5 * time.Millisecond}, log.New(io.Discard, "", 0))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan error, 1)
|
|
go func() { done <- runner.Run(ctx) }()
|
|
deadline := time.Now().Add(3 * time.Second)
|
|
for time.Now().Before(deadline) {
|
|
state, _, _ := jobs.state()
|
|
if state[0].Status == "succeeded" && state[1].Status == "failed" {
|
|
break
|
|
}
|
|
time.Sleep(5 * time.Millisecond)
|
|
}
|
|
cancel()
|
|
if err := <-done; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
state, counts, maxActive := jobs.state()
|
|
if state[0].Status != "succeeded" || state[1].Status != "failed" {
|
|
t.Fatalf("片方の失敗がもう片方へ波及した: %+v", state)
|
|
}
|
|
if counts[7] != 1 || counts[8] != 1 {
|
|
t.Fatalf("ジョブを二重取得した: %+v", counts)
|
|
}
|
|
if maxActive != 2 {
|
|
t.Fatalf("並行実行数の最大値=%d want=2", maxActive)
|
|
}
|
|
for _, id := range []int64{7, 8} {
|
|
if _, err := os.Stat(filepath.Join(dir, ".incoming", strconv.FormatInt(id, 10))); !errors.Is(err, os.ErrNotExist) {
|
|
t.Errorf("ジョブ %d の作業ディレクトリが残った: %v", id, err)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestCancelStopsProcessGroupAndLeavesCanceled(t *testing.T) {
|
|
t.Setenv("FAKE_YTDLP_MODE", "hang-drm")
|
|
dir := t.TempDir()
|
|
ready := filepath.Join(dir, "ready")
|
|
terminated := filepath.Join(dir, "terminated")
|
|
t.Setenv("FAKE_READY_FILE", ready)
|
|
t.Setenv("FAKE_TERM_FILE", terminated)
|
|
jobs := &fakeStore{job: runningJob()}
|
|
runner := testRunner(t, jobs, dir)
|
|
job := jobs.job
|
|
done := make(chan struct{})
|
|
go func() { runner.ProcessJob(context.Background(), job); close(done) }()
|
|
waitForFile(t, ready)
|
|
waitForHeartbeatCount(t, jobs, 2)
|
|
jobs.setStatus("canceled")
|
|
waitDone(t, done)
|
|
if jobs.failedKind != "" || jobs.requeued || jobs.completed || jobs.job.Status != "canceled" {
|
|
t.Fatalf("取り消し後: %+v", jobs)
|
|
}
|
|
waitForFile(t, terminated)
|
|
if _, err := os.Stat(filepath.Join(dir, ".incoming", "7")); !errors.Is(err, os.ErrNotExist) {
|
|
t.Fatalf("作業ディレクトリが残った: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestShutdownStopsProcessAndRequeues(t *testing.T) {
|
|
t.Setenv("FAKE_YTDLP_MODE", "hang")
|
|
dir := t.TempDir()
|
|
ready := filepath.Join(dir, "ready")
|
|
t.Setenv("FAKE_READY_FILE", ready)
|
|
jobs := &fakeStore{job: runningJob()}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
runner := testRunner(t, jobs, dir)
|
|
job := jobs.job
|
|
done := make(chan struct{})
|
|
go func() { runner.ProcessJob(ctx, job); close(done) }()
|
|
waitForFile(t, ready)
|
|
cancel()
|
|
waitDone(t, done)
|
|
if !jobs.requeued || jobs.job.Status != "queued" || jobs.failedKind != "" {
|
|
t.Fatalf("終了時の復帰: %+v", jobs)
|
|
}
|
|
if _, err := os.Stat(filepath.Join(dir, ".incoming", "7")); !errors.Is(err, os.ErrNotExist) {
|
|
t.Fatalf("作業ディレクトリが残った: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestRunnerRejectsConcurrencyOutsideRange(t *testing.T) {
|
|
for _, concurrency := range []int{-1, 3} {
|
|
if _, err := New(&fakeStore{}, Config{Concurrency: concurrency}, nil); err == nil {
|
|
t.Errorf("concurrency=%d を許した", concurrency)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestFailureClassificationOrderAndRedaction(t *testing.T) {
|
|
for _, tt := range []struct{ message, want string }{
|
|
{"DRM geo-restricted connection video unavailable unable to extract", "drm"},
|
|
{"geo-restricted connection video unavailable unable to extract", "geo"},
|
|
{"connection video unavailable unable to extract", "network"},
|
|
{"video unavailable unable to extract", "unavailable"},
|
|
{"unable to extract", "extract"},
|
|
{"different error", "unknown"},
|
|
} {
|
|
if got := classifyFailure(tt.message); got != tt.want {
|
|
t.Errorf("classifyFailure(%q)=%q want=%q", tt.message, got, tt.want)
|
|
}
|
|
}
|
|
redacted := redact("Authorization: Bearer bearer-secret\nCookie: session=session-secret\ntoken=token-secret platform_uid=uid-secret")
|
|
for _, secret := range []string{"bearer-secret", "session-secret", "token-secret", "uid-secret"} {
|
|
if strings.Contains(redacted, secret) {
|
|
t.Errorf("秘密が残った: %q", redacted)
|
|
}
|
|
}
|
|
}
|
|
|
|
func waitForFile(t *testing.T, path string) {
|
|
t.Helper()
|
|
deadline := time.Now().Add(2 * time.Second)
|
|
for time.Now().Before(deadline) {
|
|
if _, err := os.Stat(path); err == nil {
|
|
return
|
|
}
|
|
time.Sleep(5 * time.Millisecond)
|
|
}
|
|
t.Fatalf("ファイル待ちがタイムアウト: %s", path)
|
|
}
|
|
|
|
func waitDone(t *testing.T, done <-chan struct{}) {
|
|
t.Helper()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("worker の終了待ちがタイムアウト")
|
|
}
|
|
}
|
|
|
|
func waitForHeartbeatCount(t *testing.T, jobs *fakeStore, want int) {
|
|
t.Helper()
|
|
deadline := time.Now().Add(2 * time.Second)
|
|
for time.Now().Before(deadline) {
|
|
if jobs.heartbeatCount() >= want {
|
|
return
|
|
}
|
|
time.Sleep(5 * time.Millisecond)
|
|
}
|
|
t.Fatalf("heartbeat 待ちがタイムアウト: got=%d want=%d", jobs.heartbeatCount(), want)
|
|
}
|