From b35832ff89db3c6a1eca1cc5e5f131790e335565 Mon Sep 17 00:00:00 2001 From: Jesus Orosco Date: Fri, 6 Nov 2026 14:32:53 -0800 Subject: [PATCH 1/7] feat: support StreamingTransform scripts in init, scan, and deploy --- README.md | 14 ++ messages/datacodeBinaryExecutor.md | 13 ++ messages/init.md | 6 + messages/scan.md | 6 + package.json | 3 +- src/base/initBase.ts | 6 + src/base/scanBase.ts | 6 +- src/utils/nativeDeploy.ts | 52 +++++- src/utils/nativeInit.ts | 73 +++++++- src/utils/nativeScan.ts | 169 +++++++++++++++--- .../commands/data-code-extension/init.test.ts | 66 +++++++ .../commands/data-code-extension/scan.test.ts | 36 ++++ test/e2e/expected/streaming/config.json | 17 ++ test/e2e/expected/streaming/requests.json | 64 +++++++ test/e2e/expected/streaming/zip-entries.json | 1 + test/e2e/fixtures/streaming/entrypoint.py | 23 +++ test/e2e/smoke.e2e.ts | 89 +++++++++ test/utils/nativeDeploy.test.ts | 101 +++++++++++ test/utils/nativeInit.test.ts | 122 +++++++++++++ test/utils/nativeScan.test.ts | 126 +++++++++++++ yarn.lock | 5 + 21 files changed, 965 insertions(+), 33 deletions(-) create mode 100644 test/e2e/expected/streaming/config.json create mode 100644 test/e2e/expected/streaming/requests.json create mode 100644 test/e2e/expected/streaming/zip-entries.json create mode 100644 test/e2e/fixtures/streaming/entrypoint.py diff --git a/README.md b/README.md index eaf8529..9e52f44 100644 --- a/README.md +++ b/README.md @@ -290,6 +290,11 @@ DESCRIPTION Scans Python files in an initialized Data Code Extension package directory to identify required permissions and dependencies. Updates the config.json and requirements.txt files based on the code analysis. + For a script that calls the streaming delta methods (read_dlo_deltas or read_dmo_deltas, and write_dlo_deltas or + write_dmo_deltas), scan writes a streaming config instead. The read source can't be detected from the code, so scan + keeps streamingSource.name from config.json and uses it as the only read permission. Write permissions come from the + delta writes. + EXAMPLES Scan with a custom entrypoint file: @@ -481,6 +486,10 @@ FLAG DESCRIPTIONS Which transform feature this script is used in: BatchTransform (a bounded, batch data transform; the default) or StreamingTransform (a streaming data transform over a DLO/DMO change feed). Defaults to BatchTransform. + + StreamingTransform starts from the SDK's streaming example entrypoint, adds an empty streamingSource to config.json, + and writes a config.yaml that switches 'run' to the local delta reader and writer. Requires + salesforce-data-customcode 7.0.0 or later. ``` _See code: [src/commands/data-code-extension/script/init.ts](https://github.com/salesforcecli/plugin-data-code-extension/blob/1.6.3/src/commands/data-code-extension/script/init.ts)_ @@ -566,6 +575,11 @@ DESCRIPTION Scans Python files in an initialized Data Code Extension package directory to identify required permissions and dependencies. Updates the config.json and requirements.txt files based on the code analysis. + For a script that calls the streaming delta methods (read_dlo_deltas or read_dmo_deltas, and write_dlo_deltas or + write_dmo_deltas), scan writes a streaming config instead. The read source can't be detected from the code, so scan + keeps streamingSource.name from config.json and uses it as the only read permission. Write permissions come from the + delta writes. + EXAMPLES Scan with a custom entrypoint file: diff --git a/messages/datacodeBinaryExecutor.md b/messages/datacodeBinaryExecutor.md index d5f7ac1..33d811d 100644 --- a/messages/datacodeBinaryExecutor.md +++ b/messages/datacodeBinaryExecutor.md @@ -8,6 +8,19 @@ Could not locate the datacustomcode init template at '%s'. - Run 'pip show salesforce-data-customcode' to confirm the install location - Reinstall with 'pip install --upgrade salesforce-data-customcode' if the templates directory is missing +# error.initStreamingTemplateNotFound + +The installed salesforce-data-customcode package (version %s) doesn't include the streaming template at '%s'. StreamingTransform needs version 7.0.0 or later. + +# actions.initStreamingTemplateNotFound + +- Upgrade with 'pip install --upgrade salesforce-data-customcode' +- Run 'pip show salesforce-data-customcode' to confirm the installed version + +# error.initInvalidScriptFeature + +Invalid use-in-feature '%s' for a script package. Valid values: %s. + # error.scanEntrypointMissing Entrypoint file not found at '%s'. diff --git a/messages/init.md b/messages/init.md index 8f6dbc4..8332574 100644 --- a/messages/init.md +++ b/messages/init.md @@ -83,3 +83,9 @@ Invoke option for the script package (script packages only). # flags.useInFeatureScript.description Which transform feature this script is used in: BatchTransform (a bounded, batch data transform; the default) or StreamingTransform (a streaming data transform over a DLO/DMO change feed). Defaults to BatchTransform. + +StreamingTransform starts from the SDK's streaming example entrypoint, adds an empty streamingSource to config.json, and writes a config.yaml that switches 'run' to the local delta reader and writer. Requires salesforce-data-customcode 7.0.0 or later. + +# warning.streamingSourceNameEmpty + +Set streamingSource.name in %s to the DLO or DMO the streaming transform reads from, then run 'data-code-extension script scan'. diff --git a/messages/scan.md b/messages/scan.md index 0f833e5..da2f551 100644 --- a/messages/scan.md +++ b/messages/scan.md @@ -6,6 +6,8 @@ Scan the Data Code Extension %s package for permissions and dependencies. Scans Python files in an initialized Data Code Extension package directory to identify required permissions and dependencies. Updates the config.json and requirements.txt files based on the code analysis. +For a script that calls the streaming delta methods (read_dlo_deltas or read_dmo_deltas, and write_dlo_deltas or write_dmo_deltas), scan writes a streaming config instead. The read source can't be detected from the code, so scan keeps streamingSource.name from config.json and uses it as the only read permission. Write permissions come from the delta writes. + # examples.script - Scan with a custom entrypoint file: @@ -115,6 +117,10 @@ Skips the confirmation scan asks for when the rebuilt permissions no longer cove dataLineage references objects that are no longer in permissions (%s entries affected). The package will fail to deploy (1041/1042) until you rerun 'data-code-extension script lineage generate' or edit dataLineage. Scan doesn't change dataLineage. %s +# warning.streamingSourceNameEmpty + +streamingSource.name is empty in config.json. A streaming transform must declare its read source; set streamingSource.name to the DLO or DMO the transform reads from, then scan again. + # prompt.staleLineage Write the updated permissions to config.json anyway? diff --git a/package.json b/package.json index ae14dd6..a540c83 100644 --- a/package.json +++ b/package.json @@ -7,7 +7,8 @@ "@salesforce/core": "^8.2.7", "@salesforce/sf-plugins-core": "^12", "entities": "^4.5.0", - "jszip": "^3.10.1" + "jszip": "^3.10.1", + "yaml": "^2.9.1" }, "devDependencies": { "@oclif/plugin-command-snapshot": "^5.2.3", diff --git a/src/base/initBase.ts b/src/base/initBase.ts index adf18f2..93543fb 100644 --- a/src/base/initBase.ts +++ b/src/base/initBase.ts @@ -17,6 +17,7 @@ import { SfCommand } from '@salesforce/sf-plugins-core'; import { Messages } from '@salesforce/core'; import { executeNativeInit, type NativeInitResult } from '../utils/nativeInit.js'; import { checkEnvironment } from '../utils/environmentChecker.js'; +import { SCRIPT_USE_IN_FEATURE_STREAMING } from '../utils/constants.js'; import { type SharedResultProps } from './types.js'; export type BaseInitFlags = { @@ -57,6 +58,11 @@ export abstract class InitBase ext this.log(messages.getMessage('info.fileCreated', [file])); }); + // The streaming read source can't be scanned from the code, so the new config.json leaves it empty. + if (codeType === 'script' && additionalFlags.useInFeature === SCRIPT_USE_IN_FEATURE_STREAMING) { + this.warn(messages.getMessage('warning.streamingSourceNameEmpty', [executionResult.configPath])); + } + return { success: true, pythonVersion: pythonInfo, diff --git a/src/base/scanBase.ts b/src/base/scanBase.ts index 698ea78..ba967bc 100644 --- a/src/base/scanBase.ts +++ b/src/base/scanBase.ts @@ -16,7 +16,7 @@ import path from 'node:path'; import { SfCommand } from '@salesforce/sf-plugins-core'; import { Messages, SfError } from '@salesforce/core'; -import { executeNativeScan, type NativeScanResult } from '../utils/nativeScan.js'; +import { executeNativeScan, isStreamingSourceNameMissing, type NativeScanResult } from '../utils/nativeScan.js'; import { checkEnvironment, type EnvironmentCheckResult } from '../utils/environmentChecker.js'; import { countEntriesReferencing, @@ -71,6 +71,10 @@ export abstract class ScanBase extends SfCommand { }); this.spinner.stop(); + if (codeType === 'script' && isStreamingSourceNameMissing(executionResult.config)) { + this.warn(messages.getMessage('warning.streamingSourceNameEmpty')); + } + if (codeType === 'script' && executionResult.dryRun) { const stale = findStaleLineage(executionResult.config); if (hasStaleLineage(stale)) this.warnStaleLineage(messages, executionResult.config, stale); diff --git a/src/utils/nativeDeploy.ts b/src/utils/nativeDeploy.ts index 9bf96af..1161a5a 100644 --- a/src/utils/nativeDeploy.ts +++ b/src/utils/nativeDeploy.ts @@ -19,6 +19,7 @@ import path from 'node:path'; import { debuglog } from 'node:util'; import { Messages, SfError, type Connection } from '@salesforce/core'; import { decodeHTML } from 'entities'; +import { SCRIPT_USE_IN_FEATURE_STREAMING } from './constants.js'; import { findBaseDirectory, getPackageType, type CodeType } from './nativeScan.js'; import { zipWithSfError, ZIP_FILE_NAME, type ZipResult } from './zipBuilder.js'; @@ -27,9 +28,11 @@ const debug = debuglog('datacustomcode'); Messages.importMessagesDirectoryFromMetaUrl(import.meta.url); const messages = Messages.loadMessages('@salesforce/plugin-data-code-extension', 'deploy'); -// Mirror of the Data Cloud REST paths in `datacustomcode/deploy.py`. Kept at -// v63.0 to match the SDK's `main` branch (streaming / v67.0 is out of scope). +// Mirror of the Data Cloud REST paths in `datacustomcode/deploy.py`. const DATA_CUSTOM_CODE_PATH = 'services/data/v63.0/ssot/data-custom-code'; +// `invokeOptions` only binds at v67.0, so deployment creation routes there when +// it is set (streaming scripts). Everything else, including status polling, stays on v63.0. +const DATA_CUSTOM_CODE_INVOKE_OPTIONS_PATH = 'services/data/v67.0/ssot/data-custom-code'; const DATA_TRANSFORMS_PATH = 'services/data/v63.0/ssot/data-transforms'; // Python waits up to 3000 *seconds* (`WAIT_FOR_DEPLOYMENT_TIMEOUT`), polling once a second. const WAIT_FOR_DEPLOYMENT_TIMEOUT_MS = 3000 * 1000; @@ -73,6 +76,7 @@ export type CodeExtensionMetadata = { computeType: string; codeType: CodeType; functionInvokeOptions?: string[]; + invokeOptions?: string[]; }; export type CreateDeploymentResponse = { @@ -103,8 +107,12 @@ export type DataTransformConfig = { dataspace: string; permissions: { read: Permission; write: Permission }; dataObjects?: DataObject[]; + /** Present only for streaming transforms; its presence makes the transform STREAMING. */ + streamingSource?: StreamingSource; }; +export type StreamingSource = { type: string; name: string }; + export type FunctionConfig = { entryPoint: string; }; @@ -167,7 +175,7 @@ export function sanitizeApiName(name: string): string { */ export function buildMetadata( input: { name: string; version: string; description: string; computeType: string; codeType: CodeType }, - log: (message: string) => void = (): void => { } + log: (message: string) => void = (): void => {} ): CodeExtensionMetadata { const sanitized = sanitizeApiName(input.name); if (sanitized !== input.name) { @@ -380,7 +388,7 @@ export function buildDataTransformBody( definition, label: metadata.name, name: metadata.name, - type: 'BATCH', + type: config.streamingSource ? 'STREAMING' : 'BATCH', dataSpaceName: config.dataspace, }; } @@ -458,9 +466,25 @@ function validateDataTransformConfig(parsed: unknown, configPath: string): DataT dataspace: String(parsed.dataspace), permissions: { read, write }, dataObjects: parsed.dataObjects as DataObject[] | undefined, + ...parseStreamingSource(parsed.streamingSource, configPath), }; } +/** Mirror of `StreamingSource` in `datacustomcode/deploy.py`: optional, but when present needs string `type`/`name`. */ +function parseStreamingSource(value: unknown, configPath: string): { streamingSource?: StreamingSource } { + if (value === undefined || value === null) return {}; + if (!isRecord(value) || typeof value.type !== 'string' || typeof value.name !== 'string') { + throw new SfError( + messages.getMessage('error.configInvalid', [ + configPath, + "streamingSource must be an object with string 'type' and 'name'", + ]), + 'InvalidConfig' + ); + } + return { streamingSource: { type: value.type, name: value.name } }; +} + function validateFunctionConfig(parsed: unknown, configPath: string): FunctionConfig { if (!isRecord(parsed) || typeof parsed.entryPoint !== 'string') { throw new SfError(messages.getMessage('error.configMissingFields', [configPath, 'entryPoint']), 'InvalidConfig'); @@ -610,10 +634,19 @@ function defaultCreateDeployment(connection: Connection): NativeDeployDeps['crea if (metadata.functionInvokeOptions && metadata.functionInvokeOptions.length > 0) { body.functionInvokeOptions = metadata.functionInvokeOptions; } + const hasInvokeOptions = metadata.invokeOptions !== undefined && metadata.invokeOptions.length > 0; + if (hasInvokeOptions) { + body.invokeOptions = metadata.invokeOptions; + } let response: CreateDeploymentResponse; try { - response = await requestJson(connection, 'POST', DATA_CUSTOM_CODE_PATH, body); + response = await requestJson( + connection, + 'POST', + hasInvokeOptions ? DATA_CUSTOM_CODE_INVOKE_OPTIONS_PATH : DATA_CUSTOM_CODE_PATH, + body + ); } catch (err) { if (isConflictError(err)) { throw new SfError(messages.getMessage('error.deploymentExists', [metadata.name]), 'DeploymentExists'); @@ -671,7 +704,7 @@ export async function waitForDeployment( onStatus?: (status: string) => void ): Promise { const start = deps.now(); - for (; ;) { + for (;;) { // eslint-disable-next-line no-await-in-loop const status = await deps.getDeploymentStatus(name); if (deps.now() - start > WAIT_FOR_DEPLOYMENT_TIMEOUT_MS) { @@ -701,7 +734,7 @@ export class NativeDeployer { opts: NativeDeployOptions, overrides: Partial = {} ): Promise { - const log = opts.log ?? ((): void => { }); + const log = opts.log ?? ((): void => {}); const computeType = COMPUTE_TYPES[opts.cpuSize]; if (!computeType) { @@ -738,6 +771,11 @@ export class NativeDeployer { // Validate config.json up front so a bad package fails before any API call. const config = await getConfig(opts.packageDir, packageType); + // Mirror of `deploy_full`: a config with `streamingSource` deploys as a streaming transform. + if (packageType === 'script' && (config as DataTransformConfig).streamingSource && !metadata.invokeOptions) { + metadata.invokeOptions = [SCRIPT_USE_IN_FEATURE_STREAMING]; + } + const deps = buildDefaultDeps(opts.connection, overrides); log(messages.getMessage('info.creatingDeployment', [metadata.name])); diff --git a/src/utils/nativeInit.ts b/src/utils/nativeInit.ts index d53e5dc..f58aa8d 100644 --- a/src/utils/nativeInit.ts +++ b/src/utils/nativeInit.ts @@ -17,6 +17,12 @@ import { promises as fs } from 'node:fs'; import path from 'node:path'; import { debuglog } from 'node:util'; import { Messages, SfError } from '@salesforce/core'; +import { parseDocument } from 'yaml'; +import { + SCRIPT_USE_IN_FEATURE_BATCH, + SCRIPT_USE_IN_FEATURE_OPTIONS, + SCRIPT_USE_IN_FEATURE_STREAMING, +} from './constants.js'; import { type CodeType, updateConfig } from './nativeScan.js'; const debug = debuglog('datacustomcode'); @@ -45,6 +51,10 @@ const TESTS_DIR = 'tests'; const TEST_FILE = 'test.json'; const SDK_CONFIG_DIR = '.datacustomcode_proj'; const SDK_CONFIG_FILE = 'sdk_config.json'; +// Mirror of `datacustomcode/template.py`: the streaming example entrypoint, and the SDK's +// run config, which streaming packages get a copy of with the local delta reader/writer. +const STREAMING_EXAMPLE_ENTRYPOINT = path.join('examples', 'streaming_deltas', ENTRYPOINT_FILE); +const SDK_RUN_CONFIG_FILE = 'config.yaml'; const FEATURE_TEMPLATE_MAPPING: Readonly> = { SearchIndexChunking: 'chunking', @@ -66,12 +76,24 @@ const FEATURE_TEMPLATE_DIRS = new Set(Object.values(FEATURE_TEMPLATE_MAP */ export async function executeNativeInit(opts: NativeInitOptions): Promise { const { codeType, packageDir, pythonPackageLocation, pythonPackageVersion, useInFeature } = opts; - const templatesDir = path.join(pythonPackageLocation, 'datacustomcode', 'templates'); + const sdkDir = path.join(pythonPackageLocation, 'datacustomcode'); + const templatesDir = path.join(sdkDir, 'templates'); const filesCreated: string[] = []; + const streaming = codeType === 'script' && resolveScriptFeature(useInFeature) === SCRIPT_USE_IN_FEATURE_STREAMING; // 1. Copy template if (codeType === 'script') { - await copyScriptTemplate(path.join(templatesDir, 'script'), packageDir); + const scriptTemplateDir = path.join(templatesDir, 'script'); + // Check before copying anything so an older SDK leaves no half-initialized package. + if (streaming) await ensureStreamingTemplateExists(scriptTemplateDir, pythonPackageVersion); + await copyScriptTemplate(scriptTemplateDir, packageDir); + if (streaming) { + await fs.copyFile( + path.join(scriptTemplateDir, STREAMING_EXAMPLE_ENTRYPOINT), + path.join(packageDir, PAYLOAD_DIR, ENTRYPOINT_FILE) + ); + await writeStreamingRunConfig(path.join(sdkDir, SDK_RUN_CONFIG_FILE), path.join(packageDir, SDK_RUN_CONFIG_FILE)); + } } else { await copyFunctionTemplate(path.join(templatesDir, 'function'), packageDir, useInFeature); } @@ -84,7 +106,7 @@ export async function executeNativeInit(opts: NativeInitOptions): Promise { + const entrypoint = path.join(scriptTemplateDir, STREAMING_EXAMPLE_ENTRYPOINT); + if (!(await pathExists(entrypoint))) { + throw new SfError( + messages.getMessage('error.initStreamingTemplateNotFound', [sdkVersion, entrypoint]), + 'InitStreamingTemplateNotFound', + messages.getMessages('actions.initStreamingTemplateNotFound') + ); + } +} + +/** + * Mirror of `datacustomcode/template.py:_write_streaming_config_yaml`: copy the SDK's + * config.yaml with the reader/writer switched to the local delta implementations. Edited + * as a YAML document so comments and formatting in the SDK file are kept. + */ +async function writeStreamingRunConfig(sourcePath: string, destinationPath: string): Promise { + const doc = parseDocument(await fs.readFile(sourcePath, 'utf8')); + doc.setIn(['reader_config', 'type_config_name'], 'LocalDeltasReader'); + doc.setIn(['writer_config', 'type_config_name'], 'LocalDeltasWriter'); + debug('streaming run config: %s -> %s', sourcePath, destinationPath); + await fs.writeFile(destinationPath, doc.toString(), 'utf8'); +} + async function copyTemplateEntries( entries: Array<{ name: string; src: string; dst: string }>, label: string @@ -170,12 +228,19 @@ async function writeSdkConfig(baseDir: string, config: Record): await writeJson(path.join(sdkConfigDir, SDK_CONFIG_FILE), config); } -function buildInitialConfig(codeType: CodeType, entrypointPath: string, sdkVersion: string): Record { +function buildInitialConfig( + codeType: CodeType, + entrypointPath: string, + sdkVersion: string, + streaming: boolean +): Record { if (codeType === 'script') { + // Mirror of DATA_TRANSFORM_CONFIG_TEMPLATE / STREAMING_TRANSFORM_CONFIG_TEMPLATE in `datacustomcode/scan.py`. return { sdkVersion, entryPoint: path.basename(entrypointPath), dataspace: 'default', + ...(streaming ? { streamingSource: { type: 'dlo', name: '' } } : {}), permissions: { read: {}, write: {}, diff --git a/src/utils/nativeScan.ts b/src/utils/nativeScan.ts index f1b3f0b..e3f5b53 100644 --- a/src/utils/nativeScan.ts +++ b/src/utils/nativeScan.ts @@ -29,6 +29,16 @@ export type DataAccessLayerCalls = { writeToDmo: Set; }; +/** Mirror of `datacustomcode/scan.py:StreamingDataAccessLayerCalls`. */ +export type StreamingDataAccessLayerCalls = { + readDloDeltas: boolean; + readDmoDeltas: boolean; + writeDloDeltas: Set; + writeDmoDeltas: Set; +}; + +export type StreamingSource = { type: 'dlo' | 'dmo'; name: string }; + export type ScanPermissions = { read: { dlo?: string[]; dmo?: string[] }; write: { dlo?: string[]; dmo?: string[] }; @@ -73,6 +83,9 @@ const SDK_CONFIG_DIR = '.datacustomcode_proj'; const SDK_CONFIG_FILE = 'sdk_config.json'; const DATA_ACCESS_METHODS = new Set(['read_dlo', 'read_dmo', 'write_to_dlo', 'write_to_dmo']); +// Mirror of STREAMING_READ_METHODS / STREAMING_WRITE_METHODS in `datacustomcode/scan.py`. +const STREAMING_READ_METHODS = new Set(['read_dlo_deltas', 'read_dmo_deltas']); +const STREAMING_WRITE_METHODS = new Set(['write_dlo_deltas', 'write_dmo_deltas']); const EXCLUDED_PACKAGES = new Set(['datacustomcode', 'pyspark']); @@ -164,6 +177,35 @@ export async function getPackageType(baseDirectory: string): Promise { * Reassignments and post-call assignments behave correctly. */ export async function scanFile(filePath: string): Promise { + const { batch } = await visitFile(filePath); + validateAccessLayer(batch); + return batch; +} + +/** + * Mirror of `datacustomcode/scan.py:scan_file_streaming`. Collects the delta calls + * (`read_dlo_deltas()` / `read_dmo_deltas()` flags, and `write_dlo_deltas` / + * `write_dmo_deltas` targets resolved like the batch scan) and validates them. + */ +export async function scanFileStreaming(filePath: string): Promise { + const { streaming } = await visitFile(filePath); + validateStreamingAccessLayer(streaming); + return streaming; +} + +/** Mirror of `datacustomcode/scan.py:file_is_streaming`: true when any delta method is called. */ +export async function fileIsStreaming(filePath: string): Promise { + return isStreaming((await visitFile(filePath)).streaming); +} + +function isStreaming(calls: StreamingDataAccessLayerCalls): boolean { + return calls.readDloDeltas || calls.readDmoDeltas || calls.writeDloDeltas.size > 0 || calls.writeDmoDeltas.size > 0; +} + +/** Single pass over the entrypoint collecting both batch and streaming calls, without validating. */ +async function visitFile( + filePath: string +): Promise<{ batch: DataAccessLayerCalls; streaming: StreamingDataAccessLayerCalls }> { const code = await fs.readFile(filePath, 'utf8'); const masked = computeCommentStringMask(code); @@ -173,6 +215,12 @@ export async function scanFile(filePath: string): Promise writeToDlo: new Set(), writeToDmo: new Set(), }; + const streaming: StreamingDataAccessLayerCalls = { + readDloDeltas: false, + readDmoDeltas: false, + writeDloDeltas: new Set(), + writeDmoDeltas: new Set(), + }; // Collect assignments and method calls as ordered events, then walk in source order. type AssignEvent = { kind: 'assign'; offset: number; name: string; value: string }; @@ -198,12 +246,19 @@ export async function scanFile(filePath: string): Promise events.push({ kind: 'assign', offset: m.index, name: m[1], value: m[2] ?? m[3] ?? '' }); } + // The first argument is optional so the no-arg `read_*_deltas()` calls match; a batch or + // delta-write call without a resolvable first argument is skipped below, like the AST visitor. const callRegex = - /\.(?read_dlo|read_dmo|write_to_dlo|write_to_dmo)\s*\(\s*(?:"(?[^"\\]*)"|'(?[^'\\]*)'|(?[A-Za-z_]\w*))/g; + /\.(?read_dlo_deltas|read_dmo_deltas|write_dlo_deltas|write_dmo_deltas|read_dlo|read_dmo|write_to_dlo|write_to_dmo)\s*\(\s*(?:"(?[^"\\]*)"|'(?[^'\\]*)'|(?[A-Za-z_]\w*))?/g; while ((m = callRegex.exec(code)) !== null) { if (masked[m.index]) continue; // call lives inside a comment/string const g = m.groups as { method: string; dq?: string; sq?: string; ident?: string }; - if (!DATA_ACCESS_METHODS.has(g.method)) continue; + if (STREAMING_READ_METHODS.has(g.method)) { + if (g.method === 'read_dlo_deltas') streaming.readDloDeltas = true; + else streaming.readDmoDeltas = true; + continue; + } + if (!DATA_ACCESS_METHODS.has(g.method) && !STREAMING_WRITE_METHODS.has(g.method)) continue; const literal = g.dq ?? g.sq; events.push({ kind: 'call', @@ -216,6 +271,16 @@ export async function scanFile(filePath: string): Promise events.sort((a, b) => a.offset - b.offset); + /* eslint-disable camelcase -- keys are the SDK client method names */ + const targets: Record> = { + read_dlo: result.readDlo, + read_dmo: result.readDmo, + write_to_dlo: result.writeToDlo, + write_to_dmo: result.writeToDmo, + write_dlo_deltas: streaming.writeDloDeltas, + write_dmo_deltas: streaming.writeDmoDeltas, + }; + /* eslint-enable camelcase */ const variableValues = new Map(); for (const ev of events) { if (ev.kind === 'assign') { @@ -230,24 +295,10 @@ export async function scanFile(filePath: string): Promise } if (!name) continue; - switch (ev.method) { - case 'read_dlo': - result.readDlo.add(name); - break; - case 'read_dmo': - result.readDmo.add(name); - break; - case 'write_to_dlo': - result.writeToDlo.add(name); - break; - case 'write_to_dmo': - result.writeToDmo.add(name); - break; - } + targets[ev.method]?.add(name); } - validateAccessLayer(result); - return result; + return { batch: result, streaming }; } /** @@ -324,6 +375,44 @@ function validateAccessLayer(calls: DataAccessLayerCalls): void { } } +function streamingReadLayer(calls: StreamingDataAccessLayerCalls): 'dlo' | 'dmo' { + return calls.readDloDeltas ? 'dlo' : 'dmo'; +} + +function streamingWriteLayer(calls: StreamingDataAccessLayerCalls): 'dlo' | 'dmo' { + return calls.writeDloDeltas.size > 0 ? 'dlo' : 'dmo'; +} + +/** Mirror of `StreamingDataAccessLayerCalls.validate_access_layer` (same messages). */ +function validateStreamingAccessLayer(calls: StreamingDataAccessLayerCalls): void { + if (calls.readDloDeltas && calls.readDmoDeltas) { + throw new SfError('Cannot read DLO and DMO deltas in the same streaming transform.', 'InvalidEntrypoint'); + } + if (!calls.readDloDeltas && !calls.readDmoDeltas) { + throw new SfError( + 'A streaming transform must read from at least one DLO or DMO delta stream (read_dlo_deltas / read_dmo_deltas).', + 'InvalidEntrypoint' + ); + } + if (calls.writeDloDeltas.size === 0 && calls.writeDmoDeltas.size === 0) { + throw new SfError( + 'A streaming transform must write to at least one DLO or DMO via write_dlo_deltas / write_dmo_deltas.', + 'InvalidEntrypoint' + ); + } + if (calls.writeDloDeltas.size > 0 && calls.writeDmoDeltas.size > 0) { + throw new SfError('Cannot write DLO and DMO deltas in the same streaming transform.', 'InvalidEntrypoint'); + } + const read = streamingReadLayer(calls); + const write = streamingWriteLayer(calls); + if (read !== write) { + throw new SfError( + `A streaming transform must read and write the same layer (got read=${read}, write=${write}).`, + 'InvalidEntrypoint' + ); + } +} + let pipreqsMock: ((fileDir: string) => Promise) | null = null; /** @@ -428,13 +517,53 @@ export async function updateConfig( if (packageType === 'script') { existing.dataspace = resolveDataspace(existing); - const calls = await scanFile(entrypointPath); - existing.permissions = buildPermissions(calls); + const { batch, streaming } = await visitFile(entrypointPath); + if (isStreaming(streaming)) { + updateStreamingConfig(existing, streaming); + } else { + delete existing.streamingSource; + validateAccessLayer(batch); + existing.permissions = buildPermissions(batch); + } } return existing; } +/** + * Mirror of `datacustomcode/scan.py:_update_streaming_config`. The streaming read source + * cannot be inferred from `read_*_deltas()`, so `streamingSource.name` is kept from the + * existing config and is the only read permission; writes come from the scan. + */ +function updateStreamingConfig(existing: Record, calls: StreamingDataAccessLayerCalls): void { + validateStreamingAccessLayer(calls); + const readLayer = streamingReadLayer(calls); + const source = existing.streamingSource; + const name = isPlainObject(source) && typeof source.name === 'string' ? source.name : ''; + const streamingSource: StreamingSource = { type: readLayer, name }; + existing.streamingSource = streamingSource; + existing.permissions = { + read: { [readLayer]: name ? [name] : [] }, + write: { + [streamingWriteLayer(calls)]: [...(calls.writeDloDeltas.size > 0 ? calls.writeDloDeltas : calls.writeDmoDeltas)], + }, + }; +} + +/** + * True when a streaming config has no `streamingSource.name` yet. Callers warn, since a + * streaming transform must declare the DLO/DMO it reads from. + */ +export function isStreamingSourceNameMissing(config: Record): boolean { + const source = config.streamingSource; + if (source === undefined) return false; + return !isPlainObject(source) || typeof source.name !== 'string' || source.name === ''; +} + +function isPlainObject(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + /** Read and parse a package's config.json, with the errors `updateConfig` raises. */ export async function readConfigJson(configPath: string): Promise> { let raw: string; diff --git a/test/commands/data-code-extension/init.test.ts b/test/commands/data-code-extension/init.test.ts index 316ff47..4930d29 100644 --- a/test/commands/data-code-extension/init.test.ts +++ b/test/commands/data-code-extension/init.test.ts @@ -13,11 +13,16 @@ * See the License for the specific language governing permissions and * limitations under the License. */ +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import * as path from 'node:path'; import { TestContext } from '@salesforce/core/testSetup'; import { expect } from 'chai'; import { stubSfCommandUx } from '@salesforce/sf-plugins-core'; import ScriptInit from '../../../src/commands/data-code-extension/script/init.js'; import FunctionInit from '../../../src/commands/data-code-extension/function/init.js'; +import { PythonChecker } from '../../../src/utils/pythonChecker.js'; +import { PipChecker } from '../../../src/utils/pipChecker.js'; describe('data-code-extension init commands', () => { const $$ = new TestContext(); @@ -275,4 +280,65 @@ describe('data-code-extension init commands', () => { expect(message).to.include('StreamingTransform'); } }); + + describe('with a stubbed environment', () => { + let root: string; + + beforeEach(() => { + root = mkdtempSync(path.join(tmpdir(), 'init-command-')); + // Minimal fake SDK install with the script template, streaming example, and config.yaml. + const sdk = path.join(root, 'site-packages', 'datacustomcode'); + const script = path.join(sdk, 'templates', 'script'); + mkdirSync(path.join(script, 'payload'), { recursive: true }); + mkdirSync(path.join(script, 'examples', 'streaming_deltas'), { recursive: true }); + writeFileSync( + path.join(script, 'payload', 'entrypoint.py'), + 'c.read_dlo("a__dll")\nc.write_to_dlo("b__dll", df)\n' + ); + writeFileSync(path.join(script, 'payload', 'config.json'), '{}'); + writeFileSync( + path.join(script, 'examples', 'streaming_deltas', 'entrypoint.py'), + 'df = c.read_dlo_deltas()\nc.write_dlo_deltas("b__dll", df)\n' + ); + writeFileSync( + path.join(sdk, 'config.yaml'), + 'reader_config:\n type_config_name: R\nwriter_config:\n type_config_name: W\n' + ); + + $$.SANDBOX.stub(PythonChecker, 'checkPython311').resolves({ + command: 'python3', + version: '3.11.5', + major: 3, + minor: 11, + patch: 5, + }); + $$.SANDBOX.stub(PipChecker, 'checkPackage').resolves({ + name: 'salesforce-data-customcode', + version: '7.0.0', + location: path.join(root, 'site-packages'), + pipCommand: 'pip3', + }); + }); + + afterEach(() => { + rmSync(root, { recursive: true, force: true }); + }); + + it('warns to set streamingSource.name after a StreamingTransform init', async () => { + const packageDir = path.join(root, 'stream'); + await ScriptInit.run(['--package-dir', packageDir, '--use-in-feature', 'StreamingTransform']); + + const warned = sfCommandStubs.warn + .getCalls() + .map((call) => call.args[0] as string) + .join('\n'); + expect(warned).to.include('streamingSource.name'); + expect(warned).to.include(path.join(packageDir, 'payload', 'config.json')); + }); + + it('does not warn after a BatchTransform init', async () => { + await ScriptInit.run(['--package-dir', path.join(root, 'batch')]); + expect(sfCommandStubs.warn.called).to.be.false; + }); + }); }); diff --git a/test/commands/data-code-extension/scan.test.ts b/test/commands/data-code-extension/scan.test.ts index 6e7bf33..39bc45b 100644 --- a/test/commands/data-code-extension/scan.test.ts +++ b/test/commands/data-code-extension/scan.test.ts @@ -215,6 +215,42 @@ describe('data-code-extension scan commands', () => { }); }); + describe('script scan (streaming)', () => { + const streamingCode = + 'from datacustomcode.client import StreamingClient\n' + + 'def main():\n' + + ' c = StreamingClient()\n' + + ' df = c.read_dlo_deltas()\n' + + ' c.write_dlo_deltas("Dest__dll", df).awaitTermination()\n'; + + it('writes a streaming config and does not warn when streamingSource.name is set', async () => { + makeScriptPackage(streamingCode, { dataspace: 'default', streamingSource: { type: 'dlo', name: 'Source__dll' } }); + + await ScriptScan.run([]); + + const written = JSON.parse(fs.readFileSync(path.join(testDir, 'payload', 'config.json'), 'utf8')) as Record< + string, + unknown + >; + expect(written.streamingSource).to.deep.equal({ type: 'dlo', name: 'Source__dll' }); + expect(written.permissions).to.deep.equal({ read: { dlo: ['Source__dll'] }, write: { dlo: ['Dest__dll'] } }); + expect(sfCommandStubs.warn.called).to.be.false; + }); + + it('warns when streamingSource.name is empty', async () => { + makeScriptPackage(streamingCode, { dataspace: 'default', streamingSource: { type: 'dlo', name: '' } }); + + await ScriptScan.run(['--dry-run']); + + expect( + sfCommandStubs.warn + .getCalls() + .map((call) => call.args[0] as string) + .join('\n') + ).to.include('streamingSource.name is empty'); + }); + }); + describe('function scan', () => { it('updates config.json with entryPoint and skips permissions', async () => { makeFunctionPackage( diff --git a/test/e2e/expected/streaming/config.json b/test/e2e/expected/streaming/config.json new file mode 100644 index 0000000..dd7927a --- /dev/null +++ b/test/e2e/expected/streaming/config.json @@ -0,0 +1,17 @@ +{ + "sdkVersion": "", + "entryPoint": "entrypoint.py", + "dataspace": "default", + "streamingSource": { + "type": "dlo", + "name": "Smoke_Stream_Source__dll" + }, + "permissions": { + "read": { + "dlo": ["Smoke_Stream_Source__dll"] + }, + "write": { + "dlo": ["Smoke_Stream_Target__dll"] + } + } +} diff --git a/test/e2e/expected/streaming/requests.json b/test/e2e/expected/streaming/requests.json new file mode 100644 index 0000000..cb29ae6 --- /dev/null +++ b/test/e2e/expected/streaming/requests.json @@ -0,0 +1,64 @@ +[ + { + "method": "POST", + "path": "/services/data/v67.0/ssot/data-custom-code", + "body": { + "label": "smoke_streaming", + "name": "smoke_streaming", + "description": "Smoke test streaming script", + "version": "1.0.0", + "computeType": "CPU_M", + "codeType": "script", + "invokeOptions": ["StreamingTransform"] + } + }, + { + "method": "PUT", + "path": "/upload/smoke_streaming", + "zipEntries": ["config.json", "entrypoint.py"] + }, + { + "method": "GET", + "path": "/services/data/v63.0/ssot/data-custom-code/smoke_streaming" + }, + { + "method": "POST", + "path": "/services/data/v63.0/ssot/data-transforms", + "body": { + "definition": { + "type": "DCSQL", + "manifest": { + "nodes": { + "node1": { + "relation_name": "Smoke_Stream_Target__dll", + "config": { + "materialized": "table" + }, + "compiled_code": "" + } + }, + "sources": { + "source1": { + "relation_name": "Smoke_Stream_Source__dll" + } + }, + "macros": { + "macro.byoc": { + "arguments": [ + { + "name": "smoke_streaming", + "type": "BYOC_SCRIPT" + } + ] + } + } + }, + "version": "56.0" + }, + "label": "smoke_streaming", + "name": "smoke_streaming", + "type": "STREAMING", + "dataSpaceName": "default" + } + } +] diff --git a/test/e2e/expected/streaming/zip-entries.json b/test/e2e/expected/streaming/zip-entries.json new file mode 100644 index 0000000..84224d4 --- /dev/null +++ b/test/e2e/expected/streaming/zip-entries.json @@ -0,0 +1 @@ +["config.json", "entrypoint.py"] diff --git a/test/e2e/fixtures/streaming/entrypoint.py b/test/e2e/fixtures/streaming/entrypoint.py new file mode 100644 index 0000000..8719241 --- /dev/null +++ b/test/e2e/fixtures/streaming/entrypoint.py @@ -0,0 +1,23 @@ +"""Fixed streaming entrypoint for the e2e smoke test. + +The SDK's streaming example can change between SDK releases, so the smoke +test swaps in this file before scanning to get stable, comparable output. +""" + +from datacustomcode.client import RunMode, StreamingClient, get_run_mode + + +def main(): + target_dlo = "Smoke_Stream_Target__dll" + client = StreamingClient() + + if get_run_mode() == RunMode.DELTA_SYNC: + df = client.read_dlo_deltas() + client.write_dlo_deltas(target_dlo, df).awaitTermination() + else: + df = client.read_dlo() + client.auto_write_to_dlo(target_dlo, df) + + +if __name__ == "__main__": + main() diff --git a/test/e2e/smoke.e2e.ts b/test/e2e/smoke.e2e.ts index 1a7a02c..1e7c8f3 100644 --- a/test/e2e/smoke.e2e.ts +++ b/test/e2e/smoke.e2e.ts @@ -41,6 +41,7 @@ import { homedir, tmpdir } from 'node:os'; import * as path from 'node:path'; import { fileURLToPath } from 'node:url'; import { expect } from 'chai'; +import { parse as parseYaml } from 'yaml'; import { listZipEntries, startMockDataCloud, type MockDataCloud, type RecordedRequest } from './mockDataCloud.js'; const HERE = path.dirname(fileURLToPath(import.meta.url)); @@ -310,6 +311,94 @@ describe('e2e smoke', function () { }); }); + describe('streaming script', () => { + let cwd: string; + let pkg: string; + const configPath = (): string => path.join(pkg, 'payload', 'config.json'); + + before(() => { + cwd = path.join(workDir, 'streaming'); + pkg = path.join(cwd, 'pkg'); + mkdirSync(cwd); + }); + + it('init scaffolds a StreamingTransform package', async () => { + const result = await runOk( + ['data-code-extension', 'script', 'init', '--package-dir', pkg, '--use-in-feature', 'StreamingTransform'], + cwd + ); + expect(result.stderr).to.include('streamingSource.name'); + + // The entrypoint and its write target come from the SDK's streaming example, so + // check the shape rather than exact content. + const config = readJson(configPath()) as { + streamingSource: unknown; + permissions: { read: unknown; write: { dlo: string[] } }; + }; + expect(config.streamingSource).to.deep.equal({ type: 'dlo', name: '' }); + expect(config.permissions.read).to.deep.equal({ dlo: [] }); + expect(config.permissions.write.dlo).to.be.an('array').that.is.not.empty; + + const runConfig = parseYaml(readFileSync(path.join(pkg, 'config.yaml'), 'utf8')) as Record< + string, + { ['type_config_name']: string } + >; + expect(runConfig.reader_config.type_config_name).to.equal('LocalDeltasReader'); + expect(runConfig.writer_config.type_config_name).to.equal('LocalDeltasWriter'); + }); + + it('scan keeps streamingSource.name and writes delta permissions', async () => { + copyFileSync(path.join(FIXTURES_DIR, 'streaming', 'entrypoint.py'), path.join(pkg, 'payload', 'entrypoint.py')); + const config = readJson(configPath()) as Record; + writeFileSync( + configPath(), + JSON.stringify({ ...config, streamingSource: { type: 'dlo', name: 'Smoke_Stream_Source__dll' } }, null, 2) + ); + + const result = await runOk( + [ + 'data-code-extension', + 'script', + 'scan', + '--entrypoint', + path.join(pkg, 'payload', 'entrypoint.py'), + '--no-prompt', + ], + cwd + ); + expect(result.stderr).to.not.include('streamingSource.name is empty'); + expectMatches('streaming/config.json', readJson(configPath())); + }); + + it('zip packages the payload', async () => { + await runOk(['data-code-extension', 'script', 'zip', '--package-dir', path.join(pkg, 'payload')], cwd); + expectMatches('streaming/zip-entries.json', await zipEntries(path.join(cwd, 'deployment.zip'))); + }); + + it('deploy creates a streaming deployment and a STREAMING transform', async () => { + const start = mock.requests.length; + await runOk( + [ + 'data-code-extension', + 'script', + 'deploy', + '--name', + 'smoke_streaming', + '--package-version', + '1.0.0', + '--description', + 'Smoke test streaming script', + '--package-dir', + path.join(pkg, 'payload'), + '--target-org', + USERNAME, + ], + cwd + ); + expectMatches('streaming/requests.json', requestsSince(start)); + }); + }); + describe('function', () => { let cwd: string; let pkg: string; diff --git a/test/utils/nativeDeploy.test.ts b/test/utils/nativeDeploy.test.ts index 140d42a..b78bcef 100644 --- a/test/utils/nativeDeploy.test.ts +++ b/test/utils/nativeDeploy.test.ts @@ -50,6 +50,11 @@ const SCRIPT_DLO_CONFIG = { permissions: { read: { dlo: ['Account__dll'] }, write: { dlo: ['Result__dll'] } }, }; +const SCRIPT_STREAMING_CONFIG = { + ...SCRIPT_DLO_CONFIG, + streamingSource: { type: 'dlo', name: 'Account__dll' }, +}; + /** Await a promise expected to reject and return the error for assertions. */ async function expectReject(promise: Promise): Promise { try { @@ -305,6 +310,17 @@ describe('nativeDeploy.buildDataTransformBody', () => { ]); }); + it('marks the transform STREAMING when the config declares a streamingSource', () => { + const config: DataTransformConfig = { + entryPoint: 'entrypoint.py', + sdkVersion: '1.0.0', + dataspace: 'my_space', + permissions: { read: { dlo: ['Account__dll'] }, write: { dlo: ['Result__dll'] } }, + streamingSource: { type: 'dlo', name: 'Account__dll' }, + }; + expect(buildDataTransformBody(metadata, config).type).to.equal('STREAMING'); + }); + it('throws when a DMO transform is missing dataObjects', () => { const config: DataTransformConfig = { entryPoint: 'entrypoint.py', @@ -325,6 +341,31 @@ describe('nativeDeploy.getConfig', () => { await fs.rm(dir, { recursive: true, force: true }); }); + it('parses the optional streamingSource', async () => { + const { dir, packageDir } = await makePackage({ packageType: 'script', config: SCRIPT_STREAMING_CONFIG }); + const config = (await getConfig(packageDir, 'script')) as DataTransformConfig; + expect(config.streamingSource).to.deep.equal({ type: 'dlo', name: 'Account__dll' }); + await fs.rm(dir, { recursive: true, force: true }); + }); + + it('leaves streamingSource unset for a batch config', async () => { + const { dir, packageDir } = await makePackage({ packageType: 'script', config: SCRIPT_DLO_CONFIG }); + const config = (await getConfig(packageDir, 'script')) as DataTransformConfig; + expect(config).to.not.have.property('streamingSource'); + await fs.rm(dir, { recursive: true, force: true }); + }); + + it('rejects a malformed streamingSource', async () => { + const { dir, packageDir } = await makePackage({ + packageType: 'script', + config: { ...SCRIPT_DLO_CONFIG, streamingSource: { type: 'dlo' } }, + }); + const err = await expectReject(getConfig(packageDir, 'script')); + expect(err.name).to.equal('InvalidConfig'); + expect(err.message).to.match(/streamingSource/); + await fs.rm(dir, { recursive: true, force: true }); + }); + it('parses the optional dataObjects schema for a DLO write config', async () => { const { dir, packageDir } = await makePackage({ packageType: 'script', @@ -561,6 +602,66 @@ describe('nativeDeploy.NativeDeployer.deploy', () => { await fs.rm(dir, { recursive: true, force: true }); }); + it('deploys a streaming script with StreamingTransform invokeOptions and a STREAMING transform', async () => { + const { dir, packageDir } = await makePackage({ packageType: 'script', config: SCRIPT_STREAMING_CONFIG }); + const { deps, createdMetadata, dataTransformBodies } = recordingDeps(); + + await NativeDeployer.deploy( + { name: 'my_stream', version: '1', description: 'd', packageDir, cpuSize: 'CPU_L', connection: dummyConnection }, + deps + ); + + expect(createdMetadata[0].invokeOptions).to.deep.equal(['StreamingTransform']); + expect((dataTransformBodies[0] as Record).type).to.equal('STREAMING'); + await fs.rm(dir, { recursive: true, force: true }); + }); + + it('does not set invokeOptions for a batch script', async () => { + const { dir, packageDir } = await makePackage({ packageType: 'script', config: SCRIPT_DLO_CONFIG }); + const { deps, createdMetadata, dataTransformBodies } = recordingDeps(); + + await NativeDeployer.deploy( + { name: 'my_batch', version: '1', description: 'd', packageDir, cpuSize: 'CPU_L', connection: dummyConnection }, + deps + ); + + expect(createdMetadata[0].invokeOptions).to.equal(undefined); + expect((dataTransformBodies[0] as Record).type).to.equal('BATCH'); + await fs.rm(dir, { recursive: true, force: true }); + }); + + for (const [label, config, expectedPath, expectedInvokeOptions] of [ + ['batch', SCRIPT_DLO_CONFIG, '/services/data/v63.0/ssot/data-custom-code', undefined], + ['streaming', SCRIPT_STREAMING_CONFIG, '/services/data/v67.0/ssot/data-custom-code', ['StreamingTransform']], + ] as const) { + it(`creates a ${label} deployment on ${expectedPath}`, async () => { + const { dir, packageDir } = await makePackage({ packageType: 'script', config }); + const requests: Array<{ url: string; body: Record }> = []; + // Record the real default createDeployment request, then stop the deploy with a 409. + const connection = { + request: (req: { url: string; body: string }) => { + requests.push({ url: req.url, body: JSON.parse(req.body) as Record }); + return Promise.reject(Object.assign(new Error('conflict'), { statusCode: 409 })); + }, + } as unknown as Connection; + + await expectReject( + NativeDeployer.deploy({ + name: 'my_script', + version: '1', + description: 'd', + packageDir, + cpuSize: 'CPU_L', + connection, + }) + ); + expect(requests).to.have.length(1); + expect(requests[0].url).to.equal(expectedPath); + expect(requests[0].body.invokeOptions).to.deep.equal(expectedInvokeOptions); + await fs.rm(dir, { recursive: true, force: true }); + }); + } + it('infers the feature for a function package and skips the data transform', async () => { const { dir, packageDir } = await makePackage({ packageType: 'function', diff --git a/test/utils/nativeInit.test.ts b/test/utils/nativeInit.test.ts index 8871111..af46876 100644 --- a/test/utils/nativeInit.test.ts +++ b/test/utils/nativeInit.test.ts @@ -31,8 +31,36 @@ type FakePackageOptions = { scriptConfig?: Record; functionEntrypoint?: string; functionConfig?: Record; + /** Include the SDK 7.0.0+ streaming example and run config.yaml (default true). */ + streaming?: boolean; }; +const STREAMING_EXAMPLE = [ + 'from datacustomcode.client import StreamingClient', + 'def main():', + ' target_dlo = "Account_std_copy__dll"', + ' client = StreamingClient()', + ' df = client.read_dlo_deltas()', + ' client.write_dlo_deltas(target_dlo, df).awaitTermination()', + '', +].join('\n'); + +const SDK_CONFIG_YAML = [ + 'reader_config:', + ' type_config_name: QueryAPIDataCloudReader', + ' options:', + ' credentials_profile: default', + '', + 'writer_config:', + ' type_config_name: PrintDataCloudWriter', + '', + 'spark_config:', + ' options:', + ' # Resolve columns case-sensitively.', + " spark.sql.caseSensitive: 'true'", + '', +].join('\n'); + async function buildFakePythonPackage(opts: FakePackageOptions = {}): Promise { const root = await fs.mkdtemp(path.join(os.tmpdir(), 'nativeinit-fake-pkg-')); const templates = path.join(root, 'datacustomcode', 'templates'); @@ -62,6 +90,12 @@ async function buildFakePythonPackage(opts: FakePackageOptions = {}): Promise { + try { + await fs.access(p); + } catch { + return; + } + expect.fail(`expected ${p} not to exist`); +} + +async function expectInitError(opts: Parameters[0]): Promise { + try { + await executeNativeInit(opts); + } catch (err) { + return err as Error; + } + return expect.fail('expected init to throw'); +} + describe('executeNativeInit', () => { let pythonPackageLocation: string; let workdir: string; @@ -133,6 +185,76 @@ describe('executeNativeInit', () => { expect(config.permissions.write.dlo).to.deep.equal(['Account_std_copy__dll']); }); + it('writes no streaming files for a BatchTransform script', async () => { + const packageDir = path.join(workdir, 'my-batch'); + const result = await executeNativeInit({ + codeType: 'script', + packageDir, + pythonPackageLocation, + pythonPackageVersion: '1.2.3', + useInFeature: 'BatchTransform', + }); + + const config = JSON.parse(await fs.readFile(result.configPath, 'utf8')) as Record; + expect(config).to.not.have.property('streamingSource'); + await expectMissing(path.join(packageDir, 'config.yaml')); + }); + + it('initializes a StreamingTransform script from the streaming example', async () => { + const packageDir = path.join(workdir, 'my-stream'); + const result = await executeNativeInit({ + codeType: 'script', + packageDir, + pythonPackageLocation, + pythonPackageVersion: '7.0.0', + useInFeature: 'StreamingTransform', + }); + + expect(await fs.readFile(path.join(packageDir, 'payload', 'entrypoint.py'), 'utf8')).to.equal(STREAMING_EXAMPLE); + + const config = JSON.parse(await fs.readFile(result.configPath, 'utf8')) as Record; + expect(config.streamingSource).to.deep.equal({ type: 'dlo', name: '' }); + expect(config.permissions).to.deep.equal({ read: { dlo: [] }, write: { dlo: ['Account_std_copy__dll'] } }); + + // Reader/writer switched; everything else, comments and quoting included, kept. + expect(await fs.readFile(path.join(packageDir, 'config.yaml'), 'utf8')).to.equal( + SDK_CONFIG_YAML.replace('QueryAPIDataCloudReader', 'LocalDeltasReader').replace( + 'PrintDataCloudWriter', + 'LocalDeltasWriter' + ) + ); + }); + + it('fails before copying anything when the SDK has no streaming template', async () => { + const oldSdk = await buildFakePythonPackage({ streaming: false }); + const packageDir = path.join(workdir, 'my-stream'); + try { + const err = await expectInitError({ + codeType: 'script', + packageDir, + pythonPackageLocation: oldSdk, + pythonPackageVersion: '6.0.0', + useInFeature: 'StreamingTransform', + }); + expect(err.name).to.equal('InitStreamingTemplateNotFound'); + expect(err.message).to.include('6.0.0'); + await expectMissing(packageDir); + } finally { + await fs.rm(oldSdk, { recursive: true, force: true }); + } + }); + + it('rejects an unknown script use-in-feature', async () => { + const err = await expectInitError({ + codeType: 'script', + packageDir: path.join(workdir, 'pkg'), + pythonPackageLocation, + pythonPackageVersion: '7.0.0', + useInFeature: 'SearchIndexChunking', + }); + expect(err.name).to.equal('InvalidUseInFeature'); + }); + it('always writes dataspace=default for a script regardless of template config', async () => { // Mirrors Python behavior: dc_config_json_from_file uses the template constant // (dataspace='default') and overwrites the template's payload/config.json. diff --git a/test/utils/nativeScan.test.ts b/test/utils/nativeScan.test.ts index 7b7bfea..5457643 100644 --- a/test/utils/nativeScan.test.ts +++ b/test/utils/nativeScan.test.ts @@ -23,8 +23,11 @@ import { executeNativeScan, findBaseDirectory, getPackageType, + fileIsStreaming, + isStreamingSourceNameMissing, scanFile, scanFileForImports, + scanFileStreaming, updateConfig, writeRequirementsFile, } from '../../src/utils/nativeScan.js'; @@ -48,6 +51,92 @@ async function makePackage(opts: ScriptPackageOptions): Promise { return dir; } +/** The shape of the SDK's `streaming_deltas` example: deltas in one branch, batch calls in the other. */ +const STREAMING_ENTRYPOINT = [ + 'from datacustomcode.client import RunMode, StreamingClient, get_run_mode', + 'def main():', + ' target_dlo = "Account_std_copy__dll"', + ' client = StreamingClient()', + ' if get_run_mode() == RunMode.DELTA_SYNC:', + ' dataframe = client.read_dlo_deltas()', + ' query = client.write_dlo_deltas(target_dlo, dataframe)', + ' query.awaitTermination()', + ' else:', + ' dataframe = client.read_dlo()', + ' client.auto_write_to_dlo(target_dlo, dataframe)', + '', +].join('\n'); + +async function writeEntrypoint(code: string): Promise<{ tmp: string; file: string }> { + const tmp = await fs.mkdtemp(path.join(os.tmpdir(), 'scanfile-')); + const file = path.join(tmp, 'entrypoint.py'); + await fs.writeFile(file, code); + return { tmp, file }; +} + +describe('nativeScan: scanFileStreaming', () => { + it('detects delta reads and resolves delta write targets', async () => { + const { tmp, file } = await writeEntrypoint(STREAMING_ENTRYPOINT); + expect(await fileIsStreaming(file)).to.equal(true); + const calls = await scanFileStreaming(file); + expect(calls.readDloDeltas).to.equal(true); + expect(calls.readDmoDeltas).to.equal(false); + expect([...calls.writeDloDeltas]).to.deep.equal(['Account_std_copy__dll']); + expect(calls.writeDmoDeltas.size).to.equal(0); + await fs.rm(tmp, { recursive: true, force: true }); + }); + + it('is not streaming for a batch-only entrypoint or delta calls in comments', async () => { + const { tmp, file } = await writeEntrypoint( + '# c.read_dlo_deltas()\nc.read_dlo("a__dll")\nc.write_to_dlo("b__dll", df)\ns = "c.write_dlo_deltas(x)"\n' + ); + expect(await fileIsStreaming(file)).to.equal(false); + await fs.rm(tmp, { recursive: true, force: true }); + }); + + const invalid: Array<[string, string, RegExp]> = [ + [ + 'reading DLO and DMO deltas', + 'c.read_dlo_deltas()\nc.read_dmo_deltas()\nc.write_dlo_deltas("o__dll", df)\n', + /Cannot read DLO and DMO deltas/, + ], + ['no delta read', 'c.write_dlo_deltas("o__dll", df)\n', /must read from at least one DLO or DMO delta stream/], + ['no delta write', 'c.read_dlo_deltas()\n', /must write to at least one DLO or DMO via write_dlo_deltas/], + [ + 'writing DLO and DMO deltas', + 'c.read_dlo_deltas()\nc.write_dlo_deltas("o__dll", df)\nc.write_dmo_deltas("p__dlm", df)\n', + /Cannot write DLO and DMO deltas/, + ], + [ + 'different read and write layers', + 'c.read_dlo_deltas()\nc.write_dmo_deltas("p__dlm", df)\n', + /must read and write the same layer \(got read=dlo, write=dmo\)/, + ], + ]; + for (const [label, code, message] of invalid) { + it(`rejects ${label}`, async () => { + const { tmp, file } = await writeEntrypoint(code); + try { + await scanFileStreaming(file); + expect.fail('expected scanFileStreaming to throw'); + } catch (err) { + expect((err as SfError).name).to.equal('InvalidEntrypoint'); + expect((err as SfError).message).to.match(message); + } + await fs.rm(tmp, { recursive: true, force: true }); + }); + } +}); + +describe('nativeScan: isStreamingSourceNameMissing', () => { + it('is false for batch configs and named sources, true for an empty or malformed source', () => { + expect(isStreamingSourceNameMissing({})).to.equal(false); + expect(isStreamingSourceNameMissing({ streamingSource: { type: 'dlo', name: 'A__dll' } })).to.equal(false); + expect(isStreamingSourceNameMissing({ streamingSource: { type: 'dlo', name: '' } })).to.equal(true); + expect(isStreamingSourceNameMissing({ streamingSource: 'A__dll' })).to.equal(true); + }); +}); + describe('nativeScan: scanFile', () => { it('extracts read/write DLO calls with literal and variable args', async () => { const tmp = await fs.mkdtemp(path.join(os.tmpdir(), 'scanfile-')); @@ -386,6 +475,43 @@ describe('nativeScan: updateConfig', () => { await fs.rm(dir, { recursive: true, force: true }); }); + it('writes a streaming config, keeping streamingSource.name as the only read', async () => { + const dir = await makePackage({ + entrypoint: STREAMING_ENTRYPOINT, + config: { + dataspace: 'default', + streamingSource: { type: 'dmo', name: 'Account__dll' }, + permissions: { read: { dlo: ['Stale__dll'] }, write: { dlo: ['Stale__dll'] } }, + }, + }); + const cfg = await updateConfig(path.join(dir, 'payload', 'entrypoint.py'), 'script'); + expect(cfg.streamingSource).to.deep.equal({ type: 'dlo', name: 'Account__dll' }); + expect(cfg.permissions).to.deep.equal({ + read: { dlo: ['Account__dll'] }, + write: { dlo: ['Account_std_copy__dll'] }, + }); + await fs.rm(dir, { recursive: true, force: true }); + }); + + it('writes an empty streaming read permission when streamingSource.name is not set', async () => { + const dir = await makePackage({ entrypoint: STREAMING_ENTRYPOINT, config: { dataspace: 'default' } }); + const cfg = await updateConfig(path.join(dir, 'payload', 'entrypoint.py'), 'script'); + expect(cfg.streamingSource).to.deep.equal({ type: 'dlo', name: '' }); + expect(cfg.permissions).to.deep.equal({ read: { dlo: [] }, write: { dlo: ['Account_std_copy__dll'] } }); + await fs.rm(dir, { recursive: true, force: true }); + }); + + it('drops streamingSource when the entrypoint is no longer streaming', async () => { + const dir = await makePackage({ + entrypoint: 'c.read_dlo("In__dll")\nc.write_to_dlo("Out__dll", df)\n', + config: { dataspace: 'default', streamingSource: { type: 'dlo', name: 'In__dll' } }, + }); + const cfg = await updateConfig(path.join(dir, 'payload', 'entrypoint.py'), 'script'); + expect(cfg).to.not.have.property('streamingSource'); + expect(cfg.permissions).to.deep.equal({ read: { dlo: ['In__dll'] }, write: { dlo: ['Out__dll'] } }); + await fs.rm(dir, { recursive: true, force: true }); + }); + it('preserves a non-default dataspace value', async () => { const dir = await makePackage({ entrypoint: diff --git a/yarn.lock b/yarn.lock index 1255956..79bcccf 100644 --- a/yarn.lock +++ b/yarn.lock @@ -6691,6 +6691,11 @@ yaml@^2.9.0: resolved "https://registry.yarnpkg.com/yaml/-/yaml-2.9.1.tgz#c16233fb31944e705cfefaff38795587f57588ce" integrity sha512-3NxN8+78OdzbT7C/WjGsyfPAtJaN3FNDsWxv7Y7mcDsT/oOmgW8BpyQQFFBnvZE3j9Y2Sdz1ULFLezL7Eb2yFw== +yaml@^2.9.1: + version "2.9.1" + resolved "https://registry.yarnpkg.com/yaml/-/yaml-2.9.1.tgz#c16233fb31944e705cfefaff38795587f57588ce" + integrity sha512-3NxN8+78OdzbT7C/WjGsyfPAtJaN3FNDsWxv7Y7mcDsT/oOmgW8BpyQQFFBnvZE3j9Y2Sdz1ULFLezL7Eb2yFw== + yargs-parser@^18.1.2: version "18.1.3" resolved "https://registry.yarnpkg.com/yargs-parser/-/yargs-parser-18.1.3.tgz#be68c4975c6b2abf469236b0c870362fab09a7b0" From c0c56ee387e6a37fb21f12059d630ef57318f25d Mon Sep 17 00:00:00 2001 From: Jesus Orosco Date: Tue, 10 Nov 2026 16:59:19 -0800 Subject: [PATCH 2/7] fix: bugfix when testing locally --- src/base/initBase.ts | 8 ++++- src/utils/nativeInit.ts | 11 +++++++ src/utils/pythonChecker.ts | 30 +++++++++++++++++++ .../commands/data-code-extension/init.test.ts | 28 +++++++++++++++-- test/utils/nativeInit.test.ts | 19 ++++++++++++ test/utils/pythonChecker.test.ts | 17 +++++++++++ 6 files changed, 110 insertions(+), 3 deletions(-) diff --git a/src/base/initBase.ts b/src/base/initBase.ts index 93543fb..44d4590 100644 --- a/src/base/initBase.ts +++ b/src/base/initBase.ts @@ -13,10 +13,12 @@ * See the License for the specific language governing permissions and * limitations under the License. */ +import * as path from 'node:path'; import { SfCommand } from '@salesforce/sf-plugins-core'; import { Messages } from '@salesforce/core'; import { executeNativeInit, type NativeInitResult } from '../utils/nativeInit.js'; import { checkEnvironment } from '../utils/environmentChecker.js'; +import { PythonChecker } from '../utils/pythonChecker.js'; import { SCRIPT_USE_IN_FEATURE_STREAMING } from '../utils/constants.js'; import { type SharedResultProps } from './types.js'; @@ -42,11 +44,15 @@ export abstract class InitBase ext try { const { pythonInfo, packageInfo } = await checkEnvironment(this.spinner, this.log.bind(this), messages); + // `pip show` reports site-packages even for editable installs, whose code lives elsewhere, + // so ask Python where the SDK really is and fall back to pip's location. + const sdkDir = await PythonChecker.findPackageDir(pythonInfo.command, 'datacustomcode'); + this.spinner.start(messages.getMessage('info.executingInit')); const executionResult = await executeNativeInit({ codeType, packageDir, - pythonPackageLocation: packageInfo.location, + pythonPackageLocation: sdkDir ? path.dirname(sdkDir) : packageInfo.location, pythonPackageVersion: packageInfo.version, useInFeature: additionalFlags.useInFeature as string | undefined, }); diff --git a/src/utils/nativeInit.ts b/src/utils/nativeInit.ts index f58aa8d..a9718ed 100644 --- a/src/utils/nativeInit.ts +++ b/src/utils/nativeInit.ts @@ -33,6 +33,7 @@ const messages = Messages.loadMessages('@salesforce/plugin-data-code-extension', export type NativeInitOptions = { codeType: CodeType; packageDir: string; + /** Directory containing the `datacustomcode` package (site-packages, or `src/` for editable installs). */ pythonPackageLocation: string; pythonPackageVersion: string; useInFeature?: string; @@ -54,6 +55,8 @@ const SDK_CONFIG_FILE = 'sdk_config.json'; // Mirror of `datacustomcode/template.py`: the streaming example entrypoint, and the SDK's // run config, which streaming packages get a copy of with the local delta reader/writer. const STREAMING_EXAMPLE_ENTRYPOINT = path.join('examples', 'streaming_deltas', ENTRYPOINT_FILE); +// First SDK major version that ships the streaming template. +const STREAMING_MIN_SDK_MAJOR = 7; const SDK_RUN_CONFIG_FILE = 'config.yaml'; const FEATURE_TEMPLATE_MAPPING: Readonly> = { @@ -136,6 +139,14 @@ function resolveScriptFeature(useInFeature: string | undefined): string { async function ensureStreamingTemplateExists(scriptTemplateDir: string, sdkVersion: string): Promise { const entrypoint = path.join(scriptTemplateDir, STREAMING_EXAMPLE_ENTRYPOINT); if (!(await pathExists(entrypoint))) { + // 7.0.0+ ships the template, so a missing file there is an install problem, not an old SDK. + if (Number.parseInt(sdkVersion, 10) >= STREAMING_MIN_SDK_MAJOR) { + throw new SfError( + messages.getMessage('error.initTemplateNotFound', [entrypoint]), + 'InitTemplateNotFound', + messages.getMessages('actions.initTemplateNotFound') + ); + } throw new SfError( messages.getMessage('error.initStreamingTemplateNotFound', [sdkVersion, entrypoint]), 'InitStreamingTemplateNotFound', diff --git a/src/utils/pythonChecker.ts b/src/utils/pythonChecker.ts index 62bf1de..436f7b6 100644 --- a/src/utils/pythonChecker.ts +++ b/src/utils/pythonChecker.ts @@ -28,10 +28,40 @@ export type PythonVersionInfo = { patch: number; }; +/** + * Prints the directory of the package named by argv[1], or nothing if it can't be found. + * `find_spec` locates the package without importing it, and it follows editable installs + * (`pip install -e`, `poetry install`) whose code lives outside site-packages. + */ +const FIND_PACKAGE_DIR_SHIM = [ + 'import sys, importlib.util', + 'spec = importlib.util.find_spec(sys.argv[1])', + 'locations = list(spec.submodule_search_locations or []) if spec else []', + "print(locations[0] if locations else '')", +].join('\n'); + export class PythonChecker { private static readonly REQUIRED_MAJOR = 3; private static readonly REQUIRED_MINOR = 11; + /** + * Finds where `packageName` is installed for the given Python, as Python itself resolves it. + * + * @param command The Python command to ask (from checkPython311) + * @param packageName The importable package name, e.g. 'datacustomcode' + * @returns The package directory, or null if Python can't locate it + */ + public static async findPackageDir(command: string, packageName: string): Promise { + try { + // -P keeps the current directory off sys.path, so a local folder can't shadow the install. + const { stdout } = await spawnAsync(command, ['-P', '-c', FIND_PACKAGE_DIR_SHIM, packageName]); + const dir = stdout.trim(); + return dir.length > 0 ? dir : null; + } catch { + return null; + } + } + /** * Checks if Python 3.11 is installed on the system. * diff --git a/test/commands/data-code-extension/init.test.ts b/test/commands/data-code-extension/init.test.ts index 4930d29..906739a 100644 --- a/test/commands/data-code-extension/init.test.ts +++ b/test/commands/data-code-extension/init.test.ts @@ -13,11 +13,12 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs'; +import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; import * as path from 'node:path'; import { TestContext } from '@salesforce/core/testSetup'; import { expect } from 'chai'; +import type * as sinon from 'sinon'; import { stubSfCommandUx } from '@salesforce/sf-plugins-core'; import ScriptInit from '../../../src/commands/data-code-extension/script/init.js'; import FunctionInit from '../../../src/commands/data-code-extension/function/init.js'; @@ -283,6 +284,8 @@ describe('data-code-extension init commands', () => { describe('with a stubbed environment', () => { let root: string; + let checkPackageStub: sinon.SinonStub; + let findPackageDirStub: sinon.SinonStub; beforeEach(() => { root = mkdtempSync(path.join(tmpdir(), 'init-command-')); @@ -312,18 +315,39 @@ describe('data-code-extension init commands', () => { minor: 11, patch: 5, }); - $$.SANDBOX.stub(PipChecker, 'checkPackage').resolves({ + checkPackageStub = $$.SANDBOX.stub(PipChecker, 'checkPackage').resolves({ name: 'salesforce-data-customcode', version: '7.0.0', location: path.join(root, 'site-packages'), pipCommand: 'pip3', }); + // Default: Python can't locate the SDK, so init falls back to pip's location. + findPackageDirStub = $$.SANDBOX.stub(PythonChecker, 'findPackageDir').resolves(null); }); afterEach(() => { rmSync(root, { recursive: true, force: true }); }); + it('uses the SDK location reported by Python for an editable install', async () => { + // Editable installs: pip reports a site-packages dir that doesn't hold the package. + const emptySitePackages = path.join(root, 'venv-site-packages'); + mkdirSync(emptySitePackages); + checkPackageStub.resolves({ + name: 'salesforce-data-customcode', + version: '7.0.1.dev8', + location: emptySitePackages, + pipCommand: 'pip3', + }); + findPackageDirStub.resolves(path.join(root, 'site-packages', 'datacustomcode')); + + const packageDir = path.join(root, 'editable'); + await ScriptInit.run(['--package-dir', packageDir, '--use-in-feature', 'StreamingTransform']); + + expect(findPackageDirStub.calledOnceWith('python3', 'datacustomcode')).to.be.true; + expect(readFileSync(path.join(packageDir, 'payload', 'entrypoint.py'), 'utf8')).to.include('read_dlo_deltas'); + }); + it('warns to set streamingSource.name after a StreamingTransform init', async () => { const packageDir = path.join(root, 'stream'); await ScriptInit.run(['--package-dir', packageDir, '--use-in-feature', 'StreamingTransform']); diff --git a/test/utils/nativeInit.test.ts b/test/utils/nativeInit.test.ts index af46876..f3bf3e9 100644 --- a/test/utils/nativeInit.test.ts +++ b/test/utils/nativeInit.test.ts @@ -244,6 +244,25 @@ describe('executeNativeInit', () => { } }); + it('reports a missing template, not an old SDK, when a 7.x SDK lacks the streaming template', async () => { + const brokenSdk = await buildFakePythonPackage({ streaming: false }); + const packageDir = path.join(workdir, 'my-stream-7'); + try { + const err = await expectInitError({ + codeType: 'script', + packageDir, + pythonPackageLocation: brokenSdk, + pythonPackageVersion: '7.0.1.dev8', + useInFeature: 'StreamingTransform', + }); + expect(err.name).to.equal('InitTemplateNotFound'); + expect(err.message).to.include(path.join('examples', 'streaming_deltas', 'entrypoint.py')); + await expectMissing(packageDir); + } finally { + await fs.rm(brokenSdk, { recursive: true, force: true }); + } + }); + it('rejects an unknown script use-in-feature', async () => { const err = await expectInitError({ codeType: 'script', diff --git a/test/utils/pythonChecker.test.ts b/test/utils/pythonChecker.test.ts index 70f77ad..1458e69 100644 --- a/test/utils/pythonChecker.test.ts +++ b/test/utils/pythonChecker.test.ts @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ +import * as path from 'node:path'; import { TestContext } from '@salesforce/core/testSetup'; import { SfError } from '@salesforce/core'; import { expect } from 'chai'; @@ -25,6 +26,22 @@ describe('PythonChecker', () => { $$.restore(); }); + describe('findPackageDir', () => { + it('returns the directory of an installed package', async () => { + const dir = await PythonChecker.findPackageDir('python3', 'json'); + // python3 may be missing on the test machine; when present, json is a stdlib package. + if (dir !== null) expect(path.basename(dir)).to.equal('json'); + }); + + it('returns null for a package that is not installed', async () => { + expect(await PythonChecker.findPackageDir('python3', 'no_such_package_xyz')).to.be.null; + }); + + it('returns null when the Python command does not exist', async () => { + expect(await PythonChecker.findPackageDir('no-such-python-xyz', 'json')).to.be.null; + }); + }); + describe('checkPython311', () => { // Since we're using ES modules and can't easily stub child_process.exec, // we'll test the actual Python installation on the system From ff208c2be3a68133db53ac09545c2187bc8815c9 Mon Sep 17 00:00:00 2001 From: Jesus Orosco Date: Wed, 11 Nov 2026 09:04:50 -0800 Subject: [PATCH 3/7] fix: reject an empty streamingSource.name before deploying --- messages/deploy.md | 9 ++++ src/utils/nativeDeploy.ts | 9 ++++ test/e2e/smoke.e2e.ts | 88 +++++++++++++++++++++++++++------ test/utils/nativeDeploy.test.ts | 39 +++++++++++++++ 4 files changed, 129 insertions(+), 16 deletions(-) diff --git a/messages/deploy.md b/messages/deploy.md index d03f416..5bd7a4f 100644 --- a/messages/deploy.md +++ b/messages/deploy.md @@ -221,6 +221,15 @@ config.json at %s is missing required fields: %s. config.json at %s is invalid: %s. +# error.streamingSourceNameEmpty + +streamingSource.name is empty in config.json at %s. A streaming transform needs the DLO or DMO it reads from. + +# actions.streamingSourceNameEmpty + +- Set streamingSource.name in config.json to the DLO or DMO the streaming transform reads from. +- Run 'sf data-code-extension script scan' to update permissions, then deploy again. + # error.dmoRequiresDataObjects DMO transforms require 'dataObjects' in config.json describing the schema of each output DMO. diff --git a/src/utils/nativeDeploy.ts b/src/utils/nativeDeploy.ts index 1161a5a..245e7d9 100644 --- a/src/utils/nativeDeploy.ts +++ b/src/utils/nativeDeploy.ts @@ -482,6 +482,15 @@ function parseStreamingSource(value: unknown, configPath: string): { streamingSo 'InvalidConfig' ); } + // Stricter than the SDK: init leaves the name empty, and deploying that creates a + // STREAMING transform with no input source. + if (value.name.trim().length === 0) { + throw new SfError( + messages.getMessage('error.streamingSourceNameEmpty', [configPath]), + 'StreamingSourceNameEmpty', + messages.getMessages('actions.streamingSourceNameEmpty') + ); + } return { streamingSource: { type: value.type, name: value.name } }; } diff --git a/test/e2e/smoke.e2e.ts b/test/e2e/smoke.e2e.ts index 1e7c8f3..f25c677 100644 --- a/test/e2e/smoke.e2e.ts +++ b/test/e2e/smoke.e2e.ts @@ -59,9 +59,9 @@ let mock: MockDataCloud; let childEnv: NodeJS.ProcessEnv; /** Run the plugin asynchronously (the mock server lives in this process). */ -function runPlugin(args: string[], cwd: string): Promise { +function runPlugin(args: string[], cwd: string, env: NodeJS.ProcessEnv = {}): Promise { return new Promise((resolve, reject) => { - const child = spawn(process.execPath, [BIN, ...args], { cwd, env: childEnv }); + const child = spawn(process.execPath, [BIN, ...args], { cwd, env: { ...childEnv, ...env } }); let stdout = ''; let stderr = ''; child.stdout.on('data', (d: Buffer) => (stdout += d.toString())); @@ -71,8 +71,8 @@ function runPlugin(args: string[], cwd: string): Promise { }); } -async function runOk(args: string[], cwd: string): Promise { - const result = await runPlugin(args, cwd); +async function runOk(args: string[], cwd: string, env?: NodeJS.ProcessEnv): Promise { + const result = await runPlugin(args, cwd, env); if (result.code !== 0) { throw new Error( `sf ${args.join(' ')} exited ${String(result.code)}\n--- stdout ---\n${result.stdout}\n--- stderr ---\n${ @@ -315,6 +315,21 @@ describe('e2e smoke', function () { let cwd: string; let pkg: string; const configPath = (): string => path.join(pkg, 'payload', 'config.json'); + const deployArgs = (): string[] => [ + 'data-code-extension', + 'script', + 'deploy', + '--name', + 'smoke_streaming', + '--package-version', + '1.0.0', + '--description', + 'Smoke test streaming script', + '--package-dir', + path.join(pkg, 'payload'), + '--target-org', + USERNAME, + ]; before(() => { cwd = path.join(workDir, 'streaming'); @@ -347,6 +362,14 @@ describe('e2e smoke', function () { expect(runConfig.writer_config.type_config_name).to.equal('LocalDeltasWriter'); }); + it('deploy rejects the unedited config before any API call', async () => { + const start = mock.requests.length; + const result = await runPlugin(deployArgs(), cwd); + expect(result.code, result.stderr).to.not.equal(0); + expect(result.stderr).to.include('StreamingSourceNameEmpty'); + expect(requestsSince(start)).to.deep.equal([]); + }); + it('scan keeps streamingSource.name and writes delta permissions', async () => { copyFileSync(path.join(FIXTURES_DIR, 'streaming', 'entrypoint.py'), path.join(pkg, 'payload', 'entrypoint.py')); const config = readJson(configPath()) as Record; @@ -377,25 +400,58 @@ describe('e2e smoke', function () { it('deploy creates a streaming deployment and a STREAMING transform', async () => { const start = mock.requests.length; - await runOk( + await runOk(deployArgs(), cwd); + expectMatches('streaming/requests.json', requestsSince(start)); + }); + + // Runs last: the delta fixtures live under payload/, so they'd otherwise end up in the zip. + it('run drains local delta fixtures through the generated config.yaml', async function () { + // Local Spark streaming checkpoints need Hadoop's winutils.exe on Windows, which CI runners don't have. + if (process.platform === 'win32') this.skip(); + // LocalDeltasReader reads payload/streaming_fixtures/, relative to the + // working directory. Seeding _schema.json up front keeps it from querying the org. + const fixtures = path.join(pkg, 'payload', 'streaming_fixtures', 'Smoke_Stream_Source__dll'); + mkdirSync(fixtures, { recursive: true }); + writeFileSync( + path.join(fixtures, '_schema.json'), + JSON.stringify({ + type: 'struct', + fields: [ + { name: 'Id__c', type: 'string', nullable: true, metadata: {} }, + { name: 'Name__c', type: 'string', nullable: true, metadata: {} }, + ], + }) + ); + const change = (id: string, name: string, version: number, op: string): string => + JSON.stringify({ + ['Id__c']: id, + ['Name__c']: name, + ['_commit_version']: version, + ['_commit_timestamp']: '2026-01-01T00:00:00Z', + ['_merge_record_type']: op, + }); + writeFileSync( + path.join(fixtures, '001.json'), + [change('a1', 'smoke-upsert', 1, 'UPSERT'), change('a2', 'smoke-delete', 2, 'DELETE')].join('\n') + ); + + const result = await runOk( [ 'data-code-extension', 'script', - 'deploy', - '--name', - 'smoke_streaming', - '--package-version', - '1.0.0', - '--description', - 'Smoke test streaming script', - '--package-dir', - path.join(pkg, 'payload'), + 'run', + '--entrypoint', + path.join(pkg, 'payload', 'entrypoint.py'), '--target-org', USERNAME, ], - cwd + pkg, + { ['BYOC_RUN_MODE']: 'DELTA_SYNC' } ); - expectMatches('streaming/requests.json', requestsSince(start)); + // LocalDeltasWriter prints one preview per micro-batch, tagging each row with its operation. + expect(result.stdout).to.match(/Target=Smoke_Stream_Target__dll batch_id=0 file=\S+ rows=2/); + expect(result.stdout).to.include('smoke-upsert'); + expect(result.stdout).to.match(/smoke-delete\s*\|\s*DELETE/); }); }); diff --git a/test/utils/nativeDeploy.test.ts b/test/utils/nativeDeploy.test.ts index b78bcef..d6a1a4e 100644 --- a/test/utils/nativeDeploy.test.ts +++ b/test/utils/nativeDeploy.test.ts @@ -366,6 +366,20 @@ describe('nativeDeploy.getConfig', () => { await fs.rm(dir, { recursive: true, force: true }); }); + for (const name of ['', ' ']) { + it(`rejects a streamingSource with a blank name (${JSON.stringify(name)})`, async () => { + const { dir, packageDir } = await makePackage({ + packageType: 'script', + config: { ...SCRIPT_STREAMING_CONFIG, streamingSource: { type: 'dlo', name } }, + }); + const err = await expectReject(getConfig(packageDir, 'script')); + expect(err.name).to.equal('StreamingSourceNameEmpty'); + expect(err.message).to.include('streamingSource.name'); + expect(err.actions?.join('\n')).to.include('scan'); + await fs.rm(dir, { recursive: true, force: true }); + }); + } + it('parses the optional dataObjects schema for a DLO write config', async () => { const { dir, packageDir } = await makePackage({ packageType: 'script', @@ -602,6 +616,31 @@ describe('nativeDeploy.NativeDeployer.deploy', () => { await fs.rm(dir, { recursive: true, force: true }); }); + it('rejects a streaming script with no streamingSource.name before any API call', async () => { + // The config exactly as `script init -u StreamingTransform` writes it. + const { dir, packageDir } = await makePackage({ + packageType: 'script', + config: { ...SCRIPT_STREAMING_CONFIG, streamingSource: { type: 'dlo', name: '' } }, + }); + const { deps, order } = recordingDeps(); + const err = await expectReject( + NativeDeployer.deploy( + { + name: 'my-stream', + version: '1.0.0', + description: 'd', + packageDir, + cpuSize: 'CPU_2XL', + connection: dummyConnection, + }, + deps + ) + ); + expect(err.name).to.equal('StreamingSourceNameEmpty'); + expect(order).to.deep.equal([]); + await fs.rm(dir, { recursive: true, force: true }); + }); + it('deploys a streaming script with StreamingTransform invokeOptions and a STREAMING transform', async () => { const { dir, packageDir } = await makePackage({ packageType: 'script', config: SCRIPT_STREAMING_CONFIG }); const { deps, createdMetadata, dataTransformBodies } = recordingDeps(); From bb741ea7dbfa9aae0b198cc6fc078e078d3ee5cf Mon Sep 17 00:00:00 2001 From: Jesus Orosco Date: Wed, 11 Nov 2026 11:00:18 -0800 Subject: [PATCH 4/7] feat: clearer deploy progress output --- messages/deploy.md | 30 ++++++- src/utils/nativeDeploy.ts | 72 ++++++++++++++-- test/utils/nativeDeploy.test.ts | 147 ++++++++++++++++++++++++++++++++ 3 files changed, 240 insertions(+), 9 deletions(-) diff --git a/messages/deploy.md b/messages/deploy.md index 5bd7a4f..b0df4bb 100644 --- a/messages/deploy.md +++ b/messages/deploy.md @@ -156,9 +156,25 @@ The --%s flag value exceeds the maximum length of %s characters (%s provided). API name '%s' was sanitized to '%s'. -# info.inferredFeature +# info.deploymentType -Inferred feature: %s +Deployment type: %s (%s) + +# info.deploymentTypeReason.batch + +config.json has no streamingSource + +# info.deploymentTypeReason.streaming + +config.json streamingSource: %s %s + +# info.deploymentTypeReason.function + +inferred from the entrypoint signature + +# info.cpuSize + +CPU size: %s # info.creatingDeployment @@ -174,7 +190,15 @@ Uploading deployment package... # info.waitingForDeployment -Waiting for deployment to complete... +Waiting for deployment to complete. Package extraction can take several minutes. + +# info.waitingForDeploymentScriptNote + +Stopping this command doesn't cancel the deployment, but the data transform is only created after the deployment finishes. If you stop early, the data transform is not created. + +# info.deploymentStillWaiting + +Still waiting (status: %s, %s elapsed)... # info.deploymentStatusPolled diff --git a/src/utils/nativeDeploy.ts b/src/utils/nativeDeploy.ts index 245e7d9..f2c5e5c 100644 --- a/src/utils/nativeDeploy.ts +++ b/src/utils/nativeDeploy.ts @@ -19,7 +19,7 @@ import path from 'node:path'; import { debuglog } from 'node:util'; import { Messages, SfError, type Connection } from '@salesforce/core'; import { decodeHTML } from 'entities'; -import { SCRIPT_USE_IN_FEATURE_STREAMING } from './constants.js'; +import { SCRIPT_USE_IN_FEATURE_BATCH, SCRIPT_USE_IN_FEATURE_STREAMING } from './constants.js'; import { findBaseDirectory, getPackageType, type CodeType } from './nativeScan.js'; import { zipWithSfError, ZIP_FILE_NAME, type ZipResult } from './zipBuilder.js'; @@ -729,6 +729,38 @@ export async function waitForDeployment( } } +/** How often a "still waiting" line is printed while the status doesn't change. */ +export const STILL_WAITING_INTERVAL_MS = 30 * 1000; + +function formatElapsed(ms: number): string { + const totalSeconds = Math.floor(ms / 1000); + const minutes = Math.floor(totalSeconds / 60); + const seconds = totalSeconds % 60; + return minutes > 0 ? `${minutes}m ${seconds}s` : `${seconds}s`; +} + +/** + * Build the `onStatus` callback for `waitForDeployment`. Polling happens every + * second, so print the status only when it changes, plus a periodic + * "still waiting" line with the elapsed time. + */ +export function createStatusReporter(log: (message: string) => void, now: () => number): (status: string) => void { + const start = now(); + let lastStatus: string | undefined; + let lastLogAt = start; + return (status: string): void => { + const current = now(); + if (status !== lastStatus) { + lastStatus = status; + lastLogAt = current; + log(messages.getMessage('info.deploymentStatusPolled', [status])); + } else if (current - lastLogAt >= STILL_WAITING_INTERVAL_MS) { + lastLogAt = current; + log(messages.getMessage('info.deploymentStillWaiting', [status, formatElapsed(current - start)])); + } + }; +} + /** * Native TypeScript port of `deploy_full` + the `deploy` command in * `datacustomcode/cli.py`. Preserves the original ordering: validate config @@ -753,6 +785,8 @@ export class NativeDeployer { ); } + log(messages.getMessage('info.cpuSize', [opts.cpuSize])); + const baseDir = findBaseDirectory(opts.packageDir); const packageType = await getPackageType(baseDir); @@ -767,13 +801,14 @@ export class NativeDeployer { log ); + let functionFeature: string | undefined; if (packageType === 'function') { const entrypointPath = path.join(opts.packageDir, ENTRYPOINT_FILE); const feature = await inferUseInFeature(entrypointPath); if (!feature) { throw new SfError(messages.getMessage('error.functionSignatureMismatch'), 'FunctionSignatureMismatch'); } - log(messages.getMessage('info.inferredFeature', [feature])); + functionFeature = feature; metadata.functionInvokeOptions = [USE_IN_FEATURE_MAPPING_FOR_CONNECT_API[feature] ?? feature]; } @@ -781,10 +816,34 @@ export class NativeDeployer { const config = await getConfig(opts.packageDir, packageType); // Mirror of `deploy_full`: a config with `streamingSource` deploys as a streaming transform. - if (packageType === 'script' && (config as DataTransformConfig).streamingSource && !metadata.invokeOptions) { + const streamingSource = packageType === 'script' ? (config as DataTransformConfig).streamingSource : undefined; + if (streamingSource && !metadata.invokeOptions) { metadata.invokeOptions = [SCRIPT_USE_IN_FEATURE_STREAMING]; } + if (functionFeature) { + log( + messages.getMessage('info.deploymentType', [ + functionFeature, + messages.getMessage('info.deploymentTypeReason.function'), + ]) + ); + } else if (streamingSource) { + log( + messages.getMessage('info.deploymentType', [ + SCRIPT_USE_IN_FEATURE_STREAMING, + messages.getMessage('info.deploymentTypeReason.streaming', [streamingSource.type, streamingSource.name]), + ]) + ); + } else { + log( + messages.getMessage('info.deploymentType', [ + SCRIPT_USE_IN_FEATURE_BATCH, + messages.getMessage('info.deploymentTypeReason.batch'), + ]) + ); + } + const deps = buildDefaultDeps(opts.connection, overrides); log(messages.getMessage('info.creatingDeployment', [metadata.name])); @@ -797,9 +856,10 @@ export class NativeDeployer { await deps.uploadZip(deployment.fileUploadUrl, path.resolve(ZIP_FILE_NAME)); log(messages.getMessage('info.waitingForDeployment')); - const status = await waitForDeployment(metadata.name, deps, (s) => - log(messages.getMessage('info.deploymentStatusPolled', [s])) - ); + if (packageType === 'script') { + log(messages.getMessage('info.waitingForDeploymentScriptNote')); + } + const status = await waitForDeployment(metadata.name, deps, createStatusReporter(log, deps.now)); if (packageType === 'script') { log(messages.getMessage('info.creatingDataTransform')); diff --git a/test/utils/nativeDeploy.test.ts b/test/utils/nativeDeploy.test.ts index d6a1a4e..46d3410 100644 --- a/test/utils/nativeDeploy.test.ts +++ b/test/utils/nativeDeploy.test.ts @@ -26,6 +26,8 @@ import { buildDataTransformBody, getConfig, waitForDeployment, + createStatusReporter, + STILL_WAITING_INTERVAL_MS, htmlUnescape, COMPUTE_TYPES, NativeDeployer, @@ -540,6 +542,63 @@ describe('nativeDeploy.waitForDeployment', () => { }); }); +describe('nativeDeploy.createStatusReporter', () => { + it('logs a status only when it changes', () => { + const logs: string[] = []; + const report = createStatusReporter( + (m) => logs.push(m), + () => 0 + ); + for (const s of ['Initializing', 'Initializing', 'Initializing', 'Extracting', 'Extracting', 'Deployed']) report(s); + expect(logs).to.deep.equal([ + 'Deployment status: Initializing', + 'Deployment status: Extracting', + 'Deployment status: Deployed', + ]); + }); + + it('logs a periodic "still waiting" line with the elapsed time while the status is unchanged', () => { + const logs: string[] = []; + let t = 0; + const report = createStatusReporter( + (m) => logs.push(m), + () => t + ); + report('Initializing'); // t=0 + for (t = 1000; t < STILL_WAITING_INTERVAL_MS; t += 1000) report('Initializing'); + t = STILL_WAITING_INTERVAL_MS; + report('Initializing'); + t = STILL_WAITING_INTERVAL_MS * 2 + 15_000; + report('Initializing'); + expect(logs).to.deep.equal([ + 'Deployment status: Initializing', + 'Still waiting (status: Initializing, 30s elapsed)...', + 'Still waiting (status: Initializing, 1m 15s elapsed)...', + ]); + }); + + it('restarts the "still waiting" interval after a status change', () => { + const logs: string[] = []; + let t = 0; + const report = createStatusReporter( + (m) => logs.push(m), + () => t + ); + report('Initializing'); + t = 20_000; + report('Extracting'); + t = 40_000; // only 20s since the last line + report('Extracting'); + t = 50_000; + report('Extracting'); + expect(logs).to.deep.equal([ + 'Deployment status: Initializing', + 'Deployment status: Extracting', + 'Still waiting (status: Extracting, 50s elapsed)...', + ]); + }); +}); + describe('nativeDeploy.NativeDeployer.deploy', () => { const dummyConnection = {} as unknown as Connection; @@ -701,6 +760,94 @@ describe('nativeDeploy.NativeDeployer.deploy', () => { }); } + describe('deploy output', () => { + async function deployLogs( + pkg: PackageOptions, + cpuSize = 'CPU_L', + overrides: Partial = {} + ): Promise { + const { dir, packageDir } = await makePackage(pkg); + const { deps } = recordingDeps(overrides); + const logs: string[] = []; + try { + await NativeDeployer.deploy( + { + name: 'pkg', + version: '1', + description: 'd', + packageDir, + cpuSize, + connection: dummyConnection, + log: (m) => logs.push(m), + }, + deps + ); + } finally { + await fs.rm(dir, { recursive: true, force: true }); + } + return logs; + } + + it('logs a batch transform deployment type', async () => { + const logs = await deployLogs({ packageType: 'script', config: SCRIPT_DLO_CONFIG }); + expect(logs).to.include('Deployment type: BatchTransform (config.json has no streamingSource)'); + }); + + it('logs a streaming transform deployment type with its source', async () => { + const logs = await deployLogs({ packageType: 'script', config: SCRIPT_STREAMING_CONFIG }); + expect(logs).to.include('Deployment type: StreamingTransform (config.json streamingSource: dlo Account__dll)'); + }); + + it('logs a function deployment type with the inferred feature', async () => { + const logs = await deployLogs({ + packageType: 'function', + config: { entryPoint: 'entrypoint.py' }, + entrypoint: CHUNKING_ENTRYPOINT, + }); + expect(logs).to.include('Deployment type: SearchIndexChunking (inferred from the entrypoint signature)'); + }); + + it('logs the deployment type before creating the deployment', async () => { + const logs = await deployLogs({ packageType: 'script', config: SCRIPT_DLO_CONFIG }); + const typeIdx = logs.findIndex((m) => m.startsWith('Deployment type:')); + const createIdx = logs.findIndex((m) => m.startsWith('Creating deployment')); + expect(typeIdx).to.be.greaterThan(-1); + expect(typeIdx).to.be.lessThan(createIdx); + }); + + it('logs the selected CPU size without the compute type it maps to', async () => { + const logs = await deployLogs({ packageType: 'script', config: SCRIPT_DLO_CONFIG }, 'CPU_L'); + expect(logs).to.include('CPU size: CPU_L'); + expect(logs.join('\n')).to.not.include('CPU_XS'); + }); + + it('warns that stopping early skips the data transform for scripts only', async () => { + const scriptLogs = await deployLogs({ packageType: 'script', config: SCRIPT_DLO_CONFIG }); + expect(scriptLogs.join('\n')).to.include('Package extraction can take several minutes'); + expect(scriptLogs.join('\n')).to.include("Stopping this command doesn't cancel the deployment"); + + const fnLogs = await deployLogs({ + packageType: 'function', + config: { entryPoint: 'entrypoint.py' }, + entrypoint: CHUNKING_ENTRYPOINT, + }); + expect(fnLogs.join('\n')).to.include('Package extraction can take several minutes'); + expect(fnLogs.join('\n')).to.not.include('data transform'); + }); + + it('logs polled statuses only when they change', async () => { + const statuses = ['Initializing', 'Initializing', 'Initializing', 'Deployed']; + let i = 0; + const logs = await deployLogs({ packageType: 'script', config: SCRIPT_DLO_CONFIG }, 'CPU_L', { + getDeploymentStatus: async () => statuses[i++], + }); + expect(logs.filter((m) => m.startsWith('Deployment status:'))).to.deep.equal([ + 'Deployment status: Initializing', + 'Deployment status: Deployed', + ]); + }); + }); + it('infers the feature for a function package and skips the data transform', async () => { const { dir, packageDir } = await makePackage({ packageType: 'function', From cfb2ee925c0508488f91b99eb587a895f5f57ba3 Mon Sep 17 00:00:00 2001 From: Jesus Orosco Date: Wed, 11 Nov 2026 14:41:33 -0800 Subject: [PATCH 5/7] ci: pin java version --- .github/workflows/e2e-smoke.yml | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/.github/workflows/e2e-smoke.yml b/.github/workflows/e2e-smoke.yml index b318067..30123f1 100644 --- a/.github/workflows/e2e-smoke.yml +++ b/.github/workflows/e2e-smoke.yml @@ -37,6 +37,13 @@ jobs: with: python-version: '3.11' + # `script run` starts a local Spark session. PySpark supports Java 8/11/17 and + # fails to launch on the runner's default JDK (24+ rejects its Security Manager flag). + - uses: actions/setup-java@v4 + with: + distribution: temurin + java-version: '17' + - name: Install Python SDK shell: bash env: From 0c20e9f6b359acd4e3fc7c108c7326288721e14e Mon Sep 17 00:00:00 2001 From: Jesus Orosco Date: Thu, 12 Nov 2026 08:37:07 -0800 Subject: [PATCH 6/7] fix: reject a streaming source missing from the read permissions before deploying, Name matching is case-insensitive, matching the backend --- messages/deploy.md | 12 +++++ src/utils/nativeDeploy.ts | 32 +++++++++++- test/utils/nativeDeploy.test.ts | 89 +++++++++++++++++++++++++++++++++ 3 files changed, 131 insertions(+), 2 deletions(-) diff --git a/messages/deploy.md b/messages/deploy.md index b0df4bb..03e607f 100644 --- a/messages/deploy.md +++ b/messages/deploy.md @@ -254,6 +254,18 @@ streamingSource.name is empty in config.json at %s. A streaming transform needs - Set streamingSource.name in config.json to the DLO or DMO the streaming transform reads from. - Run 'sf data-code-extension script scan' to update permissions, then deploy again. +# error.streamingSourceTypeInvalid + +streamingSource.type in config.json at %s is '%s'. It must be 'dlo' or 'dmo'. + +# error.streamingSourceNotInReadPermissions + +streamingSource.name '%s' in config.json at %s isn't listed in %s. A streaming transform can only read from a source declared in its read permissions. + +# actions.streamingSourceNotInReadPermissions + +- Run 'sf data-code-extension script scan' to sync streamingSource.type and permissions.read with streamingSource.name, then deploy again. + # error.dmoRequiresDataObjects DMO transforms require 'dataObjects' in config.json describing the schema of each output DMO. diff --git a/src/utils/nativeDeploy.ts b/src/utils/nativeDeploy.ts index f2c5e5c..515d4cc 100644 --- a/src/utils/nativeDeploy.ts +++ b/src/utils/nativeDeploy.ts @@ -466,12 +466,16 @@ function validateDataTransformConfig(parsed: unknown, configPath: string): DataT dataspace: String(parsed.dataspace), permissions: { read, write }, dataObjects: parsed.dataObjects as DataObject[] | undefined, - ...parseStreamingSource(parsed.streamingSource, configPath), + ...parseStreamingSource(parsed.streamingSource, read, configPath), }; } /** Mirror of `StreamingSource` in `datacustomcode/deploy.py`: optional, but when present needs string `type`/`name`. */ -function parseStreamingSource(value: unknown, configPath: string): { streamingSource?: StreamingSource } { +function parseStreamingSource( + value: unknown, + read: Permission, + configPath: string +): { streamingSource?: StreamingSource } { if (value === undefined || value === null) return {}; if (!isRecord(value) || typeof value.type !== 'string' || typeof value.name !== 'string') { throw new SfError( @@ -491,6 +495,30 @@ function parseStreamingSource(value: unknown, configPath: string): { streamingSo messages.getMessages('actions.streamingSourceNameEmpty') ); } + // Also stricter than the SDK: the backend only extracts from a streaming source listed in + // permissions.read, so catch a hand-edited (or never-scanned) source before creating anything. + // Scan writes the type as the read layer and the name as the only read permission. + const type = value.type.toLowerCase(); + if (type !== 'dlo' && type !== 'dmo') { + throw new SfError( + messages.getMessage('error.streamingSourceTypeInvalid', [configPath, value.type]), + 'StreamingSourceTypeInvalid', + messages.getMessages('actions.streamingSourceNotInReadPermissions') + ); + } + // Case-insensitive, matching the backend's equalsIgnoreCase check. + const sourceName = value.name.toLowerCase(); + if (!read[type]?.some((name) => name.toLowerCase() === sourceName)) { + throw new SfError( + messages.getMessage('error.streamingSourceNotInReadPermissions', [ + value.name, + configPath, + `permissions.read.${type}`, + ]), + 'StreamingSourceNotInReadPermissions', + messages.getMessages('actions.streamingSourceNotInReadPermissions') + ); + } return { streamingSource: { type: value.type, name: value.name } }; } diff --git a/test/utils/nativeDeploy.test.ts b/test/utils/nativeDeploy.test.ts index 46d3410..a8c4c91 100644 --- a/test/utils/nativeDeploy.test.ts +++ b/test/utils/nativeDeploy.test.ts @@ -382,6 +382,70 @@ describe('nativeDeploy.getConfig', () => { }); } + it('accepts a streamingSource type case-insensitively', async () => { + const { dir, packageDir } = await makePackage({ + packageType: 'script', + config: { ...SCRIPT_STREAMING_CONFIG, streamingSource: { type: 'DLO', name: 'Account__dll' } }, + }); + const config = (await getConfig(packageDir, 'script')) as DataTransformConfig; + expect(config.streamingSource).to.deep.equal({ type: 'DLO', name: 'Account__dll' }); + await fs.rm(dir, { recursive: true, force: true }); + }); + + it('matches the streamingSource name against the read permissions case-insensitively', async () => { + const { dir, packageDir } = await makePackage({ + packageType: 'script', + config: { + ...SCRIPT_STREAMING_CONFIG, + streamingSource: { type: 'dlo', name: 'Account__dll' }, + permissions: { read: { dlo: ['account__dll'] }, write: { dlo: ['Result__dll'] } }, + }, + }); + const config = (await getConfig(packageDir, 'script')) as DataTransformConfig; + expect(config.streamingSource).to.deep.equal({ type: 'dlo', name: 'Account__dll' }); + await fs.rm(dir, { recursive: true, force: true }); + }); + + it('rejects a streamingSource with an unknown type', async () => { + const { dir, packageDir } = await makePackage({ + packageType: 'script', + config: { ...SCRIPT_STREAMING_CONFIG, streamingSource: { type: 'table', name: 'Account__dll' } }, + }); + const err = await expectReject(getConfig(packageDir, 'script')); + expect(err.name).to.equal('StreamingSourceTypeInvalid'); + expect(err.message).to.include("'table'"); + expect(err.actions?.join('\n')).to.include('scan'); + await fs.rm(dir, { recursive: true, force: true }); + }); + + const notInReadCases: Array<{ title: string; config: Record }> = [ + { + title: 'the read permissions are empty', + config: { + ...SCRIPT_STREAMING_CONFIG, + permissions: { read: { dlo: [] }, write: { dlo: ['Result__dll'] } }, + }, + }, + { + title: 'the name was edited after scan', + config: { ...SCRIPT_STREAMING_CONFIG, streamingSource: { type: 'dlo', name: 'Contact__dll' } }, + }, + { + title: 'the type does not match the read layer', + config: { ...SCRIPT_STREAMING_CONFIG, streamingSource: { type: 'dmo', name: 'Account__dll' } }, + }, + ]; + for (const { title, config } of notInReadCases) { + it(`rejects a streamingSource not in the read permissions when ${title}`, async () => { + const { dir, packageDir } = await makePackage({ packageType: 'script', config }); + const err = await expectReject(getConfig(packageDir, 'script')); + expect(err.name).to.equal('StreamingSourceNotInReadPermissions'); + expect(err.message).to.include('permissions.read.'); + expect(err.actions?.join('\n')).to.include('scan'); + await fs.rm(dir, { recursive: true, force: true }); + }); + } + it('parses the optional dataObjects schema for a DLO write config', async () => { const { dir, packageDir } = await makePackage({ packageType: 'script', @@ -700,6 +764,31 @@ describe('nativeDeploy.NativeDeployer.deploy', () => { await fs.rm(dir, { recursive: true, force: true }); }); + it('rejects a streaming source missing from the read permissions before any API call', async () => { + // A named source that was never scanned into permissions.read. + const { dir, packageDir } = await makePackage({ + packageType: 'script', + config: { ...SCRIPT_STREAMING_CONFIG, permissions: { read: { dlo: [] }, write: { dlo: ['Result__dll'] } } }, + }); + const { deps, order } = recordingDeps(); + const err = await expectReject( + NativeDeployer.deploy( + { + name: 'my-stream', + version: '1.0.0', + description: 'd', + packageDir, + cpuSize: 'CPU_2XL', + connection: dummyConnection, + }, + deps + ) + ); + expect(err.name).to.equal('StreamingSourceNotInReadPermissions'); + expect(order).to.deep.equal([]); + await fs.rm(dir, { recursive: true, force: true }); + }); + it('deploys a streaming script with StreamingTransform invokeOptions and a STREAMING transform', async () => { const { dir, packageDir } = await makePackage({ packageType: 'script', config: SCRIPT_STREAMING_CONFIG }); const { deps, createdMetadata, dataTransformBodies } = recordingDeps(); From 93062472dc79155c868a99c7970914f17044d33b Mon Sep 17 00:00:00 2001 From: Jesus Orosco Date: Thu, 12 Nov 2026 15:26:47 -0800 Subject: [PATCH 7/7] fix: address review feedback on streaming init, scan, and deploy --- messages/datacodeBinaryExecutor.md | 4 ++ messages/deploy.md | 4 ++ messages/scan.md | 4 ++ src/base/deployBase.ts | 1 + src/base/scanBase.ts | 12 +++- src/utils/nativeDeploy.ts | 24 +++++++- src/utils/nativeInit.ts | 51 ++++++++++++---- src/utils/nativeScan.ts | 23 +++++++- .../commands/data-code-extension/scan.test.ts | 15 +++++ test/utils/nativeDeploy.test.ts | 58 +++++++++++++++++++ test/utils/nativeInit.test.ts | 35 +++++++++++ test/utils/nativeScan.test.ts | 38 +++++++++++- 12 files changed, 254 insertions(+), 15 deletions(-) diff --git a/messages/datacodeBinaryExecutor.md b/messages/datacodeBinaryExecutor.md index 33d811d..3e9511e 100644 --- a/messages/datacodeBinaryExecutor.md +++ b/messages/datacodeBinaryExecutor.md @@ -8,6 +8,10 @@ Could not locate the datacustomcode init template at '%s'. - Run 'pip show salesforce-data-customcode' to confirm the install location - Reinstall with 'pip install --upgrade salesforce-data-customcode' if the templates directory is missing +# error.initRunConfigInvalid + +The datacustomcode run config at '%s' is not valid YAML: %s + # error.initStreamingTemplateNotFound The installed salesforce-data-customcode package (version %s) doesn't include the streaming template at '%s'. StreamingTransform needs version 7.0.0 or later. diff --git a/messages/deploy.md b/messages/deploy.md index 03e607f..0ec3b19 100644 --- a/messages/deploy.md +++ b/messages/deploy.md @@ -245,6 +245,10 @@ config.json at %s is missing required fields: %s. config.json at %s is invalid: %s. +# warning.streamingEntrypointBatchConfig + +%s calls the streaming delta APIs (read_dlo_deltas, write_dlo_deltas, ...), but config.json has no streamingSource, so this deploys as a BatchTransform. If this is meant to be a streaming transform, run 'sf data-code-extension script scan' and set streamingSource.name, then deploy again. + # error.streamingSourceNameEmpty streamingSource.name is empty in config.json at %s. A streaming transform needs the DLO or DMO it reads from. diff --git a/messages/scan.md b/messages/scan.md index da2f551..1baca26 100644 --- a/messages/scan.md +++ b/messages/scan.md @@ -121,6 +121,10 @@ dataLineage references objects that are no longer in permissions (%s entries aff streamingSource.name is empty in config.json. A streaming transform must declare its read source; set streamingSource.name to the DLO or DMO the transform reads from, then scan again. +# warning.ignoredBatchObjects + +The entrypoint calls the streaming delta APIs, so it was scanned as a streaming transform and these batch read/write objects were left out of permissions: %s. Ignore this if the batch calls only run outside DELTA_SYNC mode; otherwise use the delta APIs for them. + # prompt.staleLineage Write the updated permissions to config.json anyway? diff --git a/src/base/deployBase.ts b/src/base/deployBase.ts index 94a0d1a..191acc9 100644 --- a/src/base/deployBase.ts +++ b/src/base/deployBase.ts @@ -153,6 +153,7 @@ export abstract class DeployBase { this.warn(messages.getMessage('warning.streamingSourceNameEmpty')); } + if (codeType === 'script') { + const ignored = await findIgnoredBatchObjects(executionResult.entrypoint); + if (ignored.length > 0) this.warn(messages.getMessage('warning.ignoredBatchObjects', [ignored.join(', ')])); + } + if (codeType === 'script' && executionResult.dryRun) { const stale = findStaleLineage(executionResult.config); if (hasStaleLineage(stale)) this.warnStaleLineage(messages, executionResult.config, stale); diff --git a/src/utils/nativeDeploy.ts b/src/utils/nativeDeploy.ts index 515d4cc..324fcfb 100644 --- a/src/utils/nativeDeploy.ts +++ b/src/utils/nativeDeploy.ts @@ -20,7 +20,7 @@ import { debuglog } from 'node:util'; import { Messages, SfError, type Connection } from '@salesforce/core'; import { decodeHTML } from 'entities'; import { SCRIPT_USE_IN_FEATURE_BATCH, SCRIPT_USE_IN_FEATURE_STREAMING } from './constants.js'; -import { findBaseDirectory, getPackageType, type CodeType } from './nativeScan.js'; +import { fileIsStreaming, findBaseDirectory, getPackageType, type CodeType } from './nativeScan.js'; import { zipWithSfError, ZIP_FILE_NAME, type ZipResult } from './zipBuilder.js'; const debug = debuglog('datacustomcode'); @@ -130,6 +130,8 @@ export type NativeDeployOptions = { /** Authenticated org connection used for the Data Cloud REST calls. */ connection: Connection; log?: (message: string) => void; + /** Non-fatal problems the user should see. Defaults to `log`. */ + warn?: (message: string) => void; }; export type NativeDeployResult = { @@ -789,6 +791,16 @@ export function createStatusReporter(log: (message: string) => void, now: () => }; } +/** `fileIsStreaming`, treating a missing entrypoint as not streaming (the zip step reports it). */ +async function entrypointIsStreaming(entrypointPath: string): Promise { + try { + return await fileIsStreaming(entrypointPath); + } catch (err) { + if ((err as NodeJS.ErrnoException).code === 'ENOENT') return false; + throw err; + } +} + /** * Native TypeScript port of `deploy_full` + the `deploy` command in * `datacustomcode/cli.py`. Preserves the original ordering: validate config @@ -804,6 +816,7 @@ export class NativeDeployer { overrides: Partial = {} ): Promise { const log = opts.log ?? ((): void => {}); + const warn = opts.warn ?? log; const computeType = COMPUTE_TYPES[opts.cpuSize]; if (!computeType) { @@ -848,6 +861,15 @@ export class NativeDeployer { if (streamingSource && !metadata.invokeOptions) { metadata.invokeOptions = [SCRIPT_USE_IN_FEATURE_STREAMING]; } + // Not in `deploy_full`: a delta-API entrypoint with a stale or hand-edited config (no + // streamingSource) would otherwise go out as BATCH and only break at runtime. Warn + // rather than fail, since a batch run can legitimately skip the delta branch. + if (packageType === 'script' && !streamingSource) { + const entryPoint = (config as DataTransformConfig).entryPoint || ENTRYPOINT_FILE; + if (await entrypointIsStreaming(path.join(opts.packageDir, entryPoint))) { + warn(messages.getMessage('warning.streamingEntrypointBatchConfig', [entryPoint])); + } + } if (functionFeature) { log( diff --git a/src/utils/nativeInit.ts b/src/utils/nativeInit.ts index a9718ed..fcc4626 100644 --- a/src/utils/nativeInit.ts +++ b/src/utils/nativeInit.ts @@ -17,7 +17,7 @@ import { promises as fs } from 'node:fs'; import path from 'node:path'; import { debuglog } from 'node:util'; import { Messages, SfError } from '@salesforce/core'; -import { parseDocument } from 'yaml'; +import { isMap, parseDocument } from 'yaml'; import { SCRIPT_USE_IN_FEATURE_BATCH, SCRIPT_USE_IN_FEATURE_OPTIONS, @@ -87,15 +87,22 @@ export async function executeNativeInit(opts: NativeInitOptions): Promise { - const doc = parseDocument(await fs.readFile(sourcePath, 'utf8')); +async function buildStreamingRunConfig(sourcePath: string): Promise { + let raw: string; + try { + raw = await fs.readFile(sourcePath, 'utf8'); + } catch (err) { + if ((err as NodeJS.ErrnoException).code === 'ENOENT') { + throw new SfError( + messages.getMessage('error.initTemplateNotFound', [sourcePath]), + 'InitTemplateNotFound', + messages.getMessages('actions.initTemplateNotFound') + ); + } + throw err; + } + // parseDocument collects problems in doc.errors instead of throwing. + const doc = parseDocument(raw); + if (doc.errors.length > 0 || !isMap(doc.contents)) { + const reason = doc.errors[0]?.message ?? 'expected a YAML mapping at the top level'; + throw new SfError( + messages.getMessage('error.initRunConfigInvalid', [sourcePath, reason]), + 'InitRunConfigInvalid', + messages.getMessages('actions.initTemplateNotFound') + ); + } doc.setIn(['reader_config', 'type_config_name'], 'LocalDeltasReader'); doc.setIn(['writer_config', 'type_config_name'], 'LocalDeltasWriter'); - debug('streaming run config: %s -> %s', sourcePath, destinationPath); - await fs.writeFile(destinationPath, doc.toString(), 'utf8'); + return doc.toString(); } async function copyTemplateEntries( diff --git a/src/utils/nativeScan.ts b/src/utils/nativeScan.ts index e3f5b53..ee497e3 100644 --- a/src/utils/nativeScan.ts +++ b/src/utils/nativeScan.ts @@ -15,10 +15,12 @@ */ import { promises as fs, accessSync } from 'node:fs'; import path from 'node:path'; +import { debuglog } from 'node:util'; import { Messages, SfError } from '@salesforce/core'; Messages.importMessagesDirectoryFromMetaUrl(import.meta.url); const messages = Messages.loadMessages('@salesforce/plugin-data-code-extension', 'datacodeBinaryExecutor'); +const debug = debuglog('datacustomcode'); export type CodeType = 'script' | 'function'; @@ -198,6 +200,18 @@ export async function fileIsStreaming(filePath: string): Promise { return isStreaming((await visitFile(filePath)).streaming); } +/** + * DLO/DMO names from batch calls (`read_dlo`, `write_to_dlo`, ...) in a file that also + * makes delta calls. Such a file is scanned as streaming, so these names never reach + * the permissions; callers warn. Empty when the file is not streaming. Batch calls + * without a resolvable name (like the streaming template's `read_dlo()`) don't count. + */ +export async function findIgnoredBatchObjects(filePath: string): Promise { + const { batch, streaming } = await visitFile(filePath); + if (!isStreaming(streaming)) return []; + return [...new Set([...batch.readDlo, ...batch.readDmo, ...batch.writeToDlo, ...batch.writeToDmo])].sort(); +} + function isStreaming(calls: StreamingDataAccessLayerCalls): boolean { return calls.readDloDeltas || calls.readDmoDeltas || calls.writeDloDeltas.size > 0 || calls.writeDmoDeltas.size > 0; } @@ -534,12 +548,19 @@ export async function updateConfig( * Mirror of `datacustomcode/scan.py:_update_streaming_config`. The streaming read source * cannot be inferred from `read_*_deltas()`, so `streamingSource.name` is kept from the * existing config and is the only read permission; writes come from the scan. + * + * Unlike the SDK, the name is cleared when the scanned read layer differs from the + * existing `streamingSource.type` (e.g. `read_dlo_deltas()` switched to + * `read_dmo_deltas()`): a DLO name is not a valid DMO source, and an empty name makes + * scan warn and deploy fail instead of shipping the mismatch. */ function updateStreamingConfig(existing: Record, calls: StreamingDataAccessLayerCalls): void { validateStreamingAccessLayer(calls); const readLayer = streamingReadLayer(calls); const source = existing.streamingSource; - const name = isPlainObject(source) && typeof source.name === 'string' ? source.name : ''; + const layerChanged = isPlainObject(source) && source.type !== undefined && source.type !== readLayer; + if (layerChanged) debug('streamingSource.type changed %s -> %s; clearing name', String(source.type), readLayer); + const name = !layerChanged && isPlainObject(source) && typeof source.name === 'string' ? source.name : ''; const streamingSource: StreamingSource = { type: readLayer, name }; existing.streamingSource = streamingSource; existing.permissions = { diff --git a/test/commands/data-code-extension/scan.test.ts b/test/commands/data-code-extension/scan.test.ts index 39bc45b..702b3cb 100644 --- a/test/commands/data-code-extension/scan.test.ts +++ b/test/commands/data-code-extension/scan.test.ts @@ -249,6 +249,21 @@ describe('data-code-extension scan commands', () => { .join('\n') ).to.include('streamingSource.name is empty'); }); + + it('warns about named batch calls the streaming scan leaves out of permissions', async () => { + makeScriptPackage(streamingCode + ' c.write_to_dlo("Other__dll", df)\n', { + dataspace: 'default', + streamingSource: { type: 'dlo', name: 'Source__dll' }, + }); + + await ScriptScan.run(['--dry-run']); + + const warnings = sfCommandStubs.warn + .getCalls() + .map((call) => call.args[0] as string) + .join('\n'); + expect(warnings).to.include('left out of permissions').and.include('Other__dll'); + }); }); describe('function scan', () => { diff --git a/test/utils/nativeDeploy.test.ts b/test/utils/nativeDeploy.test.ts index a8c4c91..3bbe1d0 100644 --- a/test/utils/nativeDeploy.test.ts +++ b/test/utils/nativeDeploy.test.ts @@ -817,6 +817,64 @@ describe('nativeDeploy.NativeDeployer.deploy', () => { await fs.rm(dir, { recursive: true, force: true }); }); + it('warns but still deploys BATCH when a delta-API entrypoint has no streamingSource', async () => { + const { dir, packageDir } = await makePackage({ + packageType: 'script', + config: SCRIPT_DLO_CONFIG, + entrypoint: 'df = client.read_dlo_deltas()\nclient.write_dlo_deltas("Result__dll", df)\n', + }); + const { deps, createdMetadata, dataTransformBodies } = recordingDeps(); + const warnings: string[] = []; + + await NativeDeployer.deploy( + { + name: 'my_batch', + version: '1', + description: 'd', + packageDir, + cpuSize: 'CPU_L', + connection: dummyConnection, + warn: (m) => warnings.push(m), + }, + deps + ); + + expect(warnings).to.have.length(1); + expect(warnings[0]).to.include('entrypoint.py').and.include('no streamingSource'); + expect(createdMetadata[0].invokeOptions).to.equal(undefined); + expect((dataTransformBodies[0] as Record).type).to.equal('BATCH'); + await fs.rm(dir, { recursive: true, force: true }); + }); + + it('does not warn for a batch-only entrypoint, a streaming config, or a missing entrypoint', async () => { + const cases = [ + { config: SCRIPT_DLO_CONFIG, entrypoint: 'df = client.read_dlo("Account__dll")\n' }, + { config: SCRIPT_STREAMING_CONFIG, entrypoint: 'df = client.read_dlo_deltas()\n' }, + { config: { ...SCRIPT_DLO_CONFIG, entryPoint: 'missing.py' }, entrypoint: 'df = client.read_dlo_deltas()\n' }, + ]; + for (const { config, entrypoint } of cases) { + // eslint-disable-next-line no-await-in-loop + const { dir, packageDir } = await makePackage({ packageType: 'script', config, entrypoint }); + const warnings: string[] = []; + // eslint-disable-next-line no-await-in-loop + await NativeDeployer.deploy( + { + name: 'n', + version: '1', + description: 'd', + packageDir, + cpuSize: 'CPU_L', + connection: dummyConnection, + warn: (m) => warnings.push(m), + }, + recordingDeps().deps + ); + expect(warnings).to.deep.equal([]); + // eslint-disable-next-line no-await-in-loop + await fs.rm(dir, { recursive: true, force: true }); + } + }); + for (const [label, config, expectedPath, expectedInvokeOptions] of [ ['batch', SCRIPT_DLO_CONFIG, '/services/data/v63.0/ssot/data-custom-code', undefined], ['streaming', SCRIPT_STREAMING_CONFIG, '/services/data/v67.0/ssot/data-custom-code', ['StreamingTransform']], diff --git a/test/utils/nativeInit.test.ts b/test/utils/nativeInit.test.ts index f3bf3e9..98e4158 100644 --- a/test/utils/nativeInit.test.ts +++ b/test/utils/nativeInit.test.ts @@ -244,6 +244,41 @@ describe('executeNativeInit', () => { } }); + it('fails before copying anything when the SDK config.yaml is missing', async () => { + await fs.rm(path.join(pythonPackageLocation, 'datacustomcode', 'config.yaml')); + const packageDir = path.join(workdir, 'my-stream'); + const err = await expectInitError({ + codeType: 'script', + packageDir, + pythonPackageLocation, + pythonPackageVersion: '7.0.0', + useInFeature: 'StreamingTransform', + }); + expect(err.name).to.equal('InitTemplateNotFound'); + expect(err.message).to.include(path.join('datacustomcode', 'config.yaml')); + await expectMissing(packageDir); + }); + + for (const [label, yaml] of [ + ['malformed', 'reader_config:\n type_config_name: [unclosed\n'], + ['not a mapping', '- just\n- a list\n'], + ]) { + it(`fails before copying anything when the SDK config.yaml is ${label}`, async () => { + await fs.writeFile(path.join(pythonPackageLocation, 'datacustomcode', 'config.yaml'), yaml); + const packageDir = path.join(workdir, 'my-stream'); + const err = await expectInitError({ + codeType: 'script', + packageDir, + pythonPackageLocation, + pythonPackageVersion: '7.0.0', + useInFeature: 'StreamingTransform', + }); + expect(err.name).to.equal('InitRunConfigInvalid'); + expect(err.message).to.include(path.join('datacustomcode', 'config.yaml')); + await expectMissing(packageDir); + }); + } + it('reports a missing template, not an old SDK, when a 7.x SDK lacks the streaming template', async () => { const brokenSdk = await buildFakePythonPackage({ streaming: false }); const packageDir = path.join(workdir, 'my-stream-7'); diff --git a/test/utils/nativeScan.test.ts b/test/utils/nativeScan.test.ts index 5457643..0180fba 100644 --- a/test/utils/nativeScan.test.ts +++ b/test/utils/nativeScan.test.ts @@ -24,6 +24,7 @@ import { findBaseDirectory, getPackageType, fileIsStreaming, + findIgnoredBatchObjects, isStreamingSourceNameMissing, scanFile, scanFileForImports, @@ -128,6 +129,26 @@ describe('nativeScan: scanFileStreaming', () => { } }); +describe('nativeScan: findIgnoredBatchObjects', () => { + it('is empty for the streaming template shape (unnamed read_dlo) and for batch-only files', async () => { + const streaming = await writeEntrypoint(STREAMING_ENTRYPOINT); + const batch = await writeEntrypoint('c.read_dlo("In__dll")\nc.write_to_dlo("Out__dll", df)\n'); + expect(await findIgnoredBatchObjects(streaming.file)).to.deep.equal([]); + expect(await findIgnoredBatchObjects(batch.file)).to.deep.equal([]); + await fs.rm(streaming.tmp, { recursive: true, force: true }); + await fs.rm(batch.tmp, { recursive: true, force: true }); + }); + + it('lists named batch reads and writes in a streaming file', async () => { + const { tmp, file } = await writeEntrypoint( + STREAMING_ENTRYPOINT.replace('client.read_dlo()', 'client.read_dlo("B__dll")') + + 'def helper(c, df):\n c.write_to_dlo("A__dll", df)\n' + ); + expect(await findIgnoredBatchObjects(file)).to.deep.equal(['A__dll', 'B__dll']); + await fs.rm(tmp, { recursive: true, force: true }); + }); +}); + describe('nativeScan: isStreamingSourceNameMissing', () => { it('is false for batch configs and named sources, true for an empty or malformed source', () => { expect(isStreamingSourceNameMissing({})).to.equal(false); @@ -480,7 +501,7 @@ describe('nativeScan: updateConfig', () => { entrypoint: STREAMING_ENTRYPOINT, config: { dataspace: 'default', - streamingSource: { type: 'dmo', name: 'Account__dll' }, + streamingSource: { type: 'dlo', name: 'Account__dll' }, permissions: { read: { dlo: ['Stale__dll'] }, write: { dlo: ['Stale__dll'] } }, }, }); @@ -493,6 +514,21 @@ describe('nativeScan: updateConfig', () => { await fs.rm(dir, { recursive: true, force: true }); }); + it('clears streamingSource.name when the scanned read layer no longer matches its type', async () => { + const dir = await makePackage({ + entrypoint: STREAMING_ENTRYPOINT.replace('read_dlo_deltas', 'read_dmo_deltas').replace( + 'write_dlo_deltas', + 'write_dmo_deltas' + ), + config: { dataspace: 'default', streamingSource: { type: 'dlo', name: 'Orders__dll' } }, + }); + const cfg = await updateConfig(path.join(dir, 'payload', 'entrypoint.py'), 'script'); + expect(cfg.streamingSource).to.deep.equal({ type: 'dmo', name: '' }); + expect(isStreamingSourceNameMissing(cfg)).to.equal(true); + expect(cfg.permissions).to.deep.equal({ read: { dmo: [] }, write: { dmo: ['Account_std_copy__dll'] } }); + await fs.rm(dir, { recursive: true, force: true }); + }); + it('writes an empty streaming read permission when streamingSource.name is not set', async () => { const dir = await makePackage({ entrypoint: STREAMING_ENTRYPOINT, config: { dataspace: 'default' } }); const cfg = await updateConfig(path.join(dir, 'payload', 'entrypoint.py'), 'script');