254 lines
9.0 KiB
Go
254 lines
9.0 KiB
Go
package worker
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"errors"
|
|
"log"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"gitea.kazu634.com/kazu634/tver/internal/store"
|
|
"gitea.kazu634.com/kazu634/tver/internal/tver"
|
|
)
|
|
|
|
type schedulerCompletion struct {
|
|
seriesID string
|
|
episodes []store.Episode
|
|
limit int
|
|
at time.Time
|
|
}
|
|
|
|
type schedulerFailure struct {
|
|
seriesID, message string
|
|
at time.Time
|
|
}
|
|
|
|
type fakeSubscriptionStore struct {
|
|
mu sync.Mutex
|
|
due []store.Subscription
|
|
cutoffs []time.Time
|
|
completions []schedulerCompletion
|
|
failures []schedulerFailure
|
|
listErr error
|
|
completeErr error
|
|
}
|
|
|
|
func (f *fakeSubscriptionStore) ListDueSubscriptions(_ context.Context, cutoff time.Time) ([]store.Subscription, error) {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
f.cutoffs = append(f.cutoffs, cutoff)
|
|
return append([]store.Subscription(nil), f.due...), f.listErr
|
|
}
|
|
|
|
func (f *fakeSubscriptionStore) CompleteSubscriptionCheck(_ context.Context, seriesID string, episodes []store.Episode, limit int, at time.Time) (int, error) {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
f.completions = append(f.completions, schedulerCompletion{seriesID, append([]store.Episode(nil), episodes...), limit, at})
|
|
return len(episodes), f.completeErr
|
|
}
|
|
|
|
func (f *fakeSubscriptionStore) RecordSubscriptionCheckFailure(_ context.Context, seriesID string, at time.Time, message string) error {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
f.failures = append(f.failures, schedulerFailure{seriesID, message, at})
|
|
return nil
|
|
}
|
|
|
|
type fakeSeriesFetcher struct {
|
|
mu sync.Mutex
|
|
calls []string
|
|
episodes map[string][]tver.Episode
|
|
errors map[string]error
|
|
block bool
|
|
started chan string
|
|
}
|
|
|
|
func (f *fakeSeriesFetcher) FetchSeriesEpisodes(ctx context.Context, seriesID string) ([]tver.Episode, error) {
|
|
f.mu.Lock()
|
|
f.calls = append(f.calls, seriesID)
|
|
block := f.block
|
|
started := f.started
|
|
episodes := append([]tver.Episode(nil), f.episodes[seriesID]...)
|
|
err := f.errors[seriesID]
|
|
f.mu.Unlock()
|
|
if started != nil {
|
|
started <- seriesID
|
|
}
|
|
if block {
|
|
<-ctx.Done()
|
|
return nil, ctx.Err()
|
|
}
|
|
return episodes, err
|
|
}
|
|
|
|
func TestSubscriptionSchedulerChecksImmediatelyAndMapsEpisodes(t *testing.T) {
|
|
now := time.Date(2026, 10, 3, 12, 0, 0, 0, time.UTC)
|
|
subscriptions := &fakeSubscriptionStore{due: []store.Subscription{{SeriesID: "one"}, {SeriesID: "two"}}}
|
|
fetcher := &fakeSeriesFetcher{episodes: map[string][]tver.Episode{
|
|
"one": {{ID: "ep1", SeriesID: "one", SeriesTitle: "番組", Title: "回", BroadcasterName: "局", Duration: 180, EndAt: 1_800_000_000, ThumbnailPath: "/image"}},
|
|
"two": {{ID: "ep2", SeriesID: "two", Duration: 181}},
|
|
}, errors: map[string]error{}}
|
|
var waits []time.Duration
|
|
after := func(duration time.Duration) <-chan time.Time {
|
|
waits = append(waits, duration)
|
|
ch := make(chan time.Time, 1)
|
|
ch <- now
|
|
return ch
|
|
}
|
|
scheduler, err := NewSubscriptionScheduler(subscriptions, fetcher, SubscriptionSchedulerConfig{
|
|
DueAfter: 24 * time.Hour, SeriesInterval: 4 * time.Second,
|
|
Now: func() time.Time { return now }, After: after,
|
|
}, log.New(&bytes.Buffer{}, "", 0))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
scheduler.checkDue(context.Background())
|
|
if len(subscriptions.cutoffs) != 1 || !subscriptions.cutoffs[0].Equal(now.Add(-24*time.Hour)) {
|
|
t.Fatalf("確認対象日時: %v", subscriptions.cutoffs)
|
|
}
|
|
if len(fetcher.calls) != 2 || fetcher.calls[0] != "one" || fetcher.calls[1] != "two" || len(waits) != 1 || waits[0] != 4*time.Second {
|
|
t.Fatalf("順次確認と間隔: calls=%v waits=%v", fetcher.calls, waits)
|
|
}
|
|
if len(subscriptions.completions) != 2 || subscriptions.completions[0].limit != tver.PreviewMaxSeconds || !subscriptions.completions[0].at.Equal(now) {
|
|
t.Fatalf("確認完了: %+v", subscriptions.completions)
|
|
}
|
|
got := subscriptions.completions[0].episodes[0]
|
|
if got.ID != "ep1" || got.DurationSeconds != 180 || got.EndAt == nil || got.EndAt.Unix() != 1_800_000_000 {
|
|
t.Fatalf("エピソード変換: %+v", got)
|
|
}
|
|
}
|
|
|
|
func TestShortErrorLimitsUnicodeByCharacters(t *testing.T) {
|
|
message := shortError(errors.New(strings.Repeat("界", 260) + "\n\t" + strings.Repeat("語", 300)))
|
|
if got := len([]rune(message)); got != 512 {
|
|
t.Fatalf("エラー文字数=%d want=512", got)
|
|
}
|
|
if strings.ContainsAny(message, "\r\n\t") || strings.Contains(message, " ") {
|
|
t.Fatalf("空白を1行へ正規化していない: %q", message)
|
|
}
|
|
if !strings.HasSuffix(message, strings.Repeat("語", 251)) {
|
|
t.Fatalf("UTF-8文字の途中で切断または境界が不正: %q", message)
|
|
}
|
|
}
|
|
|
|
func TestSubscriptionSchedulerRecordsRedactedFailureWithoutCompleting(t *testing.T) {
|
|
now := time.Date(2026, 10, 3, 12, 0, 0, 0, time.UTC)
|
|
subscriptions := &fakeSubscriptionStore{due: []store.Subscription{{SeriesID: "failed"}}}
|
|
fetcher := &fakeSeriesFetcher{errors: map[string]error{
|
|
"failed": errors.New("request failed\nplatform_token=secret-token Cookie: session-secret"),
|
|
}}
|
|
var logs bytes.Buffer
|
|
scheduler, err := NewSubscriptionScheduler(subscriptions, fetcher, SubscriptionSchedulerConfig{Now: func() time.Time { return now }}, log.New(&logs, "", 0))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
scheduler.checkDue(context.Background())
|
|
if len(subscriptions.completions) != 0 || len(subscriptions.failures) != 1 || !subscriptions.failures[0].at.Equal(now) {
|
|
t.Fatalf("失敗の記録: completions=%v failures=%v", subscriptions.completions, subscriptions.failures)
|
|
}
|
|
message := subscriptions.failures[0].message
|
|
if strings.Contains(message, "secret-token") || strings.Contains(message, "session-secret") || strings.Contains(logs.String(), "secret") || strings.Contains(message, "\n") {
|
|
t.Fatalf("秘密値または改行が残った: message=%q logs=%q", message, logs.String())
|
|
}
|
|
}
|
|
|
|
func TestSubscriptionSchedulerRunStartsImmediatelyThenUsesScanInterval(t *testing.T) {
|
|
now := time.Date(2026, 10, 3, 12, 0, 0, 0, time.UTC)
|
|
subscriptions := &fakeSubscriptionStore{}
|
|
fetcher := &fakeSeriesFetcher{}
|
|
requested := make(chan time.Duration, 2)
|
|
tick := make(chan time.Time, 1)
|
|
after := func(duration time.Duration) <-chan time.Time {
|
|
requested <- duration
|
|
return tick
|
|
}
|
|
scheduler, err := NewSubscriptionScheduler(subscriptions, fetcher, SubscriptionSchedulerConfig{
|
|
ScanInterval: time.Hour, Now: func() time.Time { return now }, After: after,
|
|
}, nil)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan struct{})
|
|
go func() { scheduler.Run(ctx); close(done) }()
|
|
select {
|
|
case duration := <-requested:
|
|
if duration != time.Hour {
|
|
t.Fatalf("走査間隔: %v", duration)
|
|
}
|
|
case <-time.After(time.Second):
|
|
t.Fatal("起動直後の確認が行われなかった")
|
|
}
|
|
subscriptions.mu.Lock()
|
|
initialCalls := len(subscriptions.cutoffs)
|
|
subscriptions.mu.Unlock()
|
|
if initialCalls != 1 {
|
|
t.Fatalf("起動直後の確認回数: %d", initialCalls)
|
|
}
|
|
tick <- now
|
|
select {
|
|
case <-requested:
|
|
case <-time.After(time.Second):
|
|
t.Fatal("1時間後の確認が行われなかった")
|
|
}
|
|
cancel()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(time.Second):
|
|
t.Fatal("終了できなかった")
|
|
}
|
|
subscriptions.mu.Lock()
|
|
defer subscriptions.mu.Unlock()
|
|
if len(subscriptions.cutoffs) != 2 {
|
|
t.Fatalf("確認回数: %d", len(subscriptions.cutoffs))
|
|
}
|
|
}
|
|
|
|
func TestSubscriptionSchedulerCancellationInterruptsFetchAndSeriesWait(t *testing.T) {
|
|
now := time.Date(2026, 10, 3, 12, 0, 0, 0, time.UTC)
|
|
t.Run("fetch", func(t *testing.T) {
|
|
subscriptions := &fakeSubscriptionStore{due: []store.Subscription{{SeriesID: "one"}}}
|
|
fetcher := &fakeSeriesFetcher{block: true, started: make(chan string, 1)}
|
|
scheduler, _ := NewSubscriptionScheduler(subscriptions, fetcher, SubscriptionSchedulerConfig{Now: func() time.Time { return now }}, nil)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan struct{})
|
|
go func() { scheduler.checkDue(ctx); close(done) }()
|
|
select {
|
|
case <-fetcher.started:
|
|
case <-time.After(time.Second):
|
|
t.Fatal("取得が開始されなかった")
|
|
}
|
|
cancel()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(time.Second):
|
|
t.Fatal("取得中断で終了しなかった")
|
|
}
|
|
if len(subscriptions.failures) != 0 || len(subscriptions.completions) != 0 {
|
|
t.Fatalf("中断した番組を記録した: failures=%v completions=%v", subscriptions.failures, subscriptions.completions)
|
|
}
|
|
})
|
|
|
|
t.Run("between series", func(t *testing.T) {
|
|
subscriptions := &fakeSubscriptionStore{due: []store.Subscription{{SeriesID: "one"}, {SeriesID: "two"}}}
|
|
fetcher := &fakeSeriesFetcher{episodes: map[string][]tver.Episode{}, errors: map[string]error{}}
|
|
waiting := make(chan struct{})
|
|
never := make(chan time.Time)
|
|
scheduler, _ := NewSubscriptionScheduler(subscriptions, fetcher, SubscriptionSchedulerConfig{
|
|
Now: func() time.Time { return now }, After: func(time.Duration) <-chan time.Time { close(waiting); return never },
|
|
}, nil)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan struct{})
|
|
go func() { scheduler.checkDue(ctx); close(done) }()
|
|
<-waiting
|
|
cancel()
|
|
<-done
|
|
if len(fetcher.calls) != 1 || fetcher.calls[0] != "one" {
|
|
t.Fatalf("番組間の中断: %v", fetcher.calls)
|
|
}
|
|
})
|
|
}
|