Test your integration

Use your application code in tests

The SDK testing helpers run your workspace calls in memory. Your tests use the same workspace handles, refs and errors as your application, with command results and workspace state that you control. Available in @shardflux/sdk 0.16.0+ and Python shardflux 0.12.0+.

TypeScript

import assert from 'node:assert/strict';
import { createFakeShardflux } from '@shardflux/sdk/testing';

const cloud = createFakeShardflux({
  onExec: (request) => {
    assert.deepEqual(request.argv, ['python3', '-V']);
    return { stdout: 'Python 3.12.3\n', exitCode: 0 };
  },
});
const ref = cloud.workspace('acme/test', { template: 'python-node-browser' });
const result = await ref.exec(['python3', '-V'], { sessionId: 'version-check' });
assert.equal(result.exitCode, 0);
await ref.files.write('/home/user/version.txt', result.stdout);
assert.equal(await ref.files.readText('/home/user/version.txt'), 'Python 3.12.3\n');

Python

from shardflux.testing import create_fake_shardflux

def command(request, workspace):
    assert request['argv'] == ['python3', '-V']
    return {'stdout': 'Python 3.12.3\n', 'exit_code': 0}

with create_fake_shardflux(on_exec=command) as cloud:
    ref = cloud.workspace('acme/test', template='python-node-browser')
    result = ref.exec(['python3', '-V'], session_id='version-check')
    assert result.exit_code == 0
    ref.files.write('/home/user/version.txt', result.stdout)
    assert ref.files.read_text('/home/user/version.txt') == 'Python 3.12.3\n'

Exercise your recovery paths

Get the workspace with await ref.open() (TypeScript) or ref.open() (Python). Set a fault by its ID:

Use quota: 'retained_state' or 'concurrent_workspaces' when creating the TypeScript fake, and quota='retained_state' or 'concurrent_workspaces' in Python. The thrown errors are ShardfluxApiError with the same status, source, code and details as the API. Your application's normal error handling runs in the test.

For a command that stays running, return null from onExec or None from on_exec and start it through ws.cell().exec.start(...) or ws.cell().exec_start(...). Inspect ws.idle() while it runs. End it with cloud.testing.completeCommand(ws.id, sessionId, { exitCode: 0 }) or cloud.testing.complete_command(ws.id, session_id, exit_code=0).

cloud.testing.advance(milliseconds) advances the idle clock. Use latencyMs in TypeScript or latency_ms in Python to exercise request delays; the optional sleep callback integrates with your test runner's clock. A fresh fake starts with fresh state. Explicit exec session IDs make command snapshots repeatable.

Run the same checks against Shardflux

Pass a client into your integration scenario functions. Fast tests pass the fake; an integration job passes a Shardflux client with your project API key. Keep the assertions the same: open the same key twice, reuse it after deleting, filter by labels, check command exit codes, read a missing file and inspect resize results.

Use session workspaces for disposable integration runs and delete them in a finally block. Sessions end instead of suspending, so a session's suspendWhenIdle({ afterSeconds: 0 }) or suspend_when_idle(after_seconds=0) returns 409 conflict with reason session_lifetime. Test successful idle suspension with a persistent workspace, and check that a running command or keepalive postpones it. See workspace lifecycle for the lifecycle calls and errors for handling each refusal.