Files
jellybit/internal/ingest/ingest_test.go
T
avandClaude Opus 4.8 70d8758646 Восстановление зависших загрузок и уведомления о падении (state-reconciliation)
Долгий metaDL больше не убивается агрессивным таймаутом: дефолт
magnet_timeout 30m → 24h (страховочный предохранитель), базис отсчёта —
added_on из qBittorrent, а не created_at (переживает retry/усыновление).

Авто-восстановление: фоновая сверка возвращает в поток задачи, упавшие по
нашей нетерпеливости (magnet_timeout/stalled), когда источник ожил и
продвинулся за условие падения (downloading/completed по статусу торрента);
qbit_error не воскрешается. Конфликт idempotency (infohash занят другой
активной задачей) — оставляем в failed.

Уведомления: любой переход в failed/stuck пингует автора (включая приёмный
qbit_add через ingest), с дебаунсом против спама при флаппинге stalled.
Ручной retry добавлен в веб-UI и Telegram; Retry перецепляется к живому
торренту вместо слепого Add.

Дельта state-reconciliation влита в живые спеки; обновлён workflow.md.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-30 14:51:57 +03:00

211 lines
6.2 KiB
Go

package ingest
import (
"context"
"errors"
"io"
"log/slog"
"testing"
"time"
"git.vakhrushev.me/av/jellybit/internal/qbt"
"git.vakhrushev.me/av/jellybit/internal/store"
)
const sampleMagnet = "magnet:?xt=urn:btih:541ADCFF3B6DD5DBA7088EA83317D9D6FAC331D6&dn=Dune"
const sampleInfohash = "541adcff3b6dd5dba7088ea83317d9d6fac331d6"
type fakeStore struct {
active *store.Download
created []store.Download
nextID int64
stateCalls []stateCall
}
type stateCall struct {
id int64
state store.State
code string
msg string
}
func (f *fakeStore) FindActiveByInfohash(_ context.Context, _ string) (*store.Download, error) {
return f.active, nil
}
func (f *fakeStore) CreateDownload(_ context.Context, d *store.Download) (int64, error) {
f.nextID++
d.ID = f.nextID
f.created = append(f.created, *d)
return f.nextID, nil
}
func (f *fakeStore) SetDownloadState(_ context.Context, id int64, st store.State, code, msg string) error {
f.stateCalls = append(f.stateCalls, stateCall{id, st, code, msg})
return nil
}
type fakeQbt struct {
added []qbt.AddRequest
err error
}
func (f *fakeQbt) Add(_ context.Context, ar qbt.AddRequest) error {
if f.err != nil {
return f.err
}
f.added = append(f.added, ar)
return nil
}
// fakeNamer возвращает заранее заданное имя; фиксирует переданные аргументы.
type fakeNamer struct {
name string
gotContext string
gotHint string
called bool
}
func (f *fakeNamer) DeriveName(_ context.Context, contextText, hint string) string {
f.called = true
f.gotContext = contextText
f.gotHint = hint
return f.name
}
func newService(st Store, qb QBittorrent) *Service {
return newServiceWithNamer(st, qb, nil)
}
func newServiceWithNamer(st Store, qb QBittorrent, nm Namer) *Service {
return New(st, qb, nm, Config{Category: "jellybit", SavePath: "/srv/media/downloads"},
slog.New(slog.NewTextHandler(io.Discard, nil)))
}
func TestIngestHappyPath(t *testing.T) {
fs := &fakeStore{}
fq := &fakeQbt{}
res, err := newService(fs, fq).Ingest(context.Background(), Request{Source: sampleMagnet, Context: "Дюна 2"})
if err != nil {
t.Fatalf("Ingest: %v", err)
}
if res.Infohash != sampleInfohash {
t.Errorf("infohash = %q", res.Infohash)
}
if res.State != store.StateDownloading || res.Deduplicated {
t.Errorf("res = %+v", res)
}
if len(fs.created) != 1 {
t.Fatalf("создано задач: %d, want 1", len(fs.created))
}
if got := fs.created[0]; got.Context != "Дюна 2" || got.Infohash.String != sampleInfohash {
t.Errorf("сохранённая задача: %+v", got)
}
if len(fq.added) != 1 {
t.Fatalf("вызовов qbt.Add: %d, want 1", len(fq.added))
}
add := fq.added[0]
if len(add.URLs) != 1 || add.URLs[0] != sampleMagnet {
t.Errorf("URLs = %v", add.URLs)
}
if add.Category != "jellybit" || add.SavePath != "/srv/media/downloads" {
t.Errorf("category/savepath = %q/%q", add.Category, add.SavePath)
}
}
func TestIngestSetsDisplayName(t *testing.T) {
fs := &fakeStore{}
fq := &fakeQbt{}
nm := &fakeNamer{name: "Дюна: Часть вторая (2024)"}
_, err := newServiceWithNamer(fs, fq, nm).Ingest(context.Background(),
Request{Source: sampleMagnet, Context: "Дюна 2"})
if err != nil {
t.Fatalf("Ingest: %v", err)
}
if !nm.called || nm.gotContext != "Дюна 2" || nm.gotHint != "Dune" {
t.Errorf("namer получил context=%q hint=%q (called=%v)", nm.gotContext, nm.gotHint, nm.called)
}
if len(fq.added) != 1 || fq.added[0].Rename != "Дюна: Часть вторая (2024)" {
t.Errorf("rename = %q, want %q", fq.added[0].Rename, "Дюна: Часть вторая (2024)")
}
}
func TestIngestEmptyNameOmitsRename(t *testing.T) {
fs := &fakeStore{}
fq := &fakeQbt{}
nm := &fakeNamer{name: ""} // имя не получено
if _, err := newServiceWithNamer(fs, fq, nm).Ingest(context.Background(),
Request{Source: sampleMagnet}); err != nil {
t.Fatalf("Ingest: %v", err)
}
if len(fq.added) != 1 || fq.added[0].Rename != "" {
t.Errorf("rename = %q, want пусто", fq.added[0].Rename)
}
}
func TestIngestIdempotent(t *testing.T) {
existing := &store.Download{ID: 7, State: store.StateDownloading}
fs := &fakeStore{active: existing}
fq := &fakeQbt{}
res, err := newService(fs, fq).Ingest(context.Background(), Request{Source: sampleMagnet})
if err != nil {
t.Fatalf("Ingest: %v", err)
}
if !res.Deduplicated || res.DownloadID != 7 {
t.Errorf("ожидалось присоединение к задаче 7: %+v", res)
}
if len(fs.created) != 0 {
t.Error("не должно создаваться новой задачи")
}
if len(fq.added) != 0 {
t.Error("не должно быть повторного добавления в qBittorrent")
}
}
func TestIngestQbitErrorMarksFailed(t *testing.T) {
fs := &fakeStore{}
fq := &fakeQbt{err: errors.New("connection refused")}
res, err := newService(fs, fq).Ingest(context.Background(), Request{Source: sampleMagnet})
if err == nil {
t.Fatal("ожидалась ошибка")
}
if res.State != store.StateFailed {
t.Errorf("state = %q, want failed", res.State)
}
if len(fs.stateCalls) != 1 || fs.stateCalls[0].state != store.StateFailed {
t.Errorf("ожидался перевод в failed: %+v", fs.stateCalls)
}
}
func TestIngestQbitErrorNotifies(t *testing.T) {
fs := &fakeStore{}
fq := &fakeQbt{err: errors.New("connection refused")}
svc := newService(fs, fq)
got := make(chan int64, 1)
svc.SetFailureNotifier(func(id int64) { got <- id })
if _, err := svc.Ingest(context.Background(), Request{Source: sampleMagnet}); err == nil {
t.Fatal("ожидалась ошибка")
}
select {
case id := <-got:
if id == 0 {
t.Errorf("уведомление с нулевым id")
}
case <-time.After(2 * time.Second):
t.Fatal("уведомление о падении приёма не пришло")
}
}
func TestIngestRejectsNonMagnet(t *testing.T) {
fs := &fakeStore{}
fq := &fakeQbt{}
if _, err := newService(fs, fq).Ingest(context.Background(), Request{Source: "https://example.com/x.torrent"}); err == nil {
t.Fatal("ожидалась ошибка для не-magnet источника")
}
if len(fs.created) != 0 || len(fq.added) != 0 {
t.Error("не должно быть ни записи, ни добавления")
}
}