diff --git a/cli/test/headers_env_var.test.ts b/cli/test/headers_env_var.test.ts index ad483f4d05..72fdf4103f 100644 --- a/cli/test/headers_env_var.test.ts +++ b/cli/test/headers_env_var.test.ts @@ -18,6 +18,8 @@ import { expect, test } from "bun:test"; import { mkdir, writeFile } from "node:fs/promises"; +import { dirname, join } from "node:path"; +import { fileURLToPath } from "node:url"; import type { Server } from "bun"; import { withTestBackend } from "./test_backend.ts"; @@ -113,12 +115,16 @@ async function runCliThroughProxy( cwd: string, env: Record, ): Promise<{ stdout: string; stderr: string; code: number }> { - const cliDir = new URL("..", import.meta.url).pathname; + // Use fileURLToPath + node:path so the CLI entrypoint is a real OS path on + // Windows. `new URL(..).pathname` yields `/C:/...` which Bun.spawn fails to + // resolve, so the negative case "rejected" on launch instead of reaching + // the proxy and the assertion on rejectedRequests > 0 flaked. + const cliDir = join(dirname(fileURLToPath(import.meta.url)), ".."); const useNode = process.env["TEST_CLI_RUNTIME"] === "node"; const runtime = useNode ? "node" : "bun"; const entrypoint = useNode - ? `${cliDir}/npm/esm/main.js` - : `${cliDir}/src/main.ts`; + ? join(cliDir, "npm", "esm", "main.js") + : join(cliDir, "src", "main.ts"); const runtimeArgs = useNode ? [entrypoint] : ["run", entrypoint]; const fullArgs = [ diff --git a/cli/test/mixed_case_paths.test.ts b/cli/test/mixed_case_paths.test.ts index d85f81cc83..872c6f94ff 100644 --- a/cli/test/mixed_case_paths.test.ts +++ b/cli/test/mixed_case_paths.test.ts @@ -130,6 +130,57 @@ async function createFlow( throw new Error(`Failed to create flow ${flowPath}: ${error}`); } await response.text(); + + // Flow creation queues an async FlowDependencies job that generates the + // inline-script lockfile and rewrites flow.value. If we don't wait, a + // subsequent pull/push races the worker and the dry-run idempotency + // assertion sees a phantom lock-add + flow.yaml-edit diff (CI-only flake). + await waitForFlowDependencyJob(backend, flowPath); +} + +async function waitForFlowDependencyJob( + backend: any, + flowPath: string, + timeoutMs: number = 30000, +): Promise { + // /flows/get does not return `dependency_job`. The deployment_status route + // joins flow_version against deployment_metadata, which is populated in the + // same tx as the FlowDependencies push, so by the time the create/update + // API call returns, job_id is already the latest dep-job UUID. + const start = Date.now(); + while (Date.now() - start < timeoutMs) { + const statusResp = await backend.apiRequest!( + `/api/w/${backend.workspace}/flows/deployment_status/p/${flowPath}`, + ); + if (statusResp.status === 404) { + await statusResp.text().catch(() => {}); + return; + } + if (!statusResp.ok) { + await statusResp.text().catch(() => {}); + throw new Error( + `Failed to fetch deployment status for ${flowPath}: ${statusResp.status}`, + ); + } + const status = await statusResp.json(); + const depJobId: string | undefined = status?.job_id; + if (!depJobId) { + await new Promise((r) => setTimeout(r, 100)); + continue; + } + const completed = await backend.apiRequest!( + `/api/w/${backend.workspace}/jobs_u/completed/get/${depJobId}`, + ); + if (completed.ok) { + await completed.text(); + return; + } + await completed.text().catch(() => {}); + await new Promise((r) => setTimeout(r, 100)); + } + throw new Error( + `Flow dependency job for ${flowPath} did not complete within ${timeoutMs}ms`, + ); } async function createApp( @@ -384,6 +435,10 @@ excludes: [] const updatedFlow = await getFlow(backend, flowPath); expect(updatedFlow.summary).toEqual("Modified Data Processor Flow from test"); + // The CLI push enqueues a fresh FlowDependencies job. Wait for it before + // the dry-run pull so the lock/flow.value writes have committed. + await waitForFlowDependencyJob(backend, flowPath); + // Verify no diff on subsequent pull (idempotency) await verifyNoDiffOnPull(backend, tempDir); }); diff --git a/cli/test/sync_pull_push.test.ts b/cli/test/sync_pull_push.test.ts index e64af246f2..e31c4e1084 100644 --- a/cli/test/sync_pull_push.test.ts +++ b/cli/test/sync_pull_push.test.ts @@ -493,6 +493,58 @@ async function cleanupTempDir(dir: string): Promise { } } +// Polls /flows/deployment_status/p/{path} + /jobs_u/completed/get until the +// most recent flow dependency job has completed. After a flow create/update +// the API queues a FlowDependencies job that asynchronously fills inline- +// script lockfiles and rewrites flow.value; tests that round-trip through +// sync push/pull must wait for it or they race the worker (CI-only flake). +// `deployment_status` is the right read here — `/flows/get` does not return +// `dependency_job`, but the `deployment_metadata.job_id` row is written in +// the same tx as the dep-job push, so the returned `job_id` is the latest +// dep-job UUID by the time the create/update API call has returned. Returns +// silently if the flow doesn't exist on the server (treats it as a no-op +// push, since the route returns 404 when the flow row is missing). +async function waitForFlowDependencyJob( + backend: any, + flowPath: string, + timeoutMs: number = 30000, +): Promise { + const start = Date.now(); + while (Date.now() - start < timeoutMs) { + const statusResp = await backend.apiRequest!( + `/api/w/${backend.workspace}/flows/deployment_status/p/${flowPath}`, + ); + if (statusResp.status === 404) { + await statusResp.text().catch(() => {}); + return; + } + if (!statusResp.ok) { + await statusResp.text().catch(() => {}); + throw new Error( + `Failed to fetch deployment status for ${flowPath}: ${statusResp.status}`, + ); + } + const status = await statusResp.json(); + const depJobId: string | undefined = status?.job_id; + if (!depJobId) { + await new Promise((r) => setTimeout(r, 100)); + continue; + } + const completed = await backend.apiRequest!( + `/api/w/${backend.workspace}/jobs_u/completed/get/${depJobId}`, + ); + if (completed.ok) { + await completed.text(); + return; + } + await completed.text().catch(() => {}); + await new Promise((r) => setTimeout(r, 100)); + } + throw new Error( + `Flow dependency job for ${flowPath} did not complete within ${timeoutMs}ms`, + ); +} + // ============================================================================= // Tests // ============================================================================= @@ -1797,6 +1849,12 @@ excludes: [] expect(pushResult.code).toEqual(0); + // sync push of the flow enqueues an async FlowDependencies job that + // generates the inline-script lockfile and rewrites flow.value. Wait for + // it before pulling back, otherwise pull races the worker and the + // dry-run push idempotency check sees phantom diffs (CI-only flake). + await waitForFlowDependencyJob(backend, flowName); + // Pull back const pullResult = await backend.runCLICommand( ["sync", "pull", "--yes", "--includes", `f/test/mixed_*_${uniqueId}**`],