Cobre rotas HTTP, proxy, worker assíncrono e barreiras do dev auth
Os 34 testes existentes batiam só na camada de serviço: as rotas, o proxy e o loop do worker não tinham cobertura alguma. A suíte vai de 34 para 102. - `tests/http/`: rotas reais com o caminho real de autenticação, sem subir servidor. Cobre 401 sem sessão, módulo desligado, papel sem permissão, 403 ao tentar gravar workspace de terceiro, e 429 com Retry-After. - `tests/http/proxy.test.ts`: o proxy não propaga query da requisição para o login, o que impediria um returnTo forjado. - `tests/integration/worker.test.ts`: jornada assíncrona pelo loop real. A web enfileira e devolve `queued`, o download falha antes do tick, o tick conclui e os três formatos saem. Também: job sobrevive à perda da fila do Redis, worker relê o contexto e não processa com a flag desligada, TTL expira temporários sem tocar o modelo, e desligamento gracioso. - `tests/unit/dev-auth-guard.test.ts`: matriz de ambientes das três barreiras do login local. Um token de dev válido é recusado quando o ambiente é staging. Espaço em volta de um valor de ADAPTER_ENV é tolerado de propósito, e isso está declarado no teste; valor desconhecido nunca vira development. - `tests/services/`: Postgres e Redis reais. Confere `app_user` sem BYPASSRLS, FORCE RLS em todas as tabelas, isolamento entre workspaces, auditoria não alterável pelo papel da aplicação, e lock entre conexões diferentes. Sem as variáveis de serviço os casos são IGNORADOS, nunca aprovados, e ficam fora de `npm test` para a suíte padrão passar em qualquer máquina. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
1 parent
dea510c3b4
commit
ce11a50091
10 files changed
+1182
No files matched your search
@@ -0,0 +1,220 @@
|
||||
import { afterEach, beforeEach, describe, expect, it } from "vitest";
|
||||
import { addSource, createJob, createModel, saveMappings, startProcess } from "@/server/spreadsheet-adapter/service";
|
||||
import { downloadOutput } from "@/server/spreadsheet-adapter/process-job";
|
||||
import { createWorker } from "@/server/spreadsheet-adapter/worker-loop";
|
||||
import { lockKey } from "@/infra/redis/redis";
|
||||
import { ctxOf, createReadyDb, depsOf, IDS } from "../helpers/world";
|
||||
|
||||
/**
|
||||
* Jornada assíncrona pelo loop real do worker: a web enfileira e devolve, e o
|
||||
* processamento acontece depois, no tick. É o mesmo código que o processo do
|
||||
* contêiner executa. O que estes casos não cobrem é a fronteira de processo e o
|
||||
* Redis real — isso está em `tests/services/redis.test.ts` e no CI.
|
||||
*/
|
||||
|
||||
const previous = { mode: process.env.ADAPTER_JOB_MODE };
|
||||
|
||||
beforeEach(() => {
|
||||
// Modo worker: `startProcess` só enfileira e sai.
|
||||
process.env.ADAPTER_JOB_MODE = "worker";
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
if (previous.mode === undefined) delete process.env.ADAPTER_JOB_MODE;
|
||||
else process.env.ADAPTER_JOB_MODE = previous.mode;
|
||||
});
|
||||
|
||||
type World = Awaited<ReturnType<typeof prepare>>;
|
||||
|
||||
async function prepare() {
|
||||
const db = await createReadyDb();
|
||||
const deps = depsOf(db);
|
||||
const ana = ctxOf("ana", "alpha", "owner", true);
|
||||
const model = await createModel(deps, ana, {
|
||||
name: "Clientes",
|
||||
filename: "modelo.csv",
|
||||
bytes: Buffer.from("Nome,Valor\n", "utf8"),
|
||||
requiredColumns: ["Nome", "Valor"],
|
||||
});
|
||||
const modelBytes = await deps.storage.get(`workspaces/${IDS.alpha}/models/${model.id}/original`);
|
||||
const job = await createJob(deps, ana, model.id, "job-async");
|
||||
const added = await addSource(deps, ana, job.id, {
|
||||
filename: "fonte.csv",
|
||||
bytes: Buffer.from("Nome,Valor\nAna,10\nBruno,20\n", "utf8"),
|
||||
});
|
||||
const sourceId = added.job.sources[0].id;
|
||||
await saveMappings(deps, ana, job.id, sourceId, [
|
||||
{ sourceColumn: "Nome", modelColumn: "Nome" },
|
||||
{ sourceColumn: "Valor", modelColumn: "Valor" },
|
||||
]);
|
||||
return { db, deps, ana, model, modelBytes, job, sourceId };
|
||||
}
|
||||
|
||||
async function statusOf(world: World, jobId: string): Promise<string> {
|
||||
const rows = await world.db.withService((query) =>
|
||||
query.query<{ status: string }>(`select status from public.spreadsheet_jobs where id = $1`, [jobId]),
|
||||
);
|
||||
return rows[0]?.status ?? "ausente";
|
||||
}
|
||||
|
||||
describe("jornada assíncrona pelo worker", () => {
|
||||
it("enfileira na web, processa no tick e entrega os três formatos", async () => {
|
||||
const world = await prepare();
|
||||
const { deps, ana, job, sourceId } = world;
|
||||
|
||||
const queued = await startProcess(deps, ana, job.id, sourceId, "proc-async");
|
||||
expect(queued.status).toBe("queued");
|
||||
// Nada foi processado ainda: a web só enfileirou.
|
||||
await expect(downloadOutput(deps, ana, job.id, sourceId, "csv")).rejects.toMatchObject({ code: "NOT_FOUND" });
|
||||
|
||||
const worker = createWorker(deps, { heartbeatPath: null });
|
||||
const result = await worker.tick();
|
||||
|
||||
expect(result.processed).toEqual([job.id]);
|
||||
expect(await statusOf(world, job.id)).toBe("completed");
|
||||
|
||||
const csv = await downloadOutput(deps, ana, job.id, sourceId, "csv");
|
||||
expect(csv.body.toString("utf8")).toContain("Ana");
|
||||
const xlsx = await downloadOutput(deps, ana, job.id, sourceId, "xlsx");
|
||||
expect(xlsx.body.subarray(0, 2).toString()).toBe("PK");
|
||||
const pdf = await downloadOutput(deps, ana, job.id, sourceId, "pdf");
|
||||
expect(pdf.body.subarray(0, 4).toString()).toBe("%PDF");
|
||||
|
||||
const after = await deps.storage.get(`workspaces/${IDS.alpha}/models/${world.model.id}/original`);
|
||||
expect(after.equals(world.modelBytes)).toBe(true);
|
||||
});
|
||||
|
||||
it("libera o lock do usuário ao concluir, permitindo o próximo job", async () => {
|
||||
const world = await prepare();
|
||||
const { deps, ana, job, sourceId } = world;
|
||||
await startProcess(deps, ana, job.id, sourceId, "proc-lock");
|
||||
await createWorker(deps, { heartbeatPath: null }).tick();
|
||||
|
||||
// Lock livre: outro processamento consegue adquirir.
|
||||
expect(await deps.coordinator.acquireLock(lockKey(ana.userId), "outro-job", 30)).toBe(true);
|
||||
});
|
||||
|
||||
it("encontra o job pela varredura do banco quando a fila do Redis se perdeu", async () => {
|
||||
const world = await prepare();
|
||||
const { deps, ana, job, sourceId } = world;
|
||||
await startProcess(deps, ana, job.id, sourceId, "proc-sem-fila");
|
||||
|
||||
// Simula Redis reiniciado: a fila some, o estado durável no Postgres fica.
|
||||
while (await deps.coordinator.dequeue()) {
|
||||
/* drena */
|
||||
}
|
||||
expect(await deps.coordinator.dequeue()).toBeNull();
|
||||
|
||||
const result = await createWorker(deps, { heartbeatPath: null }).tick();
|
||||
expect(result.processed).toEqual([job.id]);
|
||||
expect(await statusOf(world, job.id)).toBe("completed");
|
||||
});
|
||||
|
||||
it("relê o contexto do dono e não processa com o módulo desligado", async () => {
|
||||
const world = await prepare();
|
||||
const { db, deps, ana, job, sourceId } = world;
|
||||
await startProcess(deps, ana, job.id, sourceId, "proc-flag");
|
||||
|
||||
// A flag cai entre o enfileiramento e o tick.
|
||||
await db.withService((query) =>
|
||||
query.query(
|
||||
`update public.workspace_module_flags set enabled = false where workspace_id = $1 and module_key = 'spreadsheet_adapter'`,
|
||||
[IDS.alpha],
|
||||
),
|
||||
);
|
||||
|
||||
const result = await createWorker(deps, { heartbeatPath: null }).tick();
|
||||
expect(result.processed).toEqual([]);
|
||||
expect(await statusOf(world, job.id)).not.toBe("completed");
|
||||
await expect(downloadOutput(deps, ana, job.id, sourceId, "csv")).rejects.toBeTruthy();
|
||||
});
|
||||
|
||||
it("expira temporários no tick e preserva o modelo", async () => {
|
||||
const world = await prepare();
|
||||
const { deps, ana, job, sourceId } = world;
|
||||
await startProcess(deps, ana, job.id, sourceId, "proc-ttl");
|
||||
await createWorker(deps, { heartbeatPath: null }).tick();
|
||||
await downloadOutput(deps, ana, job.id, sourceId, "csv");
|
||||
|
||||
// Avança além do TTL e roda o tick de novo.
|
||||
const expirado = { ...deps, now: () => new Date("2027-01-01T00:00:00.000Z") };
|
||||
const result = await createWorker(expirado, { heartbeatPath: null }).tick();
|
||||
expect(result.cleaned).toBeGreaterThan(0);
|
||||
await expect(downloadOutput(expirado, ana, job.id, sourceId, "csv")).rejects.toBeTruthy();
|
||||
|
||||
const model = await deps.storage.get(`workspaces/${IDS.alpha}/models/${world.model.id}/original`);
|
||||
expect(model.equals(world.modelBytes)).toBe(true);
|
||||
});
|
||||
|
||||
it("grava heartbeat ao fim do tick", async () => {
|
||||
const world = await prepare();
|
||||
const escritas: string[] = [];
|
||||
const worker = createWorker(world.deps, {
|
||||
heartbeatPath: "/tmp/teste-heartbeat",
|
||||
writeHeartbeat: async (target) => {
|
||||
escritas.push(target);
|
||||
},
|
||||
});
|
||||
await worker.tick();
|
||||
expect(escritas).toEqual(["/tmp/teste-heartbeat"]);
|
||||
});
|
||||
});
|
||||
|
||||
describe("desligamento gracioso do worker", () => {
|
||||
it("para de aceitar trabalho novo depois do shutdown", async () => {
|
||||
const world = await prepare();
|
||||
const { deps, ana, job, sourceId } = world;
|
||||
const worker = createWorker(deps, { heartbeatPath: null });
|
||||
|
||||
expect(worker.isRunning()).toBe(true);
|
||||
expect(await worker.shutdown(1000)).toBe(true);
|
||||
expect(worker.isRunning()).toBe(false);
|
||||
|
||||
await startProcess(deps, ana, job.id, sourceId, "proc-pos-shutdown");
|
||||
const result = await worker.tick();
|
||||
expect(result.processed).toEqual([]);
|
||||
expect(await statusOf(world, job.id)).toBe("queued");
|
||||
});
|
||||
|
||||
it("espera o tick em andamento e não perde o job que não começou", async () => {
|
||||
const world = await prepare();
|
||||
const { deps, ana, job, sourceId } = world;
|
||||
await startProcess(deps, ana, job.id, sourceId, "proc-espera");
|
||||
|
||||
const worker = createWorker(deps, { heartbeatPath: null });
|
||||
const inFlight = worker.tick();
|
||||
expect(worker.isBusy()).toBe(true);
|
||||
|
||||
// O encerramento espera o tick corrente sair: é essa espera que dá tempo ao
|
||||
// `finally` do processamento liberar o lock de um job já iniciado.
|
||||
const clean = await worker.shutdown(10_000);
|
||||
expect(clean).toBe(true);
|
||||
expect(worker.isBusy()).toBe(false);
|
||||
await inFlight;
|
||||
|
||||
// O job que ainda não havia começado continua `queued`: nada pela metade.
|
||||
expect(await statusOf(world, job.id)).toBe("queued");
|
||||
// O lock continua reservado para ESTE job, não vazou para outro dono. Em
|
||||
// modo worker ele é adquirido no enfileiramento, com TTL de 120s, e
|
||||
// `runProcess` o reaproveita no tick seguinte.
|
||||
expect(await deps.coordinator.ownsLock(lockKey(ana.userId), job.id)).toBe(true);
|
||||
expect(await deps.coordinator.acquireLock(lockKey(ana.userId), "outro-job", 5)).toBe(false);
|
||||
|
||||
// O worker seguinte conclui o mesmo job, como aconteceria após um restart.
|
||||
const result = await createWorker(deps, { heartbeatPath: null }).tick();
|
||||
expect(result.processed).toEqual([job.id]);
|
||||
expect(await statusOf(world, job.id)).toBe("completed");
|
||||
// Concluído, o lock é liberado pelo `finally` do processamento.
|
||||
expect(await deps.coordinator.acquireLock(lockKey(ana.userId), "sonda", 5)).toBe(true);
|
||||
});
|
||||
|
||||
it("reporta encerramento sujo quando a carência estoura", async () => {
|
||||
const world = await prepare();
|
||||
const worker = createWorker(world.deps, { heartbeatPath: null });
|
||||
const inFlight = worker.tick();
|
||||
expect(worker.isBusy()).toBe(true);
|
||||
// Carência zero: não espera nada e reporta que o tick não terminou.
|
||||
expect(await worker.shutdown(0)).toBe(false);
|
||||
await inFlight;
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user