mirror of
https://github.com/snapotter-hq/SnapOtter.git
synced 2026-08-03 07:46:42 +02:00
test: reorganize flat test files into purpose-based subdirectories (phase 6)
Group 245 flat integration tests and 25 loose unit tests into
discoverable subdirectories per spec section 6:
integration/tools/{image,video,audio,document,data}/ (156 files)
integration/platform/ (64 files)
integration/generated/ (14 files)
integration/security/ (10 files)
unit/security/ (8 files, new subdir)
unit/api/ (7 files moved in)
unit/web/ (3 files moved in)
unit/shared/ (6 files moved in)
unit/image-engine/ (1 file moved in)
All moves via git mv (history preserved). Relative imports repaired
for both depth levels (platform/generated/security = +1, tools/ = +2):
static from-imports, dynamic import() calls, vi.mock() paths,
import.meta.dirname joins, and __dirname joins.
Vitest discovery unchanged (no test.include in config, recursive glob
matches subdirs, shard-by-hash unaffected). test-server.ts and
tool-route-drift.test.ts stay at integration root. fixtures/ untouched.
Parity gate: 13189 passing test names before = 13189 after (0 dropped).
This commit is contained in:
SnapOtter
@@ -0,0 +1,657 @@
|
||||
/**
|
||||
* Integration tests for the SSE progress tracking system.
|
||||
*
|
||||
* The GET /api/v1/jobs/:jobId/progress endpoint uses reply.hijack()
|
||||
* for SSE streaming, which makes it incompatible with Fastify's inject()
|
||||
* (inject never completes for hijacked responses that wait for events).
|
||||
*
|
||||
* Instead, we test progress tracking indirectly through batch and pipeline
|
||||
* batch routes that drive updateJobProgress() and verify:
|
||||
* - X-Job-Id header presence (job was tracked)
|
||||
* - clientJobId passthrough (custom IDs are used)
|
||||
* - Completed batch with progress tracking
|
||||
* - Pipeline batch progress tracking
|
||||
* - Job persistence in the database
|
||||
*/
|
||||
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { db, schema } from "../../../apps/api/src/db/index.js";
|
||||
import {
|
||||
updateJobProgress,
|
||||
updateSingleFileProgress,
|
||||
} from "../../../apps/api/src/routes/progress.js";
|
||||
import { fixtures, readFixture } from "../../fixtures/index.js";
|
||||
import {
|
||||
buildTestApp,
|
||||
createMultipartPayload,
|
||||
loginAsAdmin,
|
||||
type TestApp,
|
||||
} from "../test-server.js";
|
||||
|
||||
const PNG = readFixture(fixtures.image.base.png200);
|
||||
const JPG = readFixture(fixtures.image.base.jpg100);
|
||||
|
||||
let testApp: TestApp;
|
||||
let app: TestApp["app"];
|
||||
let adminToken: string;
|
||||
|
||||
beforeAll(async () => {
|
||||
testApp = await buildTestApp();
|
||||
app = testApp.app;
|
||||
adminToken = await loginAsAdmin(app);
|
||||
}, 30_000);
|
||||
|
||||
afterAll(async () => {
|
||||
await testApp.cleanup();
|
||||
}, 10_000);
|
||||
|
||||
// ── Batch progress tracking via X-Job-Id ────────────────────────
|
||||
|
||||
// Fire-and-forget persist calls need time to flush to the DB.
|
||||
// With Postgres, async writes involve network round-trips that may exceed a
|
||||
// fixed delay. Poll for the expected terminal status with a generous ceiling.
|
||||
const flushPersist = async (
|
||||
jobId?: string,
|
||||
terminalStatuses: string[] = ["completed", "failed"],
|
||||
maxMs = 2000,
|
||||
) => {
|
||||
if (!jobId) {
|
||||
// Fallback: fixed delay when no jobId is available
|
||||
await new Promise((r) => setTimeout(r, 200));
|
||||
return;
|
||||
}
|
||||
const start = Date.now();
|
||||
while (Date.now() - start < maxMs) {
|
||||
const [row] = await db.select().from(schema.jobs).where(eq(schema.jobs.id, jobId));
|
||||
if (row && terminalStatuses.includes(row.status)) return;
|
||||
await new Promise((r) => setTimeout(r, 50));
|
||||
}
|
||||
};
|
||||
|
||||
describe("Batch progress tracking", () => {
|
||||
it("assigns a job ID to batch operations", async () => {
|
||||
const { body, contentType } = createMultipartPayload([
|
||||
{ name: "file", filename: "a.png", contentType: "image/png", content: PNG },
|
||||
{ name: "file", filename: "b.jpg", contentType: "image/jpeg", content: JPG },
|
||||
{ name: "settings", content: JSON.stringify({ width: 50 }) },
|
||||
]);
|
||||
|
||||
const res = await app.inject({
|
||||
method: "POST",
|
||||
url: "/api/v1/tools/resize/batch",
|
||||
headers: {
|
||||
"content-type": contentType,
|
||||
authorization: `Bearer ${adminToken}`,
|
||||
},
|
||||
body,
|
||||
});
|
||||
|
||||
expect(res.statusCode).toBe(200);
|
||||
const jobId = res.headers["x-job-id"] as string;
|
||||
expect(jobId).toBeDefined();
|
||||
expect(jobId.length).toBeGreaterThan(0);
|
||||
});
|
||||
|
||||
it("uses client-provided job ID when clientJobId is supplied", async () => {
|
||||
const clientJobId = randomUUID();
|
||||
|
||||
const { body, contentType } = createMultipartPayload([
|
||||
{ name: "file", filename: "track.png", contentType: "image/png", content: PNG },
|
||||
{ name: "settings", content: JSON.stringify({ width: 80 }) },
|
||||
{ name: "clientJobId", content: clientJobId },
|
||||
]);
|
||||
|
||||
const res = await app.inject({
|
||||
method: "POST",
|
||||
url: "/api/v1/tools/resize/batch",
|
||||
headers: {
|
||||
"content-type": contentType,
|
||||
authorization: `Bearer ${adminToken}`,
|
||||
},
|
||||
body,
|
||||
});
|
||||
|
||||
expect(res.statusCode).toBe(200);
|
||||
expect(res.headers["x-job-id"]).toBe(clientJobId);
|
||||
});
|
||||
|
||||
it("persists job progress to the database after batch completes", async () => {
|
||||
const clientJobId = randomUUID();
|
||||
|
||||
const { body, contentType } = createMultipartPayload([
|
||||
{ name: "file", filename: "persist.png", contentType: "image/png", content: PNG },
|
||||
{ name: "file", filename: "persist2.jpg", contentType: "image/jpeg", content: JPG },
|
||||
{ name: "settings", content: JSON.stringify({ width: 60 }) },
|
||||
{ name: "clientJobId", content: clientJobId },
|
||||
]);
|
||||
|
||||
const res = await app.inject({
|
||||
method: "POST",
|
||||
url: "/api/v1/tools/resize/batch",
|
||||
headers: {
|
||||
"content-type": contentType,
|
||||
authorization: `Bearer ${adminToken}`,
|
||||
},
|
||||
body,
|
||||
});
|
||||
|
||||
expect(res.statusCode).toBe(200);
|
||||
await flushPersist(clientJobId);
|
||||
|
||||
// Check the jobs table for the persisted progress
|
||||
const [job] = await db.select().from(schema.jobs).where(eq(schema.jobs.id, clientJobId));
|
||||
|
||||
expect(job).toBeDefined();
|
||||
expect(job?.status).toBe("completed");
|
||||
expect((job?.progress as { percent: number })?.percent).toBe(100);
|
||||
expect(job?.completedAt).not.toBeNull();
|
||||
});
|
||||
|
||||
it("persists failed job status to the database", async () => {
|
||||
// Provide an invalid file (empty buffer won't be uploaded, so all fail)
|
||||
const clientJobId = randomUUID();
|
||||
|
||||
const { body, contentType } = createMultipartPayload([
|
||||
{
|
||||
name: "file",
|
||||
filename: "bad.txt",
|
||||
contentType: "text/plain",
|
||||
content: Buffer.from("not an image"),
|
||||
},
|
||||
{ name: "settings", content: JSON.stringify({ width: 50 }) },
|
||||
{ name: "clientJobId", content: clientJobId },
|
||||
]);
|
||||
|
||||
const res = await app.inject({
|
||||
method: "POST",
|
||||
url: "/api/v1/tools/resize/batch",
|
||||
headers: {
|
||||
"content-type": contentType,
|
||||
authorization: `Bearer ${adminToken}`,
|
||||
},
|
||||
body,
|
||||
});
|
||||
|
||||
// Should fail (422 = all files failed)
|
||||
expect(res.statusCode).toBe(422);
|
||||
|
||||
await flushPersist(clientJobId);
|
||||
const [job] = await db.select().from(schema.jobs).where(eq(schema.jobs.id, clientJobId));
|
||||
|
||||
expect(job).toBeDefined();
|
||||
expect(job?.status).toBe("failed");
|
||||
});
|
||||
|
||||
it("tracks progress for multi-file batch with partial success", async () => {
|
||||
const clientJobId = randomUUID();
|
||||
|
||||
// Mix valid image + invalid data -- partial success
|
||||
const { body, contentType } = createMultipartPayload([
|
||||
{ name: "file", filename: "good.png", contentType: "image/png", content: PNG },
|
||||
{
|
||||
name: "file",
|
||||
filename: "bad.txt",
|
||||
contentType: "text/plain",
|
||||
content: Buffer.from("not an image"),
|
||||
},
|
||||
{ name: "settings", content: JSON.stringify({ width: 50 }) },
|
||||
{ name: "clientJobId", content: clientJobId },
|
||||
]);
|
||||
|
||||
const res = await app.inject({
|
||||
method: "POST",
|
||||
url: "/api/v1/tools/resize/batch",
|
||||
headers: {
|
||||
"content-type": contentType,
|
||||
authorization: `Bearer ${adminToken}`,
|
||||
},
|
||||
body,
|
||||
});
|
||||
|
||||
// Should succeed (at least one file processed)
|
||||
expect(res.statusCode).toBe(200);
|
||||
|
||||
await flushPersist(clientJobId);
|
||||
const [job] = await db.select().from(schema.jobs).where(eq(schema.jobs.id, clientJobId));
|
||||
|
||||
expect(job).toBeDefined();
|
||||
expect(job?.status).toBe("completed");
|
||||
// Should have error info for the failed file
|
||||
if (job?.error) {
|
||||
const errorObj = job.error as { message: string };
|
||||
const errors = errorObj.details;
|
||||
expect(errors.length).toBeGreaterThanOrEqual(1);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
// ── Pipeline batch progress ─────────────────────────────────────
|
||||
describe("Pipeline batch progress tracking", () => {
|
||||
it("tracks progress during pipeline batch execution", async () => {
|
||||
const clientJobId = randomUUID();
|
||||
|
||||
const { body, contentType } = createMultipartPayload([
|
||||
{ name: "file", filename: "pipe1.png", contentType: "image/png", content: PNG },
|
||||
{ name: "file", filename: "pipe2.jpg", contentType: "image/jpeg", content: JPG },
|
||||
{
|
||||
name: "pipeline",
|
||||
content: JSON.stringify({
|
||||
steps: [{ toolId: "resize", settings: { width: 50 } }],
|
||||
}),
|
||||
},
|
||||
{ name: "clientJobId", content: clientJobId },
|
||||
]);
|
||||
|
||||
const res = await app.inject({
|
||||
method: "POST",
|
||||
url: "/api/v1/pipeline/batch",
|
||||
headers: {
|
||||
"content-type": contentType,
|
||||
authorization: `Bearer ${adminToken}`,
|
||||
},
|
||||
body,
|
||||
});
|
||||
|
||||
expect(res.statusCode).toBe(200);
|
||||
expect(res.headers["x-job-id"]).toBe(clientJobId);
|
||||
|
||||
// Verify DB persistence
|
||||
await flushPersist(clientJobId);
|
||||
const [job] = await db.select().from(schema.jobs).where(eq(schema.jobs.id, clientJobId));
|
||||
|
||||
expect(job).toBeDefined();
|
||||
expect(job?.status).toBe("completed");
|
||||
});
|
||||
|
||||
it("pipeline batch generates job ID when not provided", async () => {
|
||||
const { body, contentType } = createMultipartPayload([
|
||||
{ name: "file", filename: "auto.png", contentType: "image/png", content: PNG },
|
||||
{
|
||||
name: "pipeline",
|
||||
content: JSON.stringify({
|
||||
steps: [{ toolId: "rotate", settings: { angle: 90 } }],
|
||||
}),
|
||||
},
|
||||
]);
|
||||
|
||||
const res = await app.inject({
|
||||
method: "POST",
|
||||
url: "/api/v1/pipeline/batch",
|
||||
headers: {
|
||||
"content-type": contentType,
|
||||
authorization: `Bearer ${adminToken}`,
|
||||
},
|
||||
body,
|
||||
});
|
||||
|
||||
expect(res.statusCode).toBe(200);
|
||||
const jobId = res.headers["x-job-id"] as string;
|
||||
expect(jobId).toBeDefined();
|
||||
expect(jobId.length).toBeGreaterThan(0);
|
||||
});
|
||||
});
|
||||
|
||||
// ── Job DB record structure ─────────────────────────────────────
|
||||
describe("Job DB record structure", () => {
|
||||
it("persisted job contains expected fields", async () => {
|
||||
const clientJobId = randomUUID();
|
||||
|
||||
const { body, contentType } = createMultipartPayload([
|
||||
{ name: "file", filename: "fields.png", contentType: "image/png", content: PNG },
|
||||
{ name: "settings", content: JSON.stringify({ width: 70 }) },
|
||||
{ name: "clientJobId", content: clientJobId },
|
||||
]);
|
||||
|
||||
await app.inject({
|
||||
method: "POST",
|
||||
url: "/api/v1/tools/resize/batch",
|
||||
headers: {
|
||||
"content-type": contentType,
|
||||
authorization: `Bearer ${adminToken}`,
|
||||
},
|
||||
body,
|
||||
});
|
||||
|
||||
await flushPersist(clientJobId);
|
||||
const [job] = await db.select().from(schema.jobs).where(eq(schema.jobs.id, clientJobId));
|
||||
|
||||
expect(job).toBeDefined();
|
||||
expect(job?.id).toBe(clientJobId);
|
||||
expect(job?.type).toBe("batch");
|
||||
const progressObj = job?.progress as { percent: number } | null;
|
||||
expect(progressObj).not.toBeNull();
|
||||
expect(typeof progressObj?.percent).toBe("number");
|
||||
expect(progressObj?.percent).toBeGreaterThanOrEqual(0);
|
||||
expect(progressObj?.percent).toBeLessThanOrEqual(100);
|
||||
});
|
||||
});
|
||||
|
||||
// ── SSE endpoint ───────────────────────────────────────────────
|
||||
describe("SSE progress endpoint", () => {
|
||||
it("returns SSE headers when connecting to progress stream", async () => {
|
||||
const jobId = randomUUID();
|
||||
|
||||
// Publish a completed event (stored in Redis terminal key)
|
||||
updateJobProgress({
|
||||
jobId,
|
||||
status: "completed",
|
||||
totalFiles: 1,
|
||||
completedFiles: 1,
|
||||
failedFiles: 0,
|
||||
errors: [],
|
||||
});
|
||||
// Wait for Redis pub/sub + setex round trip
|
||||
await new Promise((r) => setTimeout(r, 500));
|
||||
|
||||
const res = await app.inject({
|
||||
method: "GET",
|
||||
url: `/api/v1/jobs/${jobId}/progress`,
|
||||
headers: {
|
||||
authorization: `Bearer ${adminToken}`,
|
||||
},
|
||||
});
|
||||
|
||||
// Hijacked responses return 200 (or -1 in some Fastify versions)
|
||||
// The important thing is we get SSE content back
|
||||
expect(res.statusCode).toBe(200);
|
||||
const body = res.body;
|
||||
// SSE events contain "data:" prefix
|
||||
expect(body).toContain("data:");
|
||||
|
||||
// Parse the SSE data
|
||||
const dataMatch = body.match(/data: (.+)/);
|
||||
expect(dataMatch).not.toBeNull();
|
||||
const event = JSON.parse(dataMatch?.[1]);
|
||||
expect(event.status).toBe("completed");
|
||||
expect(event.type).toBe("batch");
|
||||
});
|
||||
|
||||
it("SSE endpoint returns existing progress for failed job", async () => {
|
||||
const jobId = randomUUID();
|
||||
|
||||
updateJobProgress({
|
||||
jobId,
|
||||
status: "failed",
|
||||
totalFiles: 2,
|
||||
completedFiles: 1,
|
||||
failedFiles: 1,
|
||||
errors: [{ filename: "bad.png", error: "Invalid image" }],
|
||||
});
|
||||
// Wait for Redis pub/sub + setex round trip
|
||||
await new Promise((r) => setTimeout(r, 500));
|
||||
|
||||
const res = await app.inject({
|
||||
method: "GET",
|
||||
url: `/api/v1/jobs/${jobId}/progress`,
|
||||
headers: {
|
||||
authorization: `Bearer ${adminToken}`,
|
||||
},
|
||||
});
|
||||
|
||||
expect(res.statusCode).toBe(200);
|
||||
const body = res.body;
|
||||
const dataMatch = body.match(/data: (.+)/);
|
||||
expect(dataMatch).not.toBeNull();
|
||||
const event = JSON.parse(dataMatch?.[1]);
|
||||
expect(event.status).toBe("failed");
|
||||
expect(event.failedFiles).toBe(1);
|
||||
expect(event.errors).toHaveLength(1);
|
||||
});
|
||||
});
|
||||
|
||||
// ── updateJobProgress direct tests ─────────────────────────────
|
||||
describe("updateJobProgress direct calls", () => {
|
||||
it("persists job progress to the database for a new job", async () => {
|
||||
const jobId = randomUUID();
|
||||
|
||||
updateJobProgress({
|
||||
jobId,
|
||||
status: "processing",
|
||||
totalFiles: 5,
|
||||
completedFiles: 2,
|
||||
failedFiles: 0,
|
||||
errors: [],
|
||||
});
|
||||
await flushPersist(jobId, ["processing"]);
|
||||
|
||||
const [job] = await db.select().from(schema.jobs).where(eq(schema.jobs.id, jobId));
|
||||
expect(job).toBeDefined();
|
||||
expect(job?.status).toBe("processing");
|
||||
expect((job?.progress as { percent: number })?.percent).toBe(40); // 2/5
|
||||
expect(job?.type).toBe("batch");
|
||||
});
|
||||
|
||||
it("updates existing job progress in the database", async () => {
|
||||
const jobId = randomUUID();
|
||||
|
||||
// Create initial progress
|
||||
updateJobProgress({
|
||||
jobId,
|
||||
status: "processing",
|
||||
totalFiles: 3,
|
||||
completedFiles: 1,
|
||||
failedFiles: 0,
|
||||
errors: [],
|
||||
});
|
||||
await flushPersist(jobId, ["processing"]);
|
||||
|
||||
// Update progress
|
||||
updateJobProgress({
|
||||
jobId,
|
||||
status: "completed",
|
||||
totalFiles: 3,
|
||||
completedFiles: 3,
|
||||
failedFiles: 0,
|
||||
errors: [],
|
||||
});
|
||||
await flushPersist(jobId);
|
||||
|
||||
const [job] = await db.select().from(schema.jobs).where(eq(schema.jobs.id, jobId));
|
||||
expect(job).toBeDefined();
|
||||
expect(job?.status).toBe("completed");
|
||||
expect((job?.progress as { percent: number })?.percent).toBe(100);
|
||||
expect(job?.completedAt).not.toBeNull();
|
||||
});
|
||||
|
||||
it("persists errors to the database", async () => {
|
||||
const jobId = randomUUID();
|
||||
|
||||
updateJobProgress({
|
||||
jobId,
|
||||
status: "failed",
|
||||
totalFiles: 2,
|
||||
completedFiles: 0,
|
||||
failedFiles: 2,
|
||||
errors: [
|
||||
{ filename: "a.png", error: "Invalid format" },
|
||||
{ filename: "b.png", error: "Corrupt file" },
|
||||
],
|
||||
});
|
||||
await flushPersist(jobId);
|
||||
|
||||
const [job] = await db.select().from(schema.jobs).where(eq(schema.jobs.id, jobId));
|
||||
expect(job).toBeDefined();
|
||||
expect(job?.status).toBe("failed");
|
||||
expect(job?.error).not.toBeNull();
|
||||
const errorObj = job?.error as { message: string };
|
||||
const errors = errorObj.details;
|
||||
expect(errors).toHaveLength(2);
|
||||
});
|
||||
|
||||
it("handles zero totalFiles without division by zero", async () => {
|
||||
const jobId = randomUUID();
|
||||
|
||||
updateJobProgress({
|
||||
jobId,
|
||||
status: "completed",
|
||||
totalFiles: 0,
|
||||
completedFiles: 0,
|
||||
failedFiles: 0,
|
||||
errors: [],
|
||||
});
|
||||
await flushPersist(jobId);
|
||||
|
||||
const [job] = await db.select().from(schema.jobs).where(eq(schema.jobs.id, jobId));
|
||||
expect(job).toBeDefined();
|
||||
expect((job?.progress as { percent: number })?.percent).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
// ── updateSingleFileProgress direct tests ──────────────────────
|
||||
describe("updateSingleFileProgress direct calls", () => {
|
||||
it("persists single-file progress for new job", async () => {
|
||||
const jobId = randomUUID();
|
||||
|
||||
updateSingleFileProgress({
|
||||
jobId,
|
||||
phase: "processing",
|
||||
percent: 50,
|
||||
stage: "encoding",
|
||||
});
|
||||
await flushPersist(jobId, ["processing"]);
|
||||
|
||||
const [job] = await db.select().from(schema.jobs).where(eq(schema.jobs.id, jobId));
|
||||
expect(job).toBeDefined();
|
||||
expect(job?.status).toBe("processing");
|
||||
const p = job?.progress as { percent: number; stage?: string };
|
||||
expect(p?.percent).toBe(50);
|
||||
expect(p?.stage).toBe("encoding");
|
||||
expect(job?.type).toBe("single");
|
||||
});
|
||||
|
||||
it("persists complete phase", async () => {
|
||||
const jobId = randomUUID();
|
||||
|
||||
updateSingleFileProgress({
|
||||
jobId,
|
||||
phase: "complete",
|
||||
percent: 100,
|
||||
});
|
||||
await flushPersist(jobId);
|
||||
|
||||
const [job] = await db.select().from(schema.jobs).where(eq(schema.jobs.id, jobId));
|
||||
expect(job).toBeDefined();
|
||||
expect(job?.status).toBe("completed");
|
||||
expect((job?.progress as { percent: number })?.percent).toBe(100);
|
||||
// completedAt is only set on UPDATE path (not INSERT for new jobs)
|
||||
expect(job?.type).toBe("single");
|
||||
});
|
||||
|
||||
it("persists failed phase with error", async () => {
|
||||
const jobId = randomUUID();
|
||||
|
||||
updateSingleFileProgress({
|
||||
jobId,
|
||||
phase: "failed",
|
||||
percent: 30,
|
||||
error: "Processing timeout",
|
||||
});
|
||||
await flushPersist(jobId);
|
||||
|
||||
const [job] = await db.select().from(schema.jobs).where(eq(schema.jobs.id, jobId));
|
||||
expect(job).toBeDefined();
|
||||
expect(job?.status).toBe("failed");
|
||||
expect((job?.error as { message: string })?.message).toBe("Processing timeout");
|
||||
expect(job?.type).toBe("single");
|
||||
});
|
||||
|
||||
it("sets completedAt when updating existing job to complete", async () => {
|
||||
const jobId = randomUUID();
|
||||
|
||||
// Create initial job
|
||||
updateSingleFileProgress({
|
||||
jobId,
|
||||
phase: "processing",
|
||||
percent: 50,
|
||||
});
|
||||
await flushPersist(jobId, ["processing"]);
|
||||
|
||||
// Update to complete
|
||||
updateSingleFileProgress({
|
||||
jobId,
|
||||
phase: "complete",
|
||||
percent: 100,
|
||||
});
|
||||
await flushPersist(jobId);
|
||||
|
||||
const [job] = await db.select().from(schema.jobs).where(eq(schema.jobs.id, jobId));
|
||||
expect(job).toBeDefined();
|
||||
expect(job?.status).toBe("completed");
|
||||
expect(job?.completedAt).not.toBeNull();
|
||||
});
|
||||
|
||||
it("sets completedAt when updating existing job to failed", async () => {
|
||||
const jobId = randomUUID();
|
||||
|
||||
// Create initial job
|
||||
updateSingleFileProgress({
|
||||
jobId,
|
||||
phase: "processing",
|
||||
percent: 25,
|
||||
});
|
||||
await flushPersist(jobId, ["processing"]);
|
||||
|
||||
// Update to failed
|
||||
updateSingleFileProgress({
|
||||
jobId,
|
||||
phase: "failed",
|
||||
percent: 25,
|
||||
error: "Timeout error",
|
||||
});
|
||||
await flushPersist(jobId);
|
||||
|
||||
const [job] = await db.select().from(schema.jobs).where(eq(schema.jobs.id, jobId));
|
||||
expect(job).toBeDefined();
|
||||
expect(job?.status).toBe("failed");
|
||||
expect(job?.completedAt).not.toBeNull();
|
||||
expect((job?.error as { message: string })?.message).toBe("Timeout error");
|
||||
});
|
||||
|
||||
it("updates existing single-file job progress", async () => {
|
||||
const jobId = randomUUID();
|
||||
|
||||
// Create
|
||||
updateSingleFileProgress({
|
||||
jobId,
|
||||
phase: "processing",
|
||||
percent: 25,
|
||||
stage: "analyzing",
|
||||
});
|
||||
await flushPersist(jobId, ["processing"]);
|
||||
|
||||
// Update
|
||||
updateSingleFileProgress({
|
||||
jobId,
|
||||
phase: "processing",
|
||||
percent: 75,
|
||||
stage: "encoding",
|
||||
});
|
||||
|
||||
// Poll for the expected percent value (both updates produce "processing"
|
||||
// status, so status-based polling is insufficient)
|
||||
const start = Date.now();
|
||||
let finalPercent = 0;
|
||||
while (Date.now() - start < 2000) {
|
||||
const [row] = await db.select().from(schema.jobs).where(eq(schema.jobs.id, jobId));
|
||||
if (row) {
|
||||
finalPercent = (row.progress as { percent: number })?.percent ?? 0;
|
||||
if (finalPercent === 75) break;
|
||||
}
|
||||
await new Promise((r) => setTimeout(r, 50));
|
||||
}
|
||||
|
||||
expect(finalPercent).toBe(75);
|
||||
});
|
||||
});
|
||||
|
||||
// recoverStaleJobs was removed in the Redis transport migration.
|
||||
// Stale-job recovery is now handled by BullMQ's built-in stalled-job
|
||||
// mechanism. The four tests that exercised recoverStaleJobs were:
|
||||
// - "marks processing jobs as failed on recovery"
|
||||
// - "marks queued jobs as failed on recovery"
|
||||
// - "does not modify completed jobs"
|
||||
// - "does not modify already-failed jobs"
|
||||
// All four tested a deleted internal; equivalent coverage is provided
|
||||
// by BullMQ's stalled-job handler (Task 6 worker runtime).
|
||||
Reference in New Issue
Block a user