batch_image_download_test.go 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299
  1. //go:build unit
  2. package service
  3. import (
  4. "archive/zip"
  5. "bytes"
  6. "context"
  7. "encoding/json"
  8. "io"
  9. "strings"
  10. "testing"
  11. "time"
  12. "github.com/Wei-Shaw/sub2api/internal/config"
  13. "github.com/stretchr/testify/require"
  14. )
  15. func TestBatchImageDownloadService_OpenItemContent(t *testing.T) {
  16. ctx := context.Background()
  17. t.Run("streams image bytes with safe headers data", func(t *testing.T) {
  18. svc, _, limiter := newTestBatchImageDownloadService()
  19. stream, err := svc.OpenItemContent(ctx, testBatchImageOwner(), "imgbatch_download", "cover/../001", 1)
  20. require.NoError(t, err)
  21. defer stream.Reader.Close()
  22. body, err := io.ReadAll(stream.Reader)
  23. require.NoError(t, err)
  24. require.Equal(t, []byte("second"), body)
  25. require.Equal(t, "image/jpeg", stream.ContentType)
  26. require.Equal(t, "cover___001.jpg", stream.Filename)
  27. require.Equal(t, 1, limiter.acquireCount)
  28. require.Zero(t, limiter.releaseCount)
  29. require.NoError(t, stream.Reader.Close())
  30. require.Equal(t, 1, limiter.releaseCount)
  31. })
  32. tests := []struct {
  33. name string
  34. mutate func(*fakeBatchImageRepository)
  35. id string
  36. item string
  37. index int
  38. want error
  39. }{
  40. {name: "non_owner", id: "imgbatch_download", item: "cover/../001", mutate: func(r *fakeBatchImageRepository) {
  41. v := int64(999)
  42. r.jobs["imgbatch_download"].APIKeyID = &v
  43. }, want: ErrBatchImageJobNotFound},
  44. {name: "not_completed", id: "imgbatch_download", item: "cover/../001", mutate: func(r *fakeBatchImageRepository) {
  45. r.jobs["imgbatch_download"].Status = BatchImageJobStatusRunning
  46. }, want: ErrBatchImageNotReady},
  47. {name: "output_deleted", id: "imgbatch_download", item: "cover/../001", mutate: func(r *fakeBatchImageRepository) {
  48. r.jobs["imgbatch_download"].Status = BatchImageJobStatusOutputDeleted
  49. }, want: ErrBatchImageOutputDeleted},
  50. {name: "missing_item", id: "imgbatch_download", item: "missing", want: ErrBatchImageItemNotFound},
  51. {name: "failed_item", id: "imgbatch_download", item: "bad", want: ErrBatchImageItemFailed},
  52. {name: "out_of_range", id: "imgbatch_download", item: "cover/../001", index: 2, want: ErrBatchImageItemImageIndexOutOfRange},
  53. }
  54. for _, tt := range tests {
  55. t.Run(tt.name, func(t *testing.T) {
  56. svc, repo, _ := newTestBatchImageDownloadService()
  57. if tt.mutate != nil {
  58. tt.mutate(repo)
  59. }
  60. got, err := svc.OpenItemContent(ctx, testBatchImageOwner(), tt.id, tt.item, tt.index)
  61. require.Nil(t, got)
  62. require.ErrorIs(t, err, tt.want)
  63. require.NotContains(t, err.Error(), batchImageDownloadTestBase64)
  64. require.NotContains(t, err.Error(), "providers/")
  65. require.NotContains(t, err.Error(), "gs://")
  66. })
  67. }
  68. }
  69. func TestBatchImageDownloadService_StreamZip(t *testing.T) {
  70. ctx := context.Background()
  71. t.Run("streams zip with images manifest and errors", func(t *testing.T) {
  72. svc, _, limiter := newTestBatchImageDownloadService()
  73. var buf bytes.Buffer
  74. result, err := svc.StreamZip(ctx, testBatchImageOwner(), "imgbatch_download", BatchImageZipOptions{}, &buf)
  75. require.NoError(t, err)
  76. require.Equal(t, 3, result.FileCount)
  77. require.Equal(t, 1, limiter.acquireCount)
  78. require.Equal(t, 1, limiter.releaseCount)
  79. files := readZipFiles(t, buf.Bytes())
  80. require.Equal(t, []byte("first"), files["images/cover___001.png"])
  81. require.Equal(t, []byte("second"), files["images/cover___001_2.jpg"])
  82. require.Equal(t, []byte("third"), files["images/ok_2.webp"])
  83. require.Contains(t, files, "manifest.json")
  84. require.Contains(t, files, "errors.json")
  85. zipText := string(bytes.Join(mapValues(files), []byte("\n")))
  86. require.NotContains(t, zipText, batchImageDownloadTestBase64)
  87. require.NotContains(t, zipText, "provider_job_name")
  88. require.NotContains(t, zipText, "provider_input_ref")
  89. require.NotContains(t, zipText, "gcs_output_uri")
  90. require.NotContains(t, zipText, "account_id")
  91. require.NotContains(t, zipText, "providers/")
  92. require.NotContains(t, zipText, "gs://")
  93. var manifest struct {
  94. Files []struct {
  95. CustomID string `json:"custom_id"`
  96. Filename string `json:"filename"`
  97. MimeType string `json:"mime_type"`
  98. ImageIndex int `json:"image_index"`
  99. } `json:"files"`
  100. }
  101. require.NoError(t, json.Unmarshal(files["manifest.json"], &manifest))
  102. require.Len(t, manifest.Files, 3)
  103. require.Equal(t, "images/cover___001_2.jpg", manifest.Files[1].Filename)
  104. require.Equal(t, 1, manifest.Files[1].ImageIndex)
  105. var errorsJSON []map[string]string
  106. require.NoError(t, json.Unmarshal(files["errors.json"], &errorsJSON))
  107. require.Len(t, errorsJSON, 1)
  108. require.Equal(t, "bad", errorsJSON[0]["custom_id"])
  109. require.Equal(t, "SAFETY_BLOCKED", errorsJSON[0]["code"])
  110. })
  111. t.Run("limiter denial returns public limit error", func(t *testing.T) {
  112. svc, _, limiter := newTestBatchImageDownloadService()
  113. limiter.deny = true
  114. var buf bytes.Buffer
  115. result, err := svc.StreamZip(ctx, testBatchImageOwner(), "imgbatch_download", BatchImageZipOptions{}, &buf)
  116. require.Nil(t, result)
  117. require.ErrorIs(t, err, ErrBatchImageDownloadLimited)
  118. require.Empty(t, buf.Bytes())
  119. })
  120. t.Run("rejects too many zip items before opening output", func(t *testing.T) {
  121. svc, repo, _ := newTestBatchImageDownloadService()
  122. repo.jobs["imgbatch_download"].SuccessCount = 3
  123. svc.Config.BatchImage.MaxDownloadItemsZip = 1
  124. var buf bytes.Buffer
  125. result, err := svc.StreamZip(ctx, testBatchImageOwner(), "imgbatch_download", BatchImageZipOptions{}, &buf)
  126. require.Nil(t, result)
  127. require.ErrorIs(t, err, ErrBatchImageZipTooManyItems)
  128. require.Empty(t, buf.Bytes())
  129. })
  130. }
  131. func TestExtractBatchImagePartsFromResultLine(t *testing.T) {
  132. tests := []struct {
  133. name string
  134. line string
  135. wantID string
  136. wantMime string
  137. wantError string
  138. }{
  139. {name: "inlineData_mimeType_response", line: `{"key":"a","response":{"candidates":[{"content":{"parts":[{"inlineData":{"mimeType":"image/png","data":"` + batchImageDownloadTestBase64 + `"}}]}}]}}`, wantID: "a", wantMime: "image/png"},
  140. {name: "inline_data_mime_type_top_level", line: `{"custom_id":"b","candidates":[{"content":{"parts":[{"inline_data":{"mime_type":"image/jpeg","data":"` + batchImageDownloadTestBase64 + `"}}]}}]}`, wantID: "b", wantMime: "image/jpeg"},
  141. {name: "status_failure", line: `{"key":"c","status":{"code":"INVALID_ARGUMENT","message":"bad prompt"}}`, wantID: "c", wantError: "INVALID_ARGUMENT"},
  142. {name: "error_failure", line: `{"key":"d","error":{"code":"SAFETY","message":"blocked"}}`, wantID: "d", wantError: "SAFETY_BLOCKED"},
  143. {name: "empty_output", line: `{"key":"e","response":{"candidates":[{"content":{"parts":[{"text":"none"}]}}]}}`, wantID: "e", wantError: "EMPTY_IMAGE_OUTPUT"},
  144. }
  145. for _, tt := range tests {
  146. t.Run(tt.name, func(t *testing.T) {
  147. got, err := ExtractBatchImagePartsFromResultLine([]byte(tt.line))
  148. require.NoError(t, err)
  149. require.Equal(t, tt.wantID, got.CustomID)
  150. if tt.wantMime != "" {
  151. require.Len(t, got.Images, 1)
  152. require.Equal(t, tt.wantMime, got.Images[0].MimeType)
  153. require.NotEmpty(t, got.Images[0].Base64Data)
  154. }
  155. if tt.wantError != "" {
  156. require.Equal(t, tt.wantError, got.ErrorCode)
  157. }
  158. })
  159. }
  160. _, err := ExtractBatchImagePartsFromResultLine([]byte(`{"response":{"candidates":[{"content":{"parts":[{"inlineData":{"mimeType":"image/png","data":"` + batchImageDownloadTestBase64 + `"}}]}}]}}`))
  161. require.Error(t, err)
  162. require.NotContains(t, err.Error(), batchImageDownloadTestBase64)
  163. }
  164. func TestBatchImageDownloadFilenames(t *testing.T) {
  165. require.Equal(t, "___secret_name.png", BatchImageSafeDownloadFilename("../../secret\nname", "png"))
  166. require.Equal(t, `attachment; filename="cover_001.png"`, BatchImageContentDispositionAttachment(`cover"001.png`))
  167. }
  168. func newTestBatchImageDownloadService() (*BatchImageDownloadService, *fakeBatchImageRepository, *fakeBatchImageDownloadLimiter) {
  169. repo := newFakeBatchImageRepository()
  170. apiKeyID := int64(22)
  171. accountID := int64(101)
  172. repo.jobs["imgbatch_download"] = &BatchImageJob{
  173. BatchID: "imgbatch_download",
  174. UserID: 11,
  175. APIKeyID: &apiKeyID,
  176. AccountID: &accountID,
  177. Provider: BatchImageProviderGeminiAPI,
  178. Model: "gemini-2.5-flash-image",
  179. Status: BatchImageJobStatusCompleted,
  180. ProviderJobName: batchImageStringPtr("providers/internal/job"),
  181. ProviderOutputRef: batchImageStringPtr("gs://bucket/internal/output.jsonl"),
  182. ItemCount: 3,
  183. SuccessCount: 2,
  184. FailCount: 1,
  185. CreatedAt: time.Now(),
  186. }
  187. mime := "image/png"
  188. ext := "png"
  189. webp := "image/webp"
  190. webpExt := "webp"
  191. code := "SAFETY_BLOCKED"
  192. msg := "blocked in gs://bucket/internal/output.jsonl"
  193. repo.items["imgbatch_download"] = []CreateBatchImageItemParams{
  194. {JobID: "imgbatch_download", CustomID: "cover/../001", Status: BatchImageItemStatusSuccess, MimeType: &mime, FileExtension: &ext, ImageCount: 2},
  195. {JobID: "imgbatch_download", CustomID: "bad", Status: BatchImageItemStatusFailed, ErrorCode: &code, ErrorMessage: &msg},
  196. {JobID: "imgbatch_download", CustomID: "ok_2", Status: BatchImageItemStatusSuccess, MimeType: &webp, FileExtension: &webpExt, ImageCount: 1},
  197. }
  198. provider := &publicBatchImageProvider{name: BatchImageProviderGeminiAPI, result: batchImageDownloadResultJSONL()}
  199. limiter := &fakeBatchImageDownloadLimiter{}
  200. svc := &BatchImageDownloadService{
  201. Repo: repo,
  202. ProviderRegistry: NewBatchImageProviderRegistry(provider),
  203. AccountResolver: &fakeBatchImageAccountResolver{account: &Account{ID: accountID, Platform: PlatformGemini, Type: AccountTypeAPIKey, Status: StatusActive, Schedulable: true}},
  204. Limiter: limiter,
  205. Config: &config.Config{BatchImage: config.BatchImageConfig{MaxDownloadItemsZip: 10, MaxDownloadDurationSeconds: 60}},
  206. }
  207. return svc, repo, limiter
  208. }
  209. const batchImageDownloadTestBase64 = "Zmlyc3Q="
  210. func batchImageDownloadResultJSONL() string {
  211. return strings.Join([]string{
  212. `{"key":"cover/../001","response":{"candidates":[{"content":{"parts":[{"inlineData":{"mimeType":"image/png","data":"Zmlyc3Q="}},{"inlineData":{"mimeType":"image/jpeg","data":"c2Vjb25k"}}]}}]}}`,
  213. `{"key":"bad","error":{"code":"SAFETY","message":"blocked"}}`,
  214. `{"key":"ok_2","candidates":[{"content":{"parts":[{"inline_data":{"mime_type":"image/webp","data":"dGhpcmQ="}}]}}]}`,
  215. }, "\n") + "\n"
  216. }
  217. func readZipFiles(t *testing.T, data []byte) map[string][]byte {
  218. t.Helper()
  219. reader, err := zip.NewReader(bytes.NewReader(data), int64(len(data)))
  220. require.NoError(t, err)
  221. out := make(map[string][]byte, len(reader.File))
  222. for _, file := range reader.File {
  223. rc, err := file.Open()
  224. require.NoError(t, err)
  225. body, err := io.ReadAll(rc)
  226. require.NoError(t, err)
  227. require.NoError(t, rc.Close())
  228. out[file.Name] = body
  229. }
  230. return out
  231. }
  232. func mapValues(in map[string][]byte) [][]byte {
  233. out := make([][]byte, 0, len(in))
  234. for _, value := range in {
  235. out = append(out, value)
  236. }
  237. return out
  238. }
  239. type fakeBatchImageDownloadLimiter struct {
  240. acquireCount int
  241. releaseCount int
  242. deny bool
  243. }
  244. func (l *fakeBatchImageDownloadLimiter) Acquire(context.Context, string, string) (BatchImageDownloadPermit, error) {
  245. l.acquireCount++
  246. if l.deny {
  247. return nil, ErrBatchImageDownloadLimited
  248. }
  249. return &fakeBatchImageDownloadPermit{release: func() { l.releaseCount++ }}, nil
  250. }
  251. type fakeBatchImageDownloadPermit struct {
  252. once bool
  253. release func()
  254. }
  255. func (p *fakeBatchImageDownloadPermit) Release(context.Context) error {
  256. if p.once {
  257. return nil
  258. }
  259. p.once = true
  260. if p.release != nil {
  261. p.release()
  262. }
  263. return nil
  264. }
  265. var _ BatchImageDownloadLimiter = (*fakeBatchImageDownloadLimiter)(nil)
  266. var _ BatchImageDownloadPermit = (*fakeBatchImageDownloadPermit)(nil)