iu
ec8d907565
- introduce storage.Backend with S3 and local file-system implementations - deliver all blobs via signed GET/HEAD /api/objects/<key> (Range, Content-Type) - stream S3 objects through the backend; remove presigned URL path - share shared HMAC Signer between backends - remove S3_PUBLIC_URL; rename S3_PRESIGN_TTL to OBJECT_URL_TTL - make storage label in health dynamic; nginx proxy_buffering off for objects
93 lines
2.5 KiB
Go
93 lines
2.5 KiB
Go
package photos
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
|
|
"shared-album/internal/storage"
|
|
)
|
|
|
|
func TestWorkerGeneratesVariantsAndCollects(t *testing.T) {
|
|
pool := testPool(t)
|
|
svc, cl := newTestService(t, pool)
|
|
runWorkerFlow(t, pool, svc, cl)
|
|
}
|
|
|
|
func TestWorkerGeneratesVariantsWithFS(t *testing.T) {
|
|
pool := testPool(t)
|
|
svc, cl := newTestServiceFS(t, pool)
|
|
runWorkerFlow(t, pool, svc, cl)
|
|
}
|
|
|
|
func runWorkerFlow(t *testing.T, pool *pgxpool.Pool, svc *Service, cl storage.Backend) {
|
|
t.Helper()
|
|
ctx := context.Background()
|
|
user := createUser(t, ctx, pool)
|
|
album := createAlbum(t, ctx, pool, user)
|
|
|
|
p, err := svc.Upload(ctx, UploadInput{
|
|
AlbumID: album,
|
|
UserID: user,
|
|
Filename: "photo.png",
|
|
MimeType: "image/png",
|
|
Data: makePNG(t),
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("upload: %v", err)
|
|
}
|
|
t.Cleanup(func() { cleanupPhoto(t, ctx, pool, cl, p.ID) })
|
|
|
|
w := NewWorker(svc, time.Second, 10, time.Minute)
|
|
w.processVariantJobs(ctx)
|
|
|
|
refreshed, err := svc.store.GetByID(ctx, p.ID, "")
|
|
if err != nil {
|
|
t.Fatalf("get refreshed: %v", err)
|
|
}
|
|
if refreshed.ThumbKey == "" || refreshed.PreviewKey == "" {
|
|
t.Fatalf("variants not bound: thumb=%q preview=%q", refreshed.ThumbKey, refreshed.PreviewKey)
|
|
}
|
|
if err := svc.attachURLs(ctx, refreshed); err != nil {
|
|
t.Fatalf("attach urls: %v", err)
|
|
}
|
|
if refreshed.ThumbURL == "" || refreshed.PreviewURL == "" {
|
|
t.Fatalf("variant urls not set")
|
|
}
|
|
|
|
var jobs int
|
|
if err := pool.QueryRow(ctx, `SELECT count(*) FROM variant_jobs WHERE photo_id = $1`, p.ID).Scan(&jobs); err != nil {
|
|
t.Fatalf("count jobs: %v", err)
|
|
}
|
|
if jobs != 0 {
|
|
t.Fatalf("variant job not completed: %d left", jobs)
|
|
}
|
|
|
|
objectIDs, err := svc.store.PhotoObjectIDs(ctx, p.ID)
|
|
if err != nil || len(objectIDs) != 3 {
|
|
t.Fatalf("expected 3 bound objects, got %v err=%v", objectIDs, err)
|
|
}
|
|
|
|
if err := svc.PermanentDelete(ctx, p.ID, user); err != nil {
|
|
t.Fatalf("permanent delete: %v", err)
|
|
}
|
|
// Run the collector until the objects are gone (bounded to avoid hangs).
|
|
for attempt := 0; attempt < 100 && countObjects(t, ctx, pool, objectIDs) > 0; attempt++ {
|
|
w.collectGarbage(ctx)
|
|
}
|
|
if n := countObjects(t, ctx, pool, objectIDs); n != 0 {
|
|
t.Fatalf("%d objects were not collected", n)
|
|
}
|
|
}
|
|
|
|
func countObjects(t *testing.T, ctx context.Context, pool *pgxpool.Pool, ids []string) int {
|
|
t.Helper()
|
|
var n int
|
|
if err := pool.QueryRow(ctx, `SELECT count(*) FROM objects WHERE id = ANY($1)`, ids).Scan(&n); err != nil {
|
|
t.Fatalf("count objects: %v", err)
|
|
}
|
|
return n
|
|
}
|