feat: publish av-tools 0.1.0
This commit is contained in:
269
tests/unit/jobs.test.ts
Normal file
269
tests/unit/jobs.test.ts
Normal file
@@ -0,0 +1,269 @@
|
||||
import { describe, expect, it, vi } from 'vitest';
|
||||
import {
|
||||
MediaJobQueue,
|
||||
MemoryJobMetadataStore,
|
||||
ObjectUrlRegistry,
|
||||
parseFFmpegProgress,
|
||||
progressUpdateFromFFmpeg,
|
||||
weightedJobProgress,
|
||||
type JobExecutionContext,
|
||||
type JobExecutionResult,
|
||||
type MediaJob,
|
||||
type MediaJobDefinition,
|
||||
type MediaJobExecutor,
|
||||
} from '../../src/jobs';
|
||||
import { freezeCommandPlan } from '../../src/commands';
|
||||
|
||||
function definition(id: string, stepIds = ['encode']): MediaJobDefinition {
|
||||
return {
|
||||
id,
|
||||
operation: 'test',
|
||||
steps: stepIds.map((stepId) => ({
|
||||
id: stepId,
|
||||
name: stepId,
|
||||
weight: stepId === 'read' ? 1 : 9,
|
||||
plan: freezeCommandPlan({
|
||||
id: `${id}:${stepId}`,
|
||||
operation: stepId,
|
||||
inputs: [],
|
||||
temporaryFiles: [],
|
||||
args: ['-version'],
|
||||
outputs: [],
|
||||
requiredCapabilities: {},
|
||||
diagnostics: [],
|
||||
}),
|
||||
})),
|
||||
};
|
||||
}
|
||||
|
||||
class ControlledExecutor implements MediaJobExecutor {
|
||||
readonly calls: string[] = [];
|
||||
readonly cancelCurrentJob = vi.fn(async () => undefined);
|
||||
readonly recoverAfterCancellation = vi.fn(async () => undefined);
|
||||
readonly cleanup = vi.fn(async () => undefined);
|
||||
#gates: (() => void)[] = [];
|
||||
failNext?: Error;
|
||||
|
||||
execute(
|
||||
plan: { readonly id: string },
|
||||
context: JobExecutionContext
|
||||
): Promise<JobExecutionResult> {
|
||||
this.calls.push(plan.id);
|
||||
context.onProgress({ progress: 0.5 });
|
||||
if (this.failNext) {
|
||||
const error = this.failNext;
|
||||
this.failNext = undefined;
|
||||
return Promise.reject(error);
|
||||
}
|
||||
return new Promise((resolve, reject) => {
|
||||
const release = () => {
|
||||
context.signal.removeEventListener('abort', abort);
|
||||
resolve({ outputs: new Map() });
|
||||
};
|
||||
const abort = () => {
|
||||
const index = this.#gates.indexOf(release);
|
||||
if (index >= 0) {
|
||||
this.#gates.splice(index, 1);
|
||||
}
|
||||
reject(new DOMException('cancelled', 'AbortError'));
|
||||
};
|
||||
context.signal.addEventListener('abort', abort, { once: true });
|
||||
this.#gates.push(release);
|
||||
});
|
||||
}
|
||||
|
||||
release(): void {
|
||||
this.#gates.shift()?.();
|
||||
}
|
||||
}
|
||||
|
||||
describe('progress parsing', () => {
|
||||
it('parses FFmpeg machine progress and caps completion until output read', () => {
|
||||
const parsed = parseFFmpegProgress(
|
||||
'frame=25\nfps=12.5\nout_time_us=5000000\nspeed=1.5x\nprogress=continue\n'
|
||||
);
|
||||
expect(parsed).toEqual({
|
||||
frame: 25,
|
||||
fps: 12.5,
|
||||
outTimeSeconds: 5,
|
||||
speed: 1.5,
|
||||
terminal: false,
|
||||
});
|
||||
expect(progressUpdateFromFFmpeg(parsed, 10)).toMatchObject({
|
||||
progress: 0.5,
|
||||
});
|
||||
expect(
|
||||
progressUpdateFromFFmpeg(
|
||||
parseFFmpegProgress('progress=end\nout_time_us=10000000'),
|
||||
10
|
||||
).progress
|
||||
).toBe(0.99);
|
||||
});
|
||||
|
||||
it('combines weighted multi-step progress below 100 percent', () => {
|
||||
expect(
|
||||
weightedJobProgress([
|
||||
{
|
||||
id: 'analysis',
|
||||
name: 'Analysis',
|
||||
weight: 1,
|
||||
status: 'completed',
|
||||
progress: 1,
|
||||
planId: 'one',
|
||||
},
|
||||
{
|
||||
id: 'encode',
|
||||
name: 'Encoding',
|
||||
weight: 9,
|
||||
status: 'running',
|
||||
progress: 0.5,
|
||||
planId: 'two',
|
||||
},
|
||||
])
|
||||
).toBe(0.55);
|
||||
});
|
||||
});
|
||||
|
||||
describe('sequential durable job queue', () => {
|
||||
it('runs only one command at a time and reaches 100% after cleanup', async () => {
|
||||
const executor = new ControlledExecutor();
|
||||
const store = new MemoryJobMetadataStore();
|
||||
const queue = new MediaJobQueue({ executor, store });
|
||||
await queue.initialize();
|
||||
await queue.enqueue(definition('one'));
|
||||
await queue.enqueue(definition('two'));
|
||||
await waitUntil(() => executor.calls.length === 1);
|
||||
expect(executor.calls).toEqual(['one:encode']);
|
||||
expect(queue.get('one')?.progress).toBe(0.5);
|
||||
executor.release();
|
||||
await waitUntil(() => executor.calls.length === 2);
|
||||
expect(queue.get('one')).toMatchObject({
|
||||
status: 'completed',
|
||||
progress: 1,
|
||||
});
|
||||
executor.release();
|
||||
await queue.waitForIdle();
|
||||
expect(queue.get('two')).toMatchObject({
|
||||
status: 'completed',
|
||||
progress: 1,
|
||||
});
|
||||
expect(queue.getResults('two')).toHaveLength(1);
|
||||
expect(queue.releaseResults('two')).toBe(true);
|
||||
expect(queue.getResults('two')).toBeUndefined();
|
||||
expect((await store.list()).map((job) => job.status)).toEqual([
|
||||
'completed',
|
||||
'completed',
|
||||
]);
|
||||
});
|
||||
|
||||
it('terminates processing, recovers the engine, and runs the next job', async () => {
|
||||
const executor = new ControlledExecutor();
|
||||
const queue = new MediaJobQueue({
|
||||
executor,
|
||||
store: new MemoryJobMetadataStore(),
|
||||
});
|
||||
await queue.initialize();
|
||||
await queue.enqueue(definition('cancel-me'));
|
||||
await queue.enqueue(definition('after'));
|
||||
await waitUntil(() => queue.get('cancel-me')?.status === 'running');
|
||||
await queue.cancel('cancel-me');
|
||||
expect(executor.cancelCurrentJob).toHaveBeenCalledOnce();
|
||||
expect(executor.recoverAfterCancellation).toHaveBeenCalledOnce();
|
||||
expect(queue.get('cancel-me')?.status).toBe('cancelled');
|
||||
await waitUntil(() => executor.calls.includes('after:encode'));
|
||||
executor.release();
|
||||
await queue.waitForIdle();
|
||||
expect(queue.get('after')?.status).toBe('completed');
|
||||
});
|
||||
|
||||
it('marks failures, cleans up, and retries with the retained definition', async () => {
|
||||
const executor = new ControlledExecutor();
|
||||
executor.failNext = new Error('encoder failed');
|
||||
const queue = new MediaJobQueue({
|
||||
executor,
|
||||
store: new MemoryJobMetadataStore(),
|
||||
});
|
||||
await queue.initialize();
|
||||
await queue.enqueue(definition('retry'));
|
||||
await queue.waitForIdle();
|
||||
expect(queue.get('retry')).toMatchObject({
|
||||
status: 'failed',
|
||||
error: { code: 'execution-failed', recoverable: true },
|
||||
});
|
||||
await queue.retry('retry');
|
||||
await waitUntil(() => executor.calls.length === 2);
|
||||
executor.release();
|
||||
await queue.waitForIdle();
|
||||
expect(queue.get('retry')).toMatchObject({
|
||||
status: 'completed',
|
||||
retryCount: 1,
|
||||
});
|
||||
});
|
||||
|
||||
it('recovers interrupted durable metadata as an explicit failed job', async () => {
|
||||
const store = new MemoryJobMetadataStore();
|
||||
const interrupted: MediaJob = {
|
||||
id: 'old',
|
||||
operation: 'convert',
|
||||
status: 'running',
|
||||
steps: [
|
||||
{
|
||||
id: 'encode',
|
||||
name: 'Encode',
|
||||
weight: 1,
|
||||
status: 'running',
|
||||
progress: 0.4,
|
||||
planId: 'old:encode',
|
||||
},
|
||||
],
|
||||
progress: 0.4,
|
||||
createdAt: '2026-01-01T00:00:00.000Z',
|
||||
updatedAt: '2026-01-01T00:01:00.000Z',
|
||||
startedAt: '2026-01-01T00:00:01.000Z',
|
||||
retryCount: 0,
|
||||
};
|
||||
await store.put(interrupted);
|
||||
const queue = new MediaJobQueue({
|
||||
executor: new ControlledExecutor(),
|
||||
store,
|
||||
});
|
||||
await queue.initialize();
|
||||
expect(queue.get('old')).toMatchObject({
|
||||
status: 'failed',
|
||||
recoveryNote: expect.stringContaining('previous browser session'),
|
||||
error: { code: 'session-interrupted', recoverable: true },
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe('object URL cleanup', () => {
|
||||
it('revokes replaced and cleared result URLs', () => {
|
||||
let index = 0;
|
||||
const revoke = vi.fn();
|
||||
const registry = new ObjectUrlRegistry({
|
||||
createObjectURL: () => `blob:${index++}`,
|
||||
revokeObjectURL: revoke,
|
||||
});
|
||||
expect(registry.replace('output', new Blob(['one']))).toBe('blob:0');
|
||||
expect(registry.replace('output', new Blob(['two']))).toBe('blob:1');
|
||||
registry.replace('preview', new Blob(['three']));
|
||||
expect(revoke).toHaveBeenCalledWith('blob:0');
|
||||
registry.clear();
|
||||
expect(revoke).toHaveBeenCalledWith('blob:1');
|
||||
expect(revoke).toHaveBeenCalledWith('blob:2');
|
||||
expect(registry.size).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
async function waitUntil(
|
||||
predicate: () => boolean,
|
||||
timeoutMs = 2_000
|
||||
): Promise<void> {
|
||||
const started = Date.now();
|
||||
while (!predicate()) {
|
||||
if (Date.now() - started > timeoutMs) {
|
||||
throw new Error('Timed out waiting for queue state');
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user