package processing import ( "context" "fmt" "os" "testing" "time" "github.com/google/uuid" "github.com/jackc/pgx/v5/pgxpool" ) func TestStartJobAutoSplitsAndCopyInserts(t *testing.T) { dsn := os.Getenv("DATABASE_URL") if dsn == "" { t.Skip("DATABASE_URL not set") } ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() pg, err := pgxpool.New(ctx, dsn) if err != nil { t.Fatal(err) } defer pg.Close() var userID uuid.UUID err = pg.QueryRow(ctx, `SELECT id FROM users ORDER BY created_at DESC LIMIT 1`).Scan(&userID) if errorsIsNoRows(err) { t.Skip("no users rows available") } if err != nil { t.Fatal(err) } companyID := uuid.New() if _, err := pg.Exec(ctx, `INSERT INTO companies (id, name) VALUES ($1, $2)`, companyID, "start-split-test"); err != nil { t.Fatal(err) } defer func() { _, _ = pg.Exec(context.Background(), `DELETE FROM processing_job_products WHERE job_id IN (SELECT id FROM processing_jobs WHERE company_id = $1)`, companyID) _, _ = pg.Exec(context.Background(), `DELETE FROM processing_jobs WHERE company_id = $1`, companyID) _, _ = pg.Exec(context.Background(), `DELETE FROM raw_products WHERE company_id = $1`, companyID) _, _ = pg.Exec(context.Background(), `DELETE FROM companies WHERE id = $1`, companyID) }() rawIDs := make([]uuid.UUID, 5) for i := range rawIDs { rawIDs[i] = uuid.New() gtin := fmt.Sprintf("split-test-%d-%s", i, companyID.String()[:8]) if _, err := pg.Exec(ctx, ` INSERT INTO raw_products (id, company_id, gtin, raw_data, mapped_data, is_processed, processing_status) VALUES ($1, $2, $3, '{}'::jsonb, '{}'::jsonb, false, 'unprocessed')`, rawIDs[i], companyID, gtin); err != nil { t.Fatal(err) } } testMaxJobProducts = 2 defer func() { testMaxJobProducts = 0 }() p := NewPipeline(pg) p.Billing = nil p.Limiter = nil jobs, err := p.StartJob(ctx, companyID, userID, rawIDs, "full") if err != nil { t.Fatal(err) } if len(jobs) != 3 { t.Fatalf("jobs=%d want 3 (5 products / cap 2)", len(jobs)) } totals := 0 for i, job := range jobs { want := 2 if i == 2 { want = 1 } if job.TotalProducts != want { t.Fatalf("job[%d].TotalProducts=%d want %d", i, job.TotalProducts, want) } if job.Status != "pending" { t.Fatalf("job[%d].Status=%q", i, job.Status) } totals += job.TotalProducts var n int if err := pg.QueryRow(ctx, ` SELECT count(*) FROM processing_job_products WHERE job_id = $1 AND status = 'pending'`, job.ID).Scan(&n); err != nil { t.Fatal(err) } if n != want { t.Fatalf("job[%d] product rows=%d want %d", i, n, want) } } if totals != len(rawIDs) { t.Fatalf("total products=%d want %d", totals, len(rawIDs)) } }