From 8fb790d0a25fce3b8f0091d3d318451c4eab2c69 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fran=C3=A7ois=20Chalifour?= Date: Tue, 30 Jun 2026 15:34:34 +0200 Subject: [PATCH 1/3] fix(lakehouse): validate upload files before streaming --- cli/src/commands/lakehouse.ts | 25 +++++-- cli/src/lib/lakehouse-transport.ts | 16 +++-- cli/tests/lakehouse.test.ts | 105 ++++++++++++++++++++++++++++- 3 files changed, 133 insertions(+), 13 deletions(-) diff --git a/cli/src/commands/lakehouse.ts b/cli/src/commands/lakehouse.ts index de600ec..282ab50 100644 --- a/cli/src/commands/lakehouse.ts +++ b/cli/src/commands/lakehouse.ts @@ -1,4 +1,5 @@ import type { ArgsDef } from "citty"; +import { statSync } from "node:fs"; import { CliError } from "@/lib/errors.ts"; import { booleanArg, enumArg, optionalStringArg, stringArg } from "@/lib/operation-codec.ts"; import { httpEffect, progressPlan, scopedPlan } from "@/lib/operation-effect.ts"; @@ -28,6 +29,21 @@ import { const UPLOAD_MODE_OPTIONS = ["overwrite", "upsert"] as const; +function getUploadFileSizeBytes(filePath: string): number { + try { + const fileStat = statSync(filePath); + if (!fileStat.isFile()) { + throw new CliError(`File not found: ${filePath}`); + } + return fileStat.size; + } catch (error) { + if (error instanceof CliError) { + throw error; + } + throw new CliError(`File not found: ${filePath}`); + } +} + const queryRunArgs = { statement: { type: "string", description: "SQL statement to run", required: true }, format: { @@ -291,12 +307,7 @@ export const uploadCommand = defineOperationCommand({ const mode = enumArg(args, "mode", UPLOAD_MODE_OPTIONS); const filePath = stringArg(args, "file"); validateUploadPrimaryKey(mode, args["primary-key"]); - - try { - await Bun.file(filePath).arrayBuffer(); - } catch { - throw new CliError(`File not found: ${filePath}`); - } + const fileSizeBytes = getUploadFileSizeBytes(filePath); const readTimeoutMs = parseRequestReadTimeoutMs(args); return { @@ -306,6 +317,7 @@ export const uploadCommand = defineOperationCommand({ format: stringArg(args, "format"), mode, filePath, + fileSizeBytes, primaryKey: optionalStringArg(args, "primary-key"), httpOptions: readTimeoutMs !== undefined ? { readTimeoutMs } : undefined, }; @@ -319,6 +331,7 @@ export const uploadCommand = defineOperationCommand({ format: input.format, mode: input.mode, filePath: input.filePath, + fileSizeBytes: input.fileSizeBytes, primaryKey: input.primaryKey, httpOptions: input.httpOptions, }); diff --git a/cli/src/lib/lakehouse-transport.ts b/cli/src/lib/lakehouse-transport.ts index b5c1a52..ee9d7bb 100644 --- a/cli/src/lib/lakehouse-transport.ts +++ b/cli/src/lib/lakehouse-transport.ts @@ -21,6 +21,7 @@ export type LakehouseUploadRequestInput = { format: string; mode: string; filePath: string; + fileSizeBytes: number; primaryKey?: string; httpOptions?: Partial; }; @@ -79,14 +80,17 @@ export function createLakehouseUploadRequest( } const file = Bun.file(input.filePath); - const fileSizeBytes = file.size; let body: Blob | ReadableStream = file; - let uploadProgress = createUploadProgressReporter(fileSizeBytes); + let uploadProgress = createUploadProgressReporter(input.fileSizeBytes); - if (shouldShowProgress() && fileSizeBytes > 0) { - body = wrapStreamWithByteProgress(file.stream(), fileSizeBytes, (sentBytes, totalBytes) => { - uploadProgress.report(sentBytes, totalBytes); - }); + if (shouldShowProgress() && input.fileSizeBytes > 0) { + body = wrapStreamWithByteProgress( + file.stream(), + input.fileSizeBytes, + (sentBytes, totalBytes) => { + uploadProgress.report(sentBytes, totalBytes); + }, + ); } else { uploadProgress = createUploadProgressReporter(0); } diff --git a/cli/tests/lakehouse.test.ts b/cli/tests/lakehouse.test.ts index f8db37b..b816afc 100644 --- a/cli/tests/lakehouse.test.ts +++ b/cli/tests/lakehouse.test.ts @@ -1,6 +1,6 @@ import { describe, expect, test, beforeEach, afterEach } from "bun:test"; import { runCommand } from "citty"; -import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import { join } from "node:path"; import { tmpdir } from "node:os"; import { buildMainCommand } from "@/cli.ts"; @@ -430,6 +430,7 @@ describe("lakehouse request construction", () => { format: "csv", mode: "upsert", filePath: uploadFile, + fileSizeBytes: 15, primaryKey: "id", }); try { @@ -444,6 +445,108 @@ describe("lakehouse request construction", () => { expect(logContent).toMatch(/PAYLOAD=@(?:blob|stream)/); }); + test("upload command validates file metadata without buffering the payload", async () => { + const uploadFile = join(testHome, "command-data.csv"); + writeFileSync(uploadFile, "id,name\n1,Alice", "utf8"); + writeFileSync( + mockFile, + JSON.stringify([ + { + urlPattern: "/upload", + method: "POST", + body: '{"ok":true}', + }, + ]), + ); + + const previousRuntime = getCliRuntime(); + const runtime = createCliRuntime({ debug: false, json: true, agent: false }); + runtime.output.writeJson = () => {}; + runtime.output.writeRaw = () => {}; + runtime.output.writeHuman = () => {}; + setCliRuntime(runtime); + + try { + await runCommand(buildMainCommand(), { + rawArgs: [ + "upload", + "--catalog", + "memory", + "--schema", + "main", + "--table", + "users", + "--format", + "csv", + "--mode", + "overwrite", + "--file", + uploadFile, + ], + }); + } finally { + setCliRuntime(previousRuntime); + } + + const logContent = readFileSync(logFile, "utf8"); + expect(logContent).toContain("URL=https://example.com/upload?"); + expect(logContent).toContain("PAYLOAD=@blob"); + expect(logContent).not.toContain("id,name"); + }); + + test("upload command rejects missing files and directories", async () => { + const directoryPath = join(testHome, "upload-directory"); + mkdirSync(directoryPath); + + try { + await runCommand(buildMainCommand(), { + rawArgs: [ + "upload", + "--catalog", + "memory", + "--schema", + "main", + "--table", + "users", + "--format", + "csv", + "--mode", + "overwrite", + "--file", + join(testHome, "missing.csv"), + ], + }); + throw new Error("expected missing file failure"); + } catch (error) { + expect(error).toBeInstanceOf(Error); + expect((error as Error).message).toContain("File not found"); + } + + try { + await runCommand(buildMainCommand(), { + rawArgs: [ + "upload", + "--catalog", + "memory", + "--schema", + "main", + "--table", + "users", + "--format", + "csv", + "--mode", + "overwrite", + "--file", + directoryPath, + ], + }); + throw new Error("expected directory failure"); + } catch (error) { + expect(error).toBeInstanceOf(Error); + expect((error as Error).message).toContain("File not found"); + } + }); + test("cancel URL-encodes query id in path", async () => { const queryId = "query/id+special"; const encodedQueryId = encodeURIComponent(queryId); From a5bf8473667469c280dde89e7ab893edc3e34873 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fran=C3=A7ois=20Chalifour?= Date: Tue, 30 Jun 2026 16:29:32 +0200 Subject: [PATCH 2/3] Reuse CLI test runtime in lakehouse tests --- cli/tests/cli-test-runtime.ts | 14 ++++ cli/tests/lakehouse.test.ts | 138 ++++++++++++++-------------------- 2 files changed, 69 insertions(+), 83 deletions(-) create mode 100644 cli/tests/cli-test-runtime.ts diff --git a/cli/tests/cli-test-runtime.ts b/cli/tests/cli-test-runtime.ts new file mode 100644 index 0000000..9f5b499 --- /dev/null +++ b/cli/tests/cli-test-runtime.ts @@ -0,0 +1,14 @@ +import type { CliContext } from "@/context.ts"; +import { createCliRuntime, type CliRuntime } from "@/lib/runtime.ts"; + +export function createCliTestRuntime( + context: CliContext = { debug: false, json: true, agent: false }, +): CliRuntime { + const runtime = createCliRuntime(context); + runtime.output.writeStderr = () => {}; + runtime.output.writeJson = () => {}; + runtime.output.writeRaw = () => {}; + runtime.output.writeHuman = () => {}; + runtime.output.writeMetadata = () => {}; + return runtime; +} diff --git a/cli/tests/lakehouse.test.ts b/cli/tests/lakehouse.test.ts index b816afc..568cb9b 100644 --- a/cli/tests/lakehouse.test.ts +++ b/cli/tests/lakehouse.test.ts @@ -28,12 +28,8 @@ import { runOperationPlan, } from "@/lib/operation-effect.ts"; import type { OperationContext } from "@/lib/operation-command.ts"; -import { - createCliRuntime, - getCliRuntime, - refreshCliRuntimeContext, - setCliRuntime, -} from "@/lib/runtime.ts"; +import { getCliRuntime, refreshCliRuntimeContext, runWithCliRuntime } from "@/lib/runtime.ts"; +import { createCliTestRuntime } from "@tests/cli-test-runtime.ts"; import { lakehouseAppendOperation, lakehouseAppendTaskOperation, @@ -75,6 +71,12 @@ function createOperationContext(): OperationContext { }; } +async function runCommandWithTestRuntime(rawArgs: string[]): Promise { + await runWithCliRuntime(createCliTestRuntime(), () => + runCommand(buildMainCommand(), { rawArgs }), + ); +} + async function collectLakehouseQueryStream( stream: ReadableStream, ): Promise { @@ -330,21 +332,8 @@ describe("lakehouse request construction", () => { ]), ); - const previousRuntime = getCliRuntime(); - const runtime = createCliRuntime({ debug: false, json: true, agent: false }); - runtime.output.writeJson = () => {}; - runtime.output.writeRaw = () => {}; - runtime.output.writeHuman = () => {}; - setCliRuntime(runtime); - - try { - await runCommand(buildMainCommand(), { rawArgs: ["query", "show", queryId] }); - await runCommand(buildMainCommand(), { - rawArgs: ["query", "cancel", queryId, "--session-id", "session-1"], - }); - } finally { - setCliRuntime(previousRuntime); - } + await runCommandWithTestRuntime(["query", "show", queryId]); + await runCommandWithTestRuntime(["query", "cancel", queryId, "--session-id", "session-1"]); const logContent = readFileSync(logFile, "utf8"); expect(logContent).toContain("METHOD=GET"); @@ -459,34 +448,21 @@ describe("lakehouse request construction", () => { ]), ); - const previousRuntime = getCliRuntime(); - const runtime = createCliRuntime({ debug: false, json: true, agent: false }); - runtime.output.writeJson = () => {}; - runtime.output.writeRaw = () => {}; - runtime.output.writeHuman = () => {}; - setCliRuntime(runtime); - - try { - await runCommand(buildMainCommand(), { - rawArgs: [ - "upload", - "--catalog", - "memory", - "--schema", - "main", - "--table", - "users", - "--format", - "csv", - "--mode", - "overwrite", - "--file", - uploadFile, - ], - }); - } finally { - setCliRuntime(previousRuntime); - } + await runCommandWithTestRuntime([ + "upload", + "--catalog", + "memory", + "--schema", + "main", + "--table", + "users", + "--format", + "csv", + "--mode", + "overwrite", + "--file", + uploadFile, + ]); const logContent = readFileSync(logFile, "utf8"); expect(logContent).toContain("URL=https://example.com/upload?"); @@ -499,23 +475,21 @@ describe("lakehouse request construction", () => { mkdirSync(directoryPath); try { - await runCommand(buildMainCommand(), { - rawArgs: [ - "upload", - "--catalog", - "memory", - "--schema", - "main", - "--table", - "users", - "--format", - "csv", - "--mode", - "overwrite", - "--file", - join(testHome, "missing.csv"), - ], - }); + await runCommandWithTestRuntime([ + "upload", + "--catalog", + "memory", + "--schema", + "main", + "--table", + "users", + "--format", + "csv", + "--mode", + "overwrite", + "--file", + join(testHome, "missing.csv"), + ]); throw new Error("expected missing file failure"); } catch (error) { expect(error).toBeInstanceOf(Error); @@ -523,23 +497,21 @@ describe("lakehouse request construction", () => { } try { - await runCommand(buildMainCommand(), { - rawArgs: [ - "upload", - "--catalog", - "memory", - "--schema", - "main", - "--table", - "users", - "--format", - "csv", - "--mode", - "overwrite", - "--file", - directoryPath, - ], - }); + await runCommandWithTestRuntime([ + "upload", + "--catalog", + "memory", + "--schema", + "main", + "--table", + "users", + "--format", + "csv", + "--mode", + "overwrite", + "--file", + directoryPath, + ]); throw new Error("expected directory failure"); } catch (error) { expect(error).toBeInstanceOf(Error); From e0cab158608213013c67f19d57af0cca332ef306 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fran=C3=A7ois=20Chalifour?= Date: Tue, 30 Jun 2026 16:37:34 +0200 Subject: [PATCH 3/3] test(cli): reuse shared test runtime helper --- cli/tests/api.test.ts | 9 +++------ cli/tests/cli-test-runtime.ts | 10 +++++++++- cli/tests/lakehouse.test.ts | 12 ++---------- 3 files changed, 14 insertions(+), 17 deletions(-) diff --git a/cli/tests/api.test.ts b/cli/tests/api.test.ts index e689373..c1e97f9 100644 --- a/cli/tests/api.test.ts +++ b/cli/tests/api.test.ts @@ -15,6 +15,7 @@ import { OPENAPI_OPERATIONS } from "@/generated/openapi-operations.ts"; import { setCliContext } from "@/context.ts"; import { buildCompletionSpec, flattenTopLevelNames } from "@/lib/completion-spec.ts"; import { createCliRuntime, getCliRuntime, setCliRuntime } from "@/lib/runtime.ts"; +import { runCommandWithTestRuntime } from "@tests/cli-test-runtime.ts"; function createCaptureSink(json: boolean) { const stdout: string[] = []; @@ -214,9 +215,7 @@ describe("api", () => { ]), ); - await runCommand(buildMainCommand(), { - rawArgs: normalizeApiInvocatorRawArgs(["api", "/whoami"]), - }); + await runCommandWithTestRuntime(normalizeApiInvocatorRawArgs(["api", "/whoami"])); const logContent = readFileSync(logFile, "utf8"); expect(logContent).toContain("/rest/v1/whoami"); @@ -235,9 +234,7 @@ describe("api", () => { ]), ); - await runCommand(buildMainCommand(), { - rawArgs: ["api", "POST", "/service_accounts", "-f", "label=CI Bot"], - }); + await runCommandWithTestRuntime(["api", "POST", "/service_accounts", "-f", "label=CI Bot"]); const logContent = readFileSync(logFile, "utf8"); const payloadLines = logContent.match(/^PAYLOAD=.*$/gm) ?? []; diff --git a/cli/tests/cli-test-runtime.ts b/cli/tests/cli-test-runtime.ts index 9f5b499..d59a3bb 100644 --- a/cli/tests/cli-test-runtime.ts +++ b/cli/tests/cli-test-runtime.ts @@ -1,5 +1,7 @@ +import { runCommand } from "citty"; +import { buildMainCommand } from "@/cli.ts"; import type { CliContext } from "@/context.ts"; -import { createCliRuntime, type CliRuntime } from "@/lib/runtime.ts"; +import { createCliRuntime, runWithCliRuntime, type CliRuntime } from "@/lib/runtime.ts"; export function createCliTestRuntime( context: CliContext = { debug: false, json: true, agent: false }, @@ -12,3 +14,9 @@ export function createCliTestRuntime( runtime.output.writeMetadata = () => {}; return runtime; } + +export async function runCommandWithTestRuntime(rawArgs: string[]): Promise { + await runWithCliRuntime(createCliTestRuntime(), () => + runCommand(buildMainCommand(), { rawArgs }), + ); +} diff --git a/cli/tests/lakehouse.test.ts b/cli/tests/lakehouse.test.ts index 568cb9b..3f160ac 100644 --- a/cli/tests/lakehouse.test.ts +++ b/cli/tests/lakehouse.test.ts @@ -1,9 +1,7 @@ import { describe, expect, test, beforeEach, afterEach } from "bun:test"; -import { runCommand } from "citty"; import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import { join } from "node:path"; import { tmpdir } from "node:os"; -import { buildMainCommand } from "@/cli.ts"; import { setCliContext } from "@/context.ts"; import { ParseError } from "@/lib/errors.ts"; import { @@ -28,8 +26,8 @@ import { runOperationPlan, } from "@/lib/operation-effect.ts"; import type { OperationContext } from "@/lib/operation-command.ts"; -import { getCliRuntime, refreshCliRuntimeContext, runWithCliRuntime } from "@/lib/runtime.ts"; -import { createCliTestRuntime } from "@tests/cli-test-runtime.ts"; +import { getCliRuntime, refreshCliRuntimeContext } from "@/lib/runtime.ts"; +import { runCommandWithTestRuntime } from "@tests/cli-test-runtime.ts"; import { lakehouseAppendOperation, lakehouseAppendTaskOperation, @@ -71,12 +69,6 @@ function createOperationContext(): OperationContext { }; } -async function runCommandWithTestRuntime(rawArgs: string[]): Promise { - await runWithCliRuntime(createCliTestRuntime(), () => - runCommand(buildMainCommand(), { rawArgs }), - ); -} - async function collectLakehouseQueryStream( stream: ReadableStream, ): Promise {