테스트

SDK는 에이전트 워크플로, 샌드박스 세션, Realtime 세션을 위한 결정론적이고 공급자 중립적인 테스트 더블을 제공합니다. 이러한 테스트 더블은 메모리에서 실행되고 모델, 샌드박스 공급자 또는 Realtime API 요청을 수행하지 않으며, SDK가 소유하는 정규화된 상호작용을 기록합니다.

필요한 레시피 찾기

원하는 작업	사용 항목	이동
고정된 최종 답변 반환	`assistantMessage()`와 함께 `ScriptedModel` 사용	고정 응답 반환
여러 턴의 도구 루프 실행	`functionCall()` 후 어시스턴트 응답	도구 워크플로 테스트
요청에서 응답 선택	`modelResponder()`	요청에서 응답 도출
실행기가 모델에 전송한 내용 검증	`calls`, `firstCall` 또는 `lastCall`	모델 호출 검사
스트리밍 실행 테스트	일반 응답 단계 또는 정확한 이벤트를 위한 `modelStream()`	스트리밍 테스트
오류 또는 재시도 결정 테스트	`modelError()`	모델 실패 주입
의도하지 않은 워크플로 변경 감지	정확한 단계와 `assertComplete()`	워크플로 변동 감지
샌드박스를 시작하지 않고 `SandboxAgent` 테스트	`scriptedSandboxSession()`과 `ScriptedModel`	샌드박스 에이전트 워크플로 테스트
샌드박스 호출 일치 여부 확인 또는 결과 도출	샌드박스 단계의 `match` 또는 `respond`	샌드박스 단계 설정
샌드박스 작업 실패 주입	샌드박스 단계의 `error`	샌드박스 단계 설정
`RealtimeAgent` 함수 도구 테스트	`function_call`을 내보내고 `sendFunctionCallOutput` 예상	Realtime 에이전트 도구 워크플로 테스트
Realtime 작업을 순서대로 일치시킴	`runScenario()`와 `expectCall()`	Realtime 세션 테스트
수신 Realtime 이벤트 내보내기	시나리오 내부의 `emit()`	Realtime 세션 테스트
인터럽션(중단 처리) 요청과 알림 테스트	`expectCall('interrupt')`와 `emit('audio_interrupted')`	인터럽션(중단 처리) 테스트
다음 Realtime 작업 실패 처리	`failNextCall()`	Realtime 실패 주입
원격 연결 해제 테스트	`disconnect(error?)`	원격 연결 해제 테스트
공급자 요청 직렬화 또는 유선 페이로드 테스트	모의 네트워크 전송을 사용하는 실제 공급자 어댑터	올바른 경계 선택
샌드박스 공급자 수명 주기 또는 격리 테스트	통합 테스트의 실제 샌드박스 공급자	올바른 경계 선택

임포트

대부분의 애플리케이션에서는 편의용 하위 경로를 사용할 수 있습니다.

경계	권장 임포트	정식 패키지 임포트
에이전트, `Model`, 샌드박스 워크플로	`@openai/agents/testing`	`@openai/agents-core/testing`
Realtime 전송	`@openai/agents/realtime/testing`	`@openai/agents-realtime/testing`

편의용 하위 경로는 정식 패키지와 동일한 테스트 API를 노출합니다. 테스트 심볼은 기본 런타임 진입점에서 제외됩니다.

에이전트 워크플로 레시피

고정 응답 반환

예상되는 각 모델 호출에 대해 정규화된 출력 항목 배열을 전달합니다. 출력 배열 축약형은 하나의 요청에 대해 결정론적인 응답 ID와 사용량을 받습니다.

import { Agent, Runner } from '@openai/agents';
import { ScriptedModel, assistantMessage } from '@openai/agents/testing';
import assert from 'node:assert/strict';
import test from 'node:test';

test('returns a deterministic final answer', async () => {
  const model = new ScriptedModel([
    [assistantMessage('Paris is the capital of France.')],
  ]);
  const agent = new Agent({
    name: 'Geography assistant',
    model,
  });

  // ScriptedModel replaces model I/O; disable tracing separately so this test
  // makes no network requests.
  const runner = new Runner({ tracingDisabled: true });
  const result = await runner.run(agent, 'What is the capital of France?');

  assert.equal(result.finalOutput, 'Paris is the capital of France.');
  assert.equal(model.calls.length, 1);
  // Also fail if a later agent change stops before using the whole script.
  model.assertComplete();
});

항상 model.assertComplete()로 테스트를 마무리합니다. 워크플로가 설정된 단계를 모두 소비하기 전에 중지되는 경우를 감지합니다.

도구 워크플로 테스트

도구를 호출하는 모델 응답 하나와 최종 답변을 생성하는 두 번째 응답을 스크립트로 작성합니다. 실제 SDK 도구 파이프라인은 이러한 모델 호출 사이에서 실행됩니다.

import { Agent, Runner, tool } from '@openai/agents';
import {
  ScriptedModel,
  assistantMessage,
  functionCall,
} from '@openai/agents/testing';
import assert from 'node:assert/strict';
import test from 'node:test';
import { z } from 'zod';

test('runs a multi-turn tool workflow', async () => {
  const weather = tool({
    name: 'get_weather',
    description: 'Gets the weather for a city.',
    parameters: z.object({ city: z.string() }),
    execute: async ({ city }) => `${city}: sunny`,
  });
  const model = new ScriptedModel([
    // The first model turn enters the real SDK tool execution pipeline.
    [functionCall('get_weather', { city: 'Tokyo' }, { callId: 'call_1' })],
    // The second turn sees the tool result and finishes the workflow.
    [assistantMessage('It is sunny in Tokyo.')],
  ]);
  const agent = new Agent({
    name: 'Weather assistant',
    model,
    tools: [weather],
  });

  // ScriptedModel replaces model I/O; disable tracing separately so this test
  // makes no network requests.
  const runner = new Runner({ tracingDisabled: true });
  const result = await runner.run(agent, 'What is the weather in Tokyo?');

  assert.equal(result.finalOutput, 'It is sunny in Tokyo.');
  assert.equal(model.calls.length, 2);
  const lastInput = model.lastCall?.request.input;
  assert(Array.isArray(lastInput));
  assert(lastInput.some((item) => item.type === 'function_call_result'));
  model.assertComplete();
});

이 패턴은 도구 입력 검증, 실행, 결과 변환, 훅, 가드레일, 다음 모델 턴을 실행합니다. 도구 함수를 직접 호출하여 도구 런타임을 우회하지 않습니다.

요청에서 응답 도출

응답이 정규화된 요청에 따라 달라지거나 검증이 모델 경계에서 이루어져야 하는 경우 modelResponder()를 사용합니다. 콜백은 동기 또는 비동기일 수 있습니다.

import { Agent, Runner } from '@openai/agents';
import {
  ScriptedModel,
  assistantMessage,
  modelResponder,
} from '@openai/agents/testing';
import assert from 'node:assert/strict';
import test from 'node:test';

test('derives a response from the recorded request', async () => {
  const model = new ScriptedModel([
    modelResponder((call) => {
      // The responder receives the normalized request at the Model boundary.
      assert.equal(call.index, 0);
      assert.equal(call.streamed, false);
      assert.deepEqual(call.request.input, [
        {
          type: 'message',
          role: 'user',
          content: 'Summarize this',
        },
      ]);
      return [assistantMessage(`Handled model call ${call.index}.`)];
    }),
  ]);
  const agent = new Agent({ name: 'Assistant', model });

  // ScriptedModel replaces model I/O; disable tracing separately so this test
  // makes no network requests.
  const runner = new Runner({ tracingDisabled: true });
  const result = await runner.run(agent, 'Summarize this');

  assert.equal(result.finalOutput, 'Handled model call 0.');
  model.assertComplete();
});

실제로 요청에 따라 달라지는 동작에 응답자를 사용합니다. 단순한 순서만으로 충분하다면 고정 단계를 사용하는 것이 좋습니다. 고정 스크립트를 사용하면 예상하지 못한 턴을 더 쉽게 진단할 수 있습니다.

모델 호출 검사

ScriptedModel은 선택된 단계를 처리하거나 예외를 발생시키기 전에 호출을 기록합니다.

멤버	포함 내용
`calls`	호출 순서대로 기록된 모든 `RecordedModelCall`
`firstCall`	첫 번째 호출 또는 `undefined`
`lastCall`	가장 최근 호출 또는 `undefined`
`remainingSteps`	아직 소비되지 않은 설정 단계 수

각 호출에는 calls 내 위치와 일치하는 0부터 시작하는 index, streamed 플래그, 정규화된 request가 있습니다. 기록된 요청의 변경 가능한 데이터 컨테이너는 호출 시점에 분리됩니다. 반환된 스냅샷 하나를 읽거나 변경해도 모델이 보관하는 기록은 다시 작성되지 않습니다. 콜백, 클래스 인스턴스, AbortSignal 같은 런타임 객체는 동일성을 유지합니다.

일반적인 검증 항목은 다음과 같습니다.

사용자 입력, 도구 호출, 도구 결과를 확인하는 call.request.input
모델에 전송된 유효 설정을 확인하는 call.request.modelSettings
준비된 기능을 확인하는 call.request.tools와 call.request.handoffs
스트리밍 경로와 비스트리밍 경로를 구분하는 call.streamed

스트리밍 테스트

일반 응답 단계는 스트리밍 실행과 비스트리밍 실행 모두에서 작동합니다. run(..., { stream: true })이 모델을 호출하면 ScriptedModel은 정규화된 시작 이벤트, 어시스턴트 텍스트 델타, 전체 출력과 사용량이 포함된 종료 응답 이벤트를 내보냅니다.

자동 생성된 스트림 이벤트는 분리된 스냅샷입니다. 산출된 델타를 변경해도 이후의 종료 응답 이벤트는 변경되지 않습니다.

정확한 정규화 StreamEvent 순서가 테스트 대상 동작의 일부인 경우에만 modelStream(events)을 사용합니다. 해당 순서가 기록된 호출에 따라 달라지는 경우 modelStreamResponder(callback)을 사용합니다. 이러한 이벤트는 SDK에서 정규화된 이벤트이며 Responses 또는 Chat Completions 유선 청크가 아닙니다.

비스트리밍 호출에 사용된 원문 스트림 단계는 reason: 'incompatible_call'이 포함된 InvalidScriptedModelStepError를 발생시킵니다.

모델 실패 주입

하나의 모델 호출을 실패시키려면 modelError(error, retryAdvice?)를 사용합니다. 재시도 조언은 고정 ModelRetryAdvice 값 또는 콜백일 수 있습니다. 실행기가 수행하는 재시도는 또 다른 모델 호출이며 다음 스크립트 단계를 소비합니다.

modelError(error)
modelError(error, { suggested: true, replaySafety: 'safe' })
modelError(error, ({ attempt }) => ({ suggested: attempt === 1 }))

위 텍스트가 TypeScript 예제가 아니라 시그니처를 보여주는 이유는 주변 실행기의 재시도 정책에 따라 조언이 또 다른 시도로 이어지는지 결정되기 때문입니다.

이미 중단된 요청은 단계를 소비하지 않습니다. 기록된 호출이 중단 검사 지점에 도달한 후 요청이 중단되면 이름이 AbortError이고 callIndex가 포함된 오류가 발생합니다.

사용자 지정 코드가 실행되는 동안에는 취소가 협력적으로 이루어집니다. ScriptedModel은 진행 중인 modelResponder() 또는 modelStreamResponder() 콜백이나 비동기 이터레이터에서 대기 중인 next()를 요청 신호와 경합시키지 않습니다. 무기한 대기할 수 있는 코드는 call.request.signal 자체를 관찰하거나 이터레이터 내부에서 동일한 신호를 관찰한 다음 완료되어야 합니다. 콜백이나 이터레이터가 제어를 반환하면 ScriptedModel은 응답을 수락하거나 다음 이벤트를 내보내기 전에 신호를 다시 확인합니다.

워크플로 변동 감지

스크립트로 작성된 턴을 예상 워크플로 형태로 간주합니다. 아래 테스트는 두 턴으로 이루어진 주문 조회에서 통과하며, 다른 검증이나 워크플로 자체가 실패할 때도 스크립트를 확인할 수 있도록 해제 단계에서 assertComplete()를 실행합니다.

예를 들어 애플리케이션 코드가 나중에 lookup_order에 승인이 필요하도록 변경되면 첫 번째 스크립트 응답 후에 실행이 중지됩니다. 그러면 해제 단계에서 소비되지 않은 단계 하나가 보고됩니다. 오케스트레이션 변경으로 인해 예상 단계 후에 추가 모델 요청이 발생하면 해당 요청에서 모델이 실패합니다. 따라서 기존 회귀 테스트는 실제 모델이 새로운 경로를 추론하게 하지 않고도 에이전트의 제어 흐름 변경을 감지합니다.

import { Agent, type Model, Runner, tool } from '@openai/agents';
import {
  ScriptedModel,
  assistantMessage,
  functionCall,
} from '@openai/agents/testing';
import assert from 'node:assert/strict';
import test from 'node:test';
import { z } from 'zod';

function createOrderAgent(model: Model) {
  const lookupOrder = tool({
    name: 'lookup_order',
    description: 'Looks up an order.',
    parameters: z.object({ orderId: z.string() }),
    execute: async ({ orderId }) => `${orderId}: shipped`,
  });

  return new Agent({
    name: 'Order assistant',
    model,
    tools: [lookupOrder],
  });
}

test('preserves the order lookup workflow contract', async (t) => {
  const model = new ScriptedModel([
    [
      functionCall(
        'lookup_order',
        { orderId: 'order_123' },
        { callId: 'call_1' },
      ),
    ],
    [assistantMessage('Order 123 has shipped.')],
  ]);
  // Keep this assertion in teardown so an early workflow exit is reported
  // even when another assertion fails first.
  t.after(() => model.assertComplete());
  const agent = createOrderAgent(model);

  // ScriptedModel replaces model I/O; disable tracing separately so this test
  // makes no network requests.
  const runner = new Runner({ tracingDisabled: true });
  const result = await runner.run(agent, 'Where is order 123?');

  assert.equal(result.finalOutput, 'Order 123 has shipped.');
  assert.equal(model.calls.length, 2);
});

일반 회귀 테스트에서는 이러한 불일치 오류를 포착하지 마세요. 테스트가 실패하도록 두고 오류 유형과 필드를 사용해 예상 워크플로의 어느 쪽이 변경되었는지 진단합니다.

오류	구조화된 필드	의미
`UnexpectedModelCallError`	`callIndex`, `streamed`	스크립트가 끝난 후 워크플로가 추가 모델 호출을 수행함
`UnconsumedModelStepsError`	`remainingSteps`	모든 단계를 사용하기 전에 워크플로가 종료됨
`InvalidScriptedModelStepError`	`reason`, `inputIndex?`, `callIndex?`, `stepType?`	단계의 형식이 잘못되었거나 호출 모드와 호환되지 않음
`ScriptedModelRequestAbortedError`	`callIndex?`	스크립트 작업 전이나 도중에 요청이 중단됨

형식이 잘못된 단계 엔벌로프는 모델 호출이 이를 소비하기 전에 생성자 또는 enqueue()에서 거부됩니다. 따라서 설정 오류가 워크플로 후반이 아니라 테스트 정의 근처에서 실패합니다.

샌드박스 에이전트 레시피

샌드박스 에이전트 워크플로 테스트

Docker 컨테이너나 원격 샌드박스를 생성하지 않고 실제 SandboxAgent 런타임을 실행하려면 ScriptedModel과 scriptedSandboxSession()을 결합합니다. 모델 스크립트는 기능 도구를 선택하고, 샌드박스 스크립트는 해당 SandboxSession 메서드가 반환하는 내용을 정의합니다.

import { Runner } from '@openai/agents';
import { SandboxAgent, shell } from '@openai/agents/sandbox';
import {
  ScriptedModel,
  assistantMessage,
  functionCall,
  scriptedSandboxSession,
} from '@openai/agents/testing';
import assert from 'node:assert/strict';
import test from 'node:test';

test('runs a SandboxAgent shell workflow without a real sandbox', async (t) => {
  const sandbox = scriptedSandboxSession([
    {
      method: 'execCommand',
      match: ({ cmd }) => {
        // Assert the command at the sandbox boundary, after the shell tool has
        // validated and converted the model's function-call arguments.
        assert.equal(cmd, 'pwd');
      },
      result: '/workspace\n',
    },
  ]);
  const model = new ScriptedModel([
    [functionCall('exec_command', { cmd: 'pwd' }, { callId: 'call_1' })],
    [assistantMessage('The workspace is /workspace.')],
  ]);
  const agent = new SandboxAgent({
    name: 'Workspace assistant',
    model,
    capabilities: [shell()],
  });

  // Teardown assertions also report unused steps if the workflow exits early.
  t.after(() => {
    sandbox.assertComplete();
    model.assertComplete();
  });

  // The scripted boundaries do not disable the separate tracing exporter.
  const runner = new Runner({ tracingDisabled: true });
  const result = await runner.run(agent, 'Which directory are you in?', {
    // The real SandboxAgent runtime receives the in-memory session instead of
    // creating a Docker or remote sandbox.
    sandbox: { session: sandbox },
  });

  assert.equal(result.finalOutput, 'The workspace is /workspace.');
  assert.deepEqual(
    sandbox.calls.map((call) => call.method),
    ['execCommand'],
  );
  assert.equal(model.calls.length, 2);
});

이 테스트는 SDK가 소유하는 두 정규화된 경계를 모두 통과합니다. ScriptedModel은 모델 대상 턴을 구동하고, 스크립트 기반 샌드박스는 실제 셸 기능이 생성한 execCommand 인수를 받습니다. 따라서 이 테스트는 도구 인수 검증, 기능 라우팅, 샌드박스 세션 호출, 다음 모델 턴으로의 도구 결과 전달, 최종 출력 처리를 다룹니다.

실제 모델이 명령을 선택하는지, 샌드박스 공급자가 성공적으로 시작되는지, 해당 공급자가 명령을 어떻게 실행하는지는 테스트하지 않습니다. 모델 결정 품질에는 평가를 사용하고, 공급자 수명 주기, 파일 시스템, 프로세스, 격리 동작에는 실제 샌드박스 공급자를 사용하는 통합 테스트를 사용합니다.

샌드박스 단계 설정

각 단계는 SandboxSession 메서드 호출 하나를 소비합니다. 정확히 하나의 결과를 선택하고 인수가 중요한 경우에만 매처를 추가합니다.

단계 멤버	사용 시점
`result`	메서드가 고정값을 반환해야 하는 경우
`respond(call)`	결과가 형식화된 호출 인수 또는 호출 인덱스에 따라 달라지는 경우
`error`	메서드가 특정 실패를 발생시키거나 거부해야 하는 경우
`match(...args)`	결과를 생성하기 전에 예상하지 못한 메서드 인수를 테스트에서 거부해야 하는 경우

샌드박스 메서드는 supportsPty() 같은 동기 메서드와 execCommand() 같은 비동기 메서드를 포함하여 하나의 전역 순서로 일치 여부를 확인합니다. 스크립트에서 이름을 지정한 메서드만 반환된 세션에 설치됩니다. 이를 통해 기능 감지가 유지됩니다. 예를 들어 워크플로가 대화형 셸 지원을 노출해야 한다면 supportsPty와 writeStdin을 모두 스크립트로 작성합니다.

sandbox.calls는 호출 시점에 분리된 인수 스냅샷을 기록하며, 배열 내 각 호출 위치와 일치하는 0부터 시작하는 index, method, 위치 기반 args를 포함합니다. 실패를 진단할 때 sandbox.remainingSteps를 사용하고, 워크플로가 일찍 종료되어 예상 작업이 사용되지 않은 채 남는 일을 방지하려면 sandbox.assertComplete()로 테스트를 마무리합니다.

오류	구조화된 필드	의미
`UnexpectedSandboxCallError`	`callIndex`, `actualMethod`, `expectedMethod`, `remainingSteps`	워크플로가 잘못된 메서드를 호출했거나 스크립트 종료 후에도 계속됨
`SandboxCallMatcherError`	`callIndex`, `method`	단계의 매처가 `false`를 반환함
`UnconsumedSandboxStepsError`	`remainingSteps`, `pendingMethods`	모든 단계를 사용하기 전에 워크플로가 종료됨
`InvalidScriptedSandboxStepError`	`reason`, `inputIndex`, `method?`	단계의 형식이 잘못되었거나 지원되지 않는 메서드 이름을 사용함

Realtime 에이전트 및 세션 레시피

Realtime 에이전트 도구 워크플로 테스트

실제 함수 도구를 RealtimeAgent에 연결한 다음, 스크립트 기반 전송에서 정규화된 function_call을 내보냅니다. RealtimeSession은 해당 에이전트에서 도구를 확인하고 SDK 도구 파이프라인을 통해 검증 및 실행한 다음 sendFunctionCallOutput을 통해 결과를 다시 전송합니다.

import { RealtimeAgent, RealtimeSession, tool } from '@openai/agents/realtime';
import { ScriptedRealtimeTransport } from '@openai/agents/realtime/testing';
import assert from 'node:assert/strict';
import test from 'node:test';
import { z } from 'zod';

test('executes a RealtimeAgent tool call', async () => {
  let lookedUpOrderId: string | undefined;
  const lookupOrder = tool({
    name: 'lookup_order',
    description: 'Looks up an order by ID.',
    parameters: z.object({ orderId: z.string() }),
    execute: async ({ orderId }) => {
      lookedUpOrderId = orderId;
      return `Order ${orderId} has shipped.`;
    },
  });
  const agent = new RealtimeAgent({
    name: 'Order assistant',
    instructions: 'Help customers track their orders.',
    tools: [lookupOrder],
  });
  const transport = new ScriptedRealtimeTransport();
  const session = new RealtimeSession(agent, { transport });
  let markToolOutputReturned!: () => void;
  let rejectToolOutput!: (reason?: unknown) => void;
  const toolOutputReturned = new Promise<void>((resolve, reject) => {
    markToolOutputReturned = resolve;
    rejectToolOutput = reject;
  });
  // Observe a scenario failure that happens before exercise reaches its await.
  // Awaiting the original promise still receives the same rejection.
  void toolOutputReturned.catch(() => undefined);

  await transport.runScenario({
    // Tool execution is asynchronous. Bound the scenario so a missing tool
    // output fails here instead of waiting for the test runner's timeout.
    signal: AbortSignal.timeout(5_000),
    scenario: async ({ expectCall, emit }) => {
      try {
        await expectCall('connect');
        await expectCall('sendMessage', (call) => {
          assert.equal(call.message, 'Where is order 123?');
        });

        // Script the model choosing the agent's tool. This tests how the SDK
        // handles that choice, not whether a real model would make the choice.
        emit('turn_started', {
          type: 'response_started',
          providerData: { response: { id: 'response_1' } },
        });
        emit('function_call', {
          type: 'function_call',
          name: 'lookup_order',
          callId: 'call_1',
          arguments: JSON.stringify({ orderId: 'order_123' }),
          responseId: 'response_1',
        });

        // RealtimeSession runs the actual tool and sends its result back through
        // the transport so the model could continue the response.
        await expectCall('sendFunctionCallOutput', (call) => {
          assert.equal(call.toolCall.callId, 'call_1');
          assert.equal(call.output, 'Order order_123 has shipped.');
          assert.equal(call.startResponse, true);
        });
        markToolOutputReturned();
        await expectCall('close');
      } catch (error) {
        // Unblock exercise so its finally block can close the session.
        rejectToolOutput(error);
        throw error;
      }
    },
    exercise: async () => {
      try {
        await session.connect({ apiKey: 'test' });
        session.sendMessage('Where is order 123?');
        await toolOutputReturned;
      } finally {
        // Clean up even when the scenario fails or reaches its time limit.
        session.close();
      }
    },
  });

  assert.equal(lookedUpOrderId, 'order_123');
  transport.assertComplete();
  transport.assertClosed();
});

도구와 비즈니스 로직이 RealtimeAgent에 속하므로 이는 에이전트 테스트입니다. 스크립트 기반 function_call은 해당 도구를 사용하겠다는 모델의 결정을 나타냅니다. 실제 모델이 에이전트의 instructions를 따르거나 도구를 선택한다는 것을 입증하지는 않습니다. 모델 결정 품질은 평가 또는 통합 테스트로 다룹니다.

이 실행은 세션을 닫기 전에 비동기 도구 출력을 기다리므로, 예제에서는 대기 시간에 제한을 두기 위해 AbortSignal을 전달합니다. 순서가 지정된 호출의 불일치는 여전히 즉시 실패하며, 신호는 회귀로 인해 예상 도구 출력이 전혀 전송되지 않는 경우를 처리합니다.

에이전트 소유 핸드오프, 도구 승인, 도구 가드레일을 테스트할 때도 동일한 패턴을 사용합니다. 동작을 시작하는 정규화된 입력을 내보낸 다음 실제 SDK 파이프라인이 생성한 세션 이벤트와 송신 전송 호출을 검증합니다.

Realtime 세션 테스트

실제 RealtimeSession에 ScriptedRealtimeTransport를 전달합니다. 테스트 실행기 시간 제한에 의존하지 않고 예상 전송 상호작용과 세션 실행을 함께 수행하려면 runScenario()를 사용합니다. 아래 예제는 두 번의 사용자 턴을 시뮬레이션하고 전송에서 내보낸 정규화 항목이 실제 세션 기록을 업데이트하는지 확인합니다.

import {
  RealtimeAgent,
  type RealtimeItem,
  RealtimeSession,
} from '@openai/agents/realtime';
import { ScriptedRealtimeTransport } from '@openai/agents/realtime/testing';
import assert from 'node:assert/strict';
import test from 'node:test';

function createScenarioGate() {
  let release!: () => void;
  let fail!: (reason: unknown) => void;
  const promise = new Promise<void>((resolve, reject) => {
    release = resolve;
    fail = reject;
  });
  // Observe a scenario failure that happens before exercise reaches its await.
  // Awaiting the original promise still receives the same rejection.
  void promise.catch(() => undefined);
  return { promise, release, fail };
}

test('runs a two-turn Realtime conversation', async () => {
  const transport = new ScriptedRealtimeTransport();
  const session = new RealtimeSession(
    new RealtimeAgent({ name: 'Assistant' }),
    { transport },
  );
  const conversation: RealtimeItem[] = [
    {
      itemId: 'user_1',
      type: 'message',
      role: 'user',
      status: 'completed',
      content: [{ type: 'input_text', text: 'Hello' }],
    },
    {
      itemId: 'assistant_1',
      type: 'message',
      role: 'assistant',
      status: 'completed',
      content: [{ type: 'output_text', text: 'Hi! How can I help?' }],
    },
    {
      itemId: 'user_2',
      type: 'message',
      role: 'user',
      status: 'completed',
      content: [{ type: 'input_text', text: 'What did I just say?' }],
    },
    {
      itemId: 'assistant_2',
      type: 'message',
      role: 'assistant',
      status: 'completed',
      content: [{ type: 'output_text', text: 'You said hello.' }],
    },
  ];
  const signal = AbortSignal.timeout(5_000);
  const firstTurnDelivered = createScenarioGate();
  let secondTurnDelivered: ReturnType<typeof createScenarioGate> | undefined;

  await transport.runScenario({
    // The application waits for each scripted turn. Bound that coordination
    // so a missing outbound call cannot leave both callbacks waiting forever.
    signal,
    // This side scripts the normalized transport-facing interaction.
    scenario: async ({ expectCall, emit }) => {
      try {
        await expectCall('connect');

        await expectCall('sendMessage', (call) => {
          assert.equal(call.message, 'Hello');
        });
        // Emit normalized history items to exercise RealtimeSession history.
        emit('item_update', conversation[0]);
        emit('item_update', conversation[1]);
        // Create the next gate before releasing the application to send again.
        secondTurnDelivered = createScenarioGate();
        firstTurnDelivered.release();

        await expectCall('sendMessage', (call) => {
          assert.equal(call.message, 'What did I just say?');
        });
        emit('item_update', conversation[2]);
        emit('item_update', conversation[3]);
        secondTurnDelivered.release();

        await expectCall('close');
      } catch (error) {
        // Forward scenario failures to whichever application wait is active.
        (secondTurnDelivered ?? firstTurnDelivered).fail(error);
        throw error;
      }
    },
    // This side drives the application through the public session API.
    exercise: async () => {
      try {
        await session.connect({ apiKey: 'test' });
        session.sendMessage('Hello');
        await firstTurnDelivered.promise;

        session.sendMessage('What did I just say?');
        assert(secondTurnDelivered);
        await secondTurnDelivered.promise;
      } finally {
        // Close the session after success, timeout, or scenario failure.
        session.close();
      }
    },
  });

  assert.deepEqual(session.history, conversation);
  transport.assertComplete();
  transport.assertClosed();
});

scenario 콜백은 전송 대상 측을 스크립트로 작성합니다. 즉, 송신 호출의 일치 여부를 확인하고 정규화된 수신 이벤트를 내보냅니다. exercise 콜백은 애플리케이션 코드처럼 동작하며 공개 세션 API를 구동합니다. 예제의 주석은 이 경계를 명시적으로 표시합니다. 이는 공급자 서버 프레임이나 변환이 아니라 RealtimeSession 동작을 테스트합니다.

애플리케이션은 다음 메시지를 보내기 전에 스크립트로 작성된 각 턴을 기다리므로, 예제에서는 제한이 설정된 AbortSignal을 runScenario()에 전달합니다. 신호가 만료되면 runScenario()은 대기 중인 기대 항목을 거부하고, 시나리오는 해당 실패를 활성 테스트 전용 조정 게이트로 전달하며, exercise는 두 콜백을 모두 차단된 상태로 두는 대신 세션 정리를 실행합니다.

expectCall(method, matcher?)은 메서드별로 형식이 지정되며 하나의 전역 순서로 호출의 일치 여부를 확인합니다. 매처는 검증을 포함할 수 있으며 호출을 거부하려면 false를 반환할 수 있습니다. 기대 항목이 등록되기 직전에 호출이 발생하면 전송은 다음 기대 항목을 위해 이를 보관합니다.

runScenario()은 어느 한쪽이 실패하거나, AbortSignal이 시나리오를 취소하거나, 기대 항목이 완료되지 않은 상태에서 exercise가 끝나면 종료됩니다. 기대 항목이 남아 있는 동안 exercise가 완료되면 IncompleteRealtimeScenarioError가 발생합니다. 위 예제와 같이 서로 대기하는 콜백이 명시적인 신호 기한에 도달하면 runScenario()은 ScriptedRealtimeScenarioCancelledError를 생성하고 대기 중인 기대 항목을 보고합니다.

인터럽션(중단 처리) 테스트

애플리케이션에 중요한 방향으로 인터럽션(중단 처리)을 테스트합니다.

중지 버튼 또는 유사한 애플리케이션 동작을 테스트하려면 어시스턴트 오디오 응답의 시작을 내보내고 세션이 이를 관찰할 때까지 기다린 다음, session.interrupt()를 호출하고 송신 interrupt 호출을 예상합니다.
인터럽션(중단 처리) 알림에 대한 애플리케이션 처리를 테스트하려면 전송에서 audio_interrupted를 내보내고 세션 리스너를 검증합니다.

예제에서는 하나의 이벤트가 자동으로 다른 이벤트를 발생시킨다는 인상을 주지 않도록 이를 별도 테스트로 유지합니다.

import { RealtimeAgent, RealtimeSession } from '@openai/agents/realtime';
import { ScriptedRealtimeTransport } from '@openai/agents/realtime/testing';
import assert from 'node:assert/strict';
import test from 'node:test';

function createScenarioGate() {
  let release!: () => void;
  let fail!: (reason: unknown) => void;
  const promise = new Promise<void>((resolve, reject) => {
    release = resolve;
    fail = reject;
  });
  // Observe a scenario failure that happens before exercise reaches its await.
  // Awaiting the original promise still receives the same rejection.
  void promise.catch(() => undefined);
  return { promise, release, fail };
}

function createSession() {
  const transport = new ScriptedRealtimeTransport();
  const session = new RealtimeSession(
    new RealtimeAgent({ name: 'Assistant' }),
    { transport },
  );
  return { session, transport };
}

test('sends an interrupt request through the transport', async () => {
  const { session, transport } = createSession();
  const signal = AbortSignal.timeout(5_000);
  const readyToInterrupt = createScenarioGate();
  let sessionObservedAudio = false;
  session.once('audio_start', () => {
    sessionObservedAudio = true;
  });

  await transport.runScenario({
    signal,
    scenario: async ({ expectCall, emit }) => {
      try {
        await expectCall('connect');
        await expectCall('sendMessage');

        // Simulate the beginning of an assistant utterance. RealtimeSession
        // emits audio_start when it receives the first audio chunk.
        emit('turn_started', {
          type: 'response_started',
          providerData: { response: { id: 'response_1' } },
        });
        emit('audio', {
          type: 'audio',
          responseId: 'response_1',
          data: new Uint8Array([1, 2, 3]).buffer,
        });
        // emit() delivers events synchronously. Verify that RealtimeSession saw
        // the chunk before releasing the concurrently running application side.
        assert.equal(sessionObservedAudio, true);
        readyToInterrupt.release();

        await expectCall('interrupt');
        await expectCall('close');
      } catch (error) {
        readyToInterrupt.fail(error);
        throw error;
      }
    },
    exercise: async () => {
      try {
        await session.connect({ apiKey: 'test' });
        session.sendMessage('Tell me a long story');
        await readyToInterrupt.promise;

        // Application code, such as a stop button, interrupts active playback.
        session.interrupt();
      } finally {
        session.close();
      }
    },
  });

  transport.assertComplete();
  transport.assertClosed();
});

test('forwards an interruption notification to the application', async () => {
  const { session, transport } = createSession();
  let interruptionObserved = false;
  session.once('audio_interrupted', () => {
    interruptionObserved = true;
  });

  const connectCall = transport.expectCall('connect');
  await session.connect({ apiKey: 'test' });
  await connectCall;

  // Independently, the transport can notify the session of an interruption.
  transport.emit('audio_interrupted');
  assert.equal(interruptionObserved, true);

  const closeCall = transport.expectCall('close');
  session.close();
  await closeCall;
  transport.assertComplete();
  transport.assertClosed();
});

Realtime 실패 주입

failNextCall(method, error)은 해당 메서드의 다음 호출에서 제공된 값을 발생시킵니다. 해당 시도는 계속 기록되고 일치 여부가 확인되므로 테스트에서 작업과 실패를 모두 검증할 수 있습니다.

import { RealtimeAgent, RealtimeSession } from '@openai/agents/realtime';
import { ScriptedRealtimeTransport } from '@openai/agents/realtime/testing';
import assert from 'node:assert/strict';
import test from 'node:test';

test('injects a transport failure into the next matching call', async () => {
  const transport = new ScriptedRealtimeTransport();
  const session = new RealtimeSession(
    new RealtimeAgent({ name: 'Assistant' }),
    { transport },
  );
  const failure = new Error('connection failed');

  // Failure injection and call matching are independent: the failed attempt
  // is still recorded and must satisfy the expectation.
  transport.failNextCall('connect', failure);
  const connectCall = transport.expectCall('connect');

  await assert.rejects(session.connect({ apiKey: 'test' }), failure);
  assert.equal((await connectCall).method, 'connect');
  assert.equal(transport.status, 'disconnected');
  transport.assertComplete();
  transport.assertClosed();
});

한 메서드에 대한 여러 실패는 등록 순서대로 소비됩니다. 사용되지 않은 주입 실패가 있으면 assertComplete()가 실패합니다.

원격 연결 해제 테스트

원격 엔드포인트가 연결을 닫는 상황을 시뮬레이션하려면 transport.disconnect(error?)를 호출합니다. 이 메서드는 전송 상태를 disconnected로 전환하고 connection_change를 내보내며 선택적으로 전송 error를 내보냅니다.

연결 손실 및 원격 종료에는 disconnect()를 사용합니다. 애플리케이션이 정리를 시작하고 테스트에서 송신 close 호출을 예상해야 할 때는 session.close()를 사용합니다.

Realtime 호출 검사

transport.calls는 호출 순서대로 분리된 스냅샷을 포함합니다. 변경 가능한 배열, 일반 객체, 버퍼는 호출 경계에서 복사되며 런타임 클래스 인스턴스는 동일성을 유지합니다. 기록된 connect 호출에는 apiKeyProvided가 포함되지만 API 키는 저장되지 않습니다.

일반적인 메서드 이름에는 connect, sendEvent, requestResponse, sendMessage, addImage, sendAudio, updateSessionConfig, close, mute, sendFunctionCallOutput, interrupt, resetHistory, sendMcpResponse가 포함됩니다. 내보낸 메서드와 호출 유형은 RealtimeTransportLayer에서 파생되므로 새로운 전송 메서드가 형식화된 테스트 표면에서 조용히 사라질 수 없습니다.

완료 및 정리 확인

다음 두 검증으로 Realtime 테스트를 마무리합니다.

검증	확인 내용
`transport.assertComplete()`	일치하지 않은 호출, 대기 중인 기대 항목 또는 사용되지 않은 주입 실패가 남아 있지 않음
`transport.assertClosed()`	수명 주기 상태가 `disconnected`임

Realtime 오류도 구조화된 상태를 노출합니다.

오류	구조화된 필드
`UnexpectedRealtimeCallError`	`callIndex`, `expectedMethod`, `actualMethod`
`RealtimeCallMatcherError`	`callIndex`, `actualMethod`
`IncompleteRealtimeScenarioError`	`unconsumedCalls`, `pendingExpectations`, `pendingFailures`
`ScriptedRealtimeScenarioCancelledError`	`pendingExpectations`
`RealtimeTransportNotClosedError`	`status`
`ScriptedRealtimeConnectionSupersededError`	`operation`

API 요약표

모델 단계 팩토리

팩토리	사용 시점
`modelResponse(response)`	명시적 응답 단계가 필요한 경우. 출력 배열을 직접 전달할 수도 있음
`modelResponder(callback)`	응답이 기록된 호출에 따라 달라지는 경우
`modelError(error, retryAdvice?)`	하나의 모델 호출이 실패해야 하는 경우
`modelStream(events)`	정확하게 정규화된 스트림이 필요한 경우
`modelStreamResponder(callback)`	정확한 스트림이 기록된 호출에 따라 달라지는 경우

일반 출력 헬퍼

헬퍼	생성 결과
`assistantMessage(text, options?)`	완료된 어시스턴트 텍스트 메시지
`functionCall(name, arguments, options)`	완료된 함수 호출. `options.callId`는 필수

추론, 호스티드 툴, 이미지, 오디오 또는 의도적으로 형식이 잘못된 프로토콜 사례에는 정규화된 출력 항목을 직접 전달합니다. 테스트 모듈이 두 번째 공급자 변환 계층이 되지 않도록 헬퍼 집합은 의도적으로 작게 유지됩니다.

샌드박스 세션 제어

멤버	목적
`scriptedSandboxSession(steps)`	스크립트에서 이름을 지정한 메서드를 포함하는 인메모리 `SandboxSession` 생성
`match(...args)`	예상 메서드 호출 하나의 형식화된 인수 검증
`result`	고정된 메서드 결과 반환
`respond(call)`	형식화되어 기록된 호출에서 결과 계산
`error`	주입된 실패를 발생시키거나 거부
`calls`	호출 순서대로 분리된 호출 스냅샷 검사
`assertComplete()`	워크플로가 소비하지 않은 샌드박스 단계 찾기

Realtime 제어

메서드	목적
`runScenario({ scenario, exercise, signal? })`	양쪽을 함께 실행. 테스트 전용 조정으로 인해 양쪽이 계속 대기할 수 있는 경우 `signal` 전달
`expectCall(method, matcher?)`	다음 송신 전송 호출의 일치 여부 확인
`emit(event, ...args)`	형식화된 수신 이벤트를 동기적으로 전달
`failNextCall(method, error)`	한 메서드의 다음 호출 실패 처리
`disconnect(error?)`	최종 원격 연결 해제 시뮬레이션
`assertComplete()`	일치하지 않은 호출, 기대 항목 또는 실패 찾기
`assertClosed()`	수명 주기 정리 확인

올바른 경계 선택

모델 공급자에 의존하지 않고 SDK 실행 루프, 도구, 핸드오프, 가드레일, 세션, 재시도 또는 스트리밍을 테스트해야 할 때 ScriptedModel을 사용합니다.

샌드박스 공급자를 시작하지 않고 SandboxAgent 기능과 오케스트레이션을 테스트해야 할 때 scriptedSandboxSession()을 ScriptedModel과 함께 사용합니다. 공급자 생성, 프로세스 실행, 파일 시스템 충실도, 지속성, 격리 검사는 실제 공급자를 대상으로 하는 통합 테스트에서 수행합니다.

WebRTC 또는 WebSocket 연결을 열지 않고 RealtimeSession 동작이나 RealtimeAgent 도구 및 핸드오프 오케스트레이션을 테스트해야 할 때 ScriptedRealtimeTransport를 사용합니다. 에이전트 비즈니스 로직은 RealtimeAgent에 연결된 도구, 핸드오프, 가드레일에 유지하고, 연결, 기록, 오디오, 인터럽션(중단 처리) 검증은 세션 및 전송 경계에서 유지합니다.

공급자 요청 변환, HTTP 또는 WebSocket 페이로드, 인증 헤더, 공급자별 스트리밍 청크, 샌드박스 공급자 수명 주기 및 격리를 테스트하는 데 이러한 더블을 사용하지 마세요. 모델 유선 테스트에는 모의 네트워크 전송과 함께 실제 모델 어댑터를 사용하고, 샌드박스 통합 테스트에는 실제 샌드박스 공급자를 사용합니다.

최종 체크리스트

정규화된 모델, 샌드박스 세션 또는 Realtime 전송 경계가 소유한 상호작용만 스크립트로 작성
비공개 실행기 상태 대신 중요한 요청 필드 검증
고정 응답 단계 우선 사용, 요청에 따라 달라지는 동작에만 응답자 사용
자동 스트리밍 우선 사용, 이벤트 수준 동작이 중요한 경우에만 정확한 스트림 사용
모델 테스트를 model.assertComplete()로 종료
샌드박스 테스트를 sandbox.assertComplete()로 종료
Realtime 테스트를 transport.assertComplete()와 transport.assertClosed()로 종료
오류 메시지를 파싱하는 대신 구조화된 오류 필드 검증
모의 네트워크 전송을 사용하는 실제 어댑터에서 공급자 유선 테스트 유지

범위 및 현재 제한 사항

테스트 모듈에서 의도적으로 제외된 항목

이러한 유틸리티는 SDK가 소유하는 정규화된 경계를 대체합니다. 모델이나 외부 공급자가 소유하는 동작을 입증하기 위한 것이 아닙니다.

모델 품질, instructions 준수, 도구 선택 품질은 실제 모델을 사용하는 평가 또는 테스트에서 다뤄야 합니다.
Responses 및 Chat Completions 요청 직렬화, 인증, 공급자 기본값, HTTP 동작, 공급자 스트림 청크에는 모의 또는 제어된 네트워크 전송과 함께 실제 모델 어댑터가 필요합니다.
샌드박스 시작, 프로세스 실행, 파일 시스템 충실도, 지속성, 리소스 제한, 보안 격리, 공급자 정리에는 실제 샌드박스 공급자를 사용하는 통합 테스트가 필요합니다.
Realtime 서버 동작, 원문 WebSocket 또는 WebRTC 프레임, 오디오 인코딩 및 재생, 인증, 네트워크 복구에는 실제 전송 또는 통합 환경이 필요합니다.

스크립트 기반 유틸리티는 이러한 계층을 에뮬레이션하지 않습니다. 여기에 공급자별 동작을 추가하면 워크플로 테스트가 공급자 프로토콜의 불완전한 두 번째 구현에 의존하게 됩니다.

현재 제공되지 않는 항목

현재 테스트 모듈은 다음을 제공하지 않습니다.

모든 정규화된 모델 출력 항목을 위한 편의 빌더. 일반적인 어시스턴트 메시지와 함수 호출에는 소규모 헬퍼를 사용하고 다른 정규화 항목은 직접 전달합니다.
고수준의 시뮬레이션된 Realtime 모델. Realtime 테스트는 송신 호출을 명시적으로 일치시키고 시나리오에 필요한 정규화된 수신 이벤트를 내보냅니다.
순서가 지정되지 않은 샌드박스 기대 항목. scriptedSandboxSession()은 하나의 전역 순서로 메서드 단계를 소비합니다.
테스트 실행기별 매처, 픽스처 또는 자동 해제. Node.js, Vitest, Jest 또는 다른 실행기의 검증 및 해제 API를 사용하고 제공된 완료 및 정리 검증을 명시적으로 호출합니다.
공급자 유선 테스트용 모의 전송. 해당 테스트에서는 실제 공급자 어댑터를 유지하고 네트워크 경계를 모의 처리합니다.

이는 현재 API 표면에 대한 설명이며, 향후 릴리스에서 생략된 기능을 추가하겠다는 약속이 아닙니다.