363 lines
12 KiB
Go
363 lines
12 KiB
Go
package pipeline
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"fmt"
|
||
"io"
|
||
"net/http"
|
||
"net/http/httptest"
|
||
"os"
|
||
"path/filepath"
|
||
"strings"
|
||
"sync/atomic"
|
||
"testing"
|
||
"time"
|
||
|
||
"textmachine/backend/internal/obs"
|
||
)
|
||
|
||
// e2e-тест мини-раннера: temp-проект книги против httptest-провайдера. Деньги,
|
||
// чекпоинты и resume проверяются через реальные конфиги/store — то, что
|
||
// демонстрирует приёмка Фазы 0, только на моке.
|
||
|
||
func writeFile(t *testing.T, path, content string) {
|
||
t.Helper()
|
||
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := os.WriteFile(path, []byte(content), 0o644); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|
||
|
||
// newFakeProvider serves OpenAI-совместимые ответы: draft-стадии отвечает
|
||
// «ЧЕРНОВИК», edit-стадии — «РЕДАКТУРА» (различает по наличию слова ЧЕРНОВИК
|
||
// в промпте редактора). Считает вызовы.
|
||
func newFakeProvider(t *testing.T, calls *atomic.Int32) *httptest.Server {
|
||
return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||
calls.Add(1)
|
||
body, _ := io.ReadAll(r.Body)
|
||
text := "ЧЕРНОВИК ПЕРЕВОДА"
|
||
if bytes.Contains(body, []byte("Черновик перевода для редактуры")) {
|
||
text = "ОТРЕДАКТИРОВАННЫЙ ПЕРЕВОД"
|
||
}
|
||
fmt.Fprintf(w, `{"id":"fake-%d","model":"fake-model","choices":[{"message":{"content":"%s"},"finish_reason":"stop"}],
|
||
"usage":{"prompt_tokens":1000,"completion_tokens":500,"prompt_tokens_details":{"cached_tokens":200}}}`,
|
||
calls.Load(), text)
|
||
}))
|
||
}
|
||
|
||
func setupProject(t *testing.T, providerURL string) string {
|
||
t.Helper()
|
||
dir := t.TempDir()
|
||
|
||
writeFile(t, filepath.Join(dir, "prompts", "translator.md"),
|
||
"Переводи с {{source_lang}} на {{target_lang}}.\n---USER---\n{{text}}")
|
||
writeFile(t, filepath.Join(dir, "prompts", "editor.md"),
|
||
"Редактируй перевод.\n---USER---\nИсходник: {{text}}\nЧерновик перевода для редактуры: {{draft}}")
|
||
|
||
writeFile(t, filepath.Join(dir, "models.yaml"), fmt.Sprintf(`
|
||
prices_checked: %q
|
||
default_model: fake-model
|
||
providers:
|
||
fake:
|
||
kind: openai
|
||
base_url: %q
|
||
timeouts: { attempt_s: 5, max_attempts: 2, backoff_cap_s: 1 }
|
||
models:
|
||
fake-model:
|
||
provider: fake
|
||
price: { input_per_m: 1.0, cached_per_m: 0.1, cache_write_per_m: 0, output_per_m: 2.0 }
|
||
`, time.Now().UTC().Format("2006-01-02"), providerURL))
|
||
|
||
writeFile(t, filepath.Join(dir, "pipeline.yaml"), `
|
||
core: C1
|
||
version: 1
|
||
defaults: { max_output_ratio: 2.0, min_max_tokens: 512 }
|
||
retries: { regenerate_before_escalate: 1 }
|
||
stages:
|
||
- { name: draft, role: translator, model: fake-model, prompt: prompts/translator.md, prompt_version: v-test, temperature: 0.3, reasoning: "off" }
|
||
- { name: edit, role: editor, model: fake-model, prompt: prompts/editor.md, prompt_version: v-test, temperature: 0.4, reasoning: "off" }
|
||
`)
|
||
|
||
writeFile(t, filepath.Join(dir, "source.txt"), "静かな図書館の朝。")
|
||
writeFile(t, filepath.Join(dir, "book.yaml"), `
|
||
book_id: test-book
|
||
title: Тест
|
||
source_lang: ja
|
||
target_lang: ru
|
||
genre: ранобэ
|
||
audience: тест
|
||
venuti: 0.5
|
||
honorifics: keep
|
||
transcription: polivanov
|
||
footnotes: minimal
|
||
pipeline: pipeline.yaml
|
||
models: models.yaml
|
||
source_file: source.txt
|
||
ceilings: { book_usd: 1.0, day_usd: 2.0 }
|
||
`)
|
||
return filepath.Join(dir, "book.yaml")
|
||
}
|
||
|
||
func TestRunnerEndToEndWithResume(t *testing.T) {
|
||
var calls atomic.Int32
|
||
srv := newFakeProvider(t, &calls)
|
||
defer srv.Close()
|
||
bookPath := setupProject(t, srv.URL)
|
||
ctx := obs.WithReqInfo(context.Background(), obs.ReqInfo{TraceID: obs.NewTraceID()})
|
||
|
||
// Прогон 1: два вызова (draft, edit), деньги учтены.
|
||
r1, err := NewRunner(bookPath, obs.NewLogger())
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
res1, err := r1.TranslateOneChunk(ctx)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if calls.Load() != 2 {
|
||
t.Fatalf("expected 2 provider calls, got %d", calls.Load())
|
||
}
|
||
if len(res1.Stages) != 2 || res1.FinalText != "ОТРЕДАКТИРОВАННЫЙ ПЕРЕВОД" {
|
||
t.Fatalf("run1 = %+v", res1)
|
||
}
|
||
// Стоимость по формуле: (1000-200)*1 + 200*0.1 + 500*2 за 1M — на стадию.
|
||
wantStage := (800*1.0 + 200*0.1 + 500*2.0) / 1e6
|
||
if diff := res1.TotalUSD - 2*wantStage; diff > 1e-12 || diff < -1e-12 {
|
||
t.Fatalf("total = %v, want %v", res1.TotalUSD, 2*wantStage)
|
||
}
|
||
committed, reserved, err := r1.Store.SpentUSD("test-book")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if reserved != 0 || committed != res1.TotalUSD {
|
||
t.Fatalf("ledger: committed=%v reserved=%v want committed=%v", committed, reserved, res1.TotalUSD)
|
||
}
|
||
r1.Close()
|
||
|
||
// Прогон 2 (новый процесс — новый Runner над тем же project.db): оба
|
||
// вызова обслуживаются чекпоинтами, провайдер не трогается, доплаты нет.
|
||
r2, err := NewRunner(bookPath, obs.NewLogger())
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
defer r2.Close()
|
||
res2, err := r2.TranslateOneChunk(ctx)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if calls.Load() != 2 {
|
||
t.Fatalf("resume must not call the provider again, calls=%d", calls.Load())
|
||
}
|
||
for _, st := range res2.Stages {
|
||
if !st.FromResume || st.CostUSD != 0 {
|
||
t.Fatalf("stage %s must be served from checkpoint at $0: %+v", st.Stage, st)
|
||
}
|
||
}
|
||
if res2.FinalText != res1.FinalText {
|
||
t.Fatalf("resume text differs: %q vs %q", res2.FinalText, res1.FinalText)
|
||
}
|
||
committed2, _, err := r2.Store.SpentUSD("test-book")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if committed2 != committed {
|
||
t.Fatalf("resume must not add spend: %v -> %v", committed, committed2)
|
||
}
|
||
}
|
||
|
||
// Р6: правка конфига/промпта после старта джоб — не тихая инвалидация
|
||
// чекпоинтов, а громкая ошибка; пере-перевод только явным --resnapshot.
|
||
func TestRunnerSnapshotPinning(t *testing.T) {
|
||
var calls atomic.Int32
|
||
srv := newFakeProvider(t, &calls)
|
||
defer srv.Close()
|
||
bookPath := setupProject(t, srv.URL)
|
||
ctx := context.Background()
|
||
|
||
r1, err := NewRunner(bookPath, obs.NewLogger())
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := r1.TranslateOneChunk(ctx); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
r1.Close()
|
||
if calls.Load() != 2 {
|
||
t.Fatalf("run1 calls = %d", calls.Load())
|
||
}
|
||
|
||
// Меняем промпт переводчика → новый snapshot.
|
||
promptPath := filepath.Join(filepath.Dir(bookPath), "prompts", "translator.md")
|
||
writeFile(t, promptPath, "НОВЫЙ промпт с {{source_lang}}.\n---USER---\n{{text}}")
|
||
|
||
r2, err := NewRunner(bookPath, obs.NewLogger())
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
_, err = r2.TranslateOneChunk(ctx)
|
||
r2.Close()
|
||
if err == nil || !strings.Contains(err.Error(), "resnapshot") {
|
||
t.Fatalf("changed config must fail loud mentioning --resnapshot, got: %v", err)
|
||
}
|
||
if calls.Load() != 2 {
|
||
t.Fatalf("denied resume must not call the provider, calls=%d", calls.Load())
|
||
}
|
||
|
||
// Явное согласие: джобы перепривязываются, вызовы повторяются и оплачиваются.
|
||
r3, err := NewRunner(bookPath, obs.NewLogger())
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
defer r3.Close()
|
||
r3.Resnapshot = true
|
||
res, err := r3.TranslateOneChunk(ctx)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if calls.Load() != 4 {
|
||
t.Fatalf("resnapshot run must re-call both stages, calls=%d", calls.Load())
|
||
}
|
||
if res.TotalUSD <= 0 {
|
||
t.Fatal("re-translation must be billed")
|
||
}
|
||
}
|
||
|
||
// D5.2/B: правка ТОЛЬКО каппы модели (wire-форма) обязана сдвинуть snapshot и
|
||
// уронить resume громко — иначе изменённое тело запроса подалось бы из старого
|
||
// чекпоинта (тихий расходящийся ре-пэй). Гейт на то, что stageSnap реально
|
||
// сворачивает резолвнутую Capability: без `ss.Capability` этот тест бы прошёл
|
||
// resume молча (находка селфревью вехи 1).
|
||
func TestRunnerSnapshotPinsCapability(t *testing.T) {
|
||
var calls atomic.Int32
|
||
srv := newFakeProvider(t, &calls)
|
||
defer srv.Close()
|
||
bookPath := setupProject(t, srv.URL)
|
||
ctx := context.Background()
|
||
|
||
r1, err := NewRunner(bookPath, obs.NewLogger())
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := r1.TranslateOneChunk(ctx); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
r1.Close()
|
||
if calls.Load() != 2 {
|
||
t.Fatalf("run1 calls = %d", calls.Load())
|
||
}
|
||
|
||
// Меняем ТОЛЬКО каппу fake-model (temperature force) — промпт/сэмплинг/
|
||
// провайдер/цены те же. Резолвнутая Capability меняется → snapshotID обязан
|
||
// сдвинуться → resume падает на snapshot-pinning, а не подаёт чекпоинт.
|
||
modelsPath := filepath.Join(filepath.Dir(bookPath), "models.yaml")
|
||
raw, err := os.ReadFile(modelsPath)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
patched := strings.Replace(string(raw),
|
||
"output_per_m: 2.0 }",
|
||
"output_per_m: 2.0 }\n capabilities: { temperature: { mode: force, value: 0.9 } }", 1)
|
||
if patched == string(raw) {
|
||
t.Fatal("failed to inject capability into models.yaml")
|
||
}
|
||
writeFile(t, modelsPath, patched)
|
||
|
||
r2, err := NewRunner(bookPath, obs.NewLogger())
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
_, err = r2.TranslateOneChunk(ctx)
|
||
r2.Close()
|
||
if err == nil || !strings.Contains(err.Error(), "resnapshot") {
|
||
t.Fatalf("a capability edit must fail loud mentioning --resnapshot (snapshot must fold the resolved capability), got: %v", err)
|
||
}
|
||
if calls.Load() != 2 {
|
||
t.Fatalf("denied resume must not call the provider, calls=%d", calls.Load())
|
||
}
|
||
}
|
||
|
||
// Платный 2xx с нулевым usage не должен селтлиться в $0 (иначе потолок слепнет):
|
||
// берётся консервативная оценка резерва.
|
||
func TestRunnerZeroUsagePaidSettlesEstimate(t *testing.T) {
|
||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||
fmt.Fprint(w, `{"id":"z","model":"deepseek-v4-flash","choices":[{"message":{"content":"перевод"},"finish_reason":"stop"}],
|
||
"usage":{"prompt_tokens":0,"completion_tokens":0}}`)
|
||
}))
|
||
defer srv.Close()
|
||
bookPath := setupProject(t, srv.URL)
|
||
|
||
r, err := NewRunner(bookPath, obs.NewLogger())
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
defer r.Close()
|
||
res, err := r.TranslateOneChunk(context.Background())
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if res.TotalUSD <= 0 {
|
||
t.Fatalf("zero-usage paid 2xx must settle a non-zero estimate, got $%.6f", res.TotalUSD)
|
||
}
|
||
committed, _, err := r.Store.SpentUSD("test-book")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if committed != res.TotalUSD {
|
||
t.Fatalf("committed %v != total %v", committed, res.TotalUSD)
|
||
}
|
||
}
|
||
|
||
// NewRunner обязан отклонить конфиг с нереализованной механикой ДО открытия
|
||
// store и любых вызовов (guard границы «конфиг vs код»).
|
||
func TestRunnerRejectsUnrunnableConfig(t *testing.T) {
|
||
var calls atomic.Int32
|
||
srv := newFakeProvider(t, &calls)
|
||
defer srv.Close()
|
||
bookPath := setupProject(t, srv.URL)
|
||
|
||
// Ломаем pipeline на fanout>1 (механика Фазы 2).
|
||
pipePath := filepath.Join(filepath.Dir(bookPath), "pipeline.yaml")
|
||
raw, err := os.ReadFile(pipePath)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
writeFile(t, pipePath, string(raw)+"\nfanout: { candidates: 3 }\n")
|
||
|
||
if _, err := NewRunner(bookPath, obs.NewLogger()); err == nil {
|
||
t.Fatal("fanout.candidates>1 must be rejected by NewRunner (Phase-2 mechanics)")
|
||
}
|
||
if calls.Load() != 0 {
|
||
t.Fatalf("rejected config must not reach the provider, calls=%d", calls.Load())
|
||
}
|
||
}
|
||
|
||
func TestRunnerCeilingDenies(t *testing.T) {
|
||
var calls atomic.Int32
|
||
srv := newFakeProvider(t, &calls)
|
||
defer srv.Close()
|
||
bookPath := setupProject(t, srv.URL)
|
||
|
||
// Потолок $0: первый же reserve обязан отказать ДО вызова провайдера.
|
||
raw, err := os.ReadFile(bookPath)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
patched := strings.Replace(string(raw), "book_usd: 1.0", "book_usd: 0.0000001", 1)
|
||
writeFile(t, bookPath, patched)
|
||
|
||
r, err := NewRunner(bookPath, obs.NewLogger())
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
defer r.Close()
|
||
_, err = r.TranslateOneChunk(context.Background())
|
||
if err == nil {
|
||
t.Fatal("ceiling must deny the run")
|
||
}
|
||
if calls.Load() != 0 {
|
||
t.Fatalf("denied reserve must not reach the provider, calls=%d", calls.Load())
|
||
}
|
||
}
|