테스트

SDK는 에이전트 워크플로, 샌드박스 세션, Realtime 세션을 위한 결정론적이고 공급자에 종속되지 않는 테스트 대역을 제공합니다. 이들은 메모리에서 실행되며, 모델, 샌드박스 공급자 또는 Realtime API에 요청을 보내지 않고, SDK가 담당하는 정규화된 상호작용을 기록합니다.

필요한 레시피 찾기

목적	사용할 항목	바로 가기
고정된 최종 답변 반환	`ScriptedModel`과 `assistantMessage()`	고정 응답 반환
여러 턴의 도구 루프 실행	`functionCall()` 다음에 어시스턴트 응답 배치	도구 워크플로 테스트
요청에 따라 응답 선택	`modelResponder()`	요청에서 응답 도출
러너가 모델에 보낸 내용 검증	`calls`, `firstCall` 또는 `lastCall`	모델 호출 검사
스트리밍 실행 테스트	일반 응답 단계 또는 정확한 이벤트를 위한 `modelStream()`	스트리밍 테스트
오류 또는 재시도 결정 테스트	`modelError()`	모델 실패 주입
의도하지 않은 워크플로 변경 감지	정확한 단계와 `assertComplete()`	워크플로 드리프트 감지
샌드박스를 시작하지 않고 `SandboxAgent` 테스트	`scriptedSandboxSession()`과 `ScriptedModel`	샌드박스 에이전트 워크플로 테스트
샌드박스 호출 매칭 또는 결과 도출	샌드박스 단계의 `match` 또는 `respond`	샌드박스 단계 설정
샌드박스 작업 실패 주입	샌드박스 단계의 `error`	샌드박스 단계 설정
`RealtimeAgent` 함수 도구 테스트	`function_call`을 발생시키고 `sendFunctionCallOutput` 예상	음성 에이전트 도구 워크플로 테스트
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();
});

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

모델 호출 검사

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()에서 거부됩니다. 따라서 설정 오류는 워크플로 후반이 아니라 테스트 정의에 가까운 지점에서 실패로 드러납니다.

샌드박스 에이전트 레시피

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

ScriptedModel과 scriptedSandboxSession()을 함께 사용하면 Docker 컨테이너나 원격 샌드박스를 만들지 않고도 실제 SandboxAgent 런타임을 테스트할 수 있습니다. 모델 스크립트는 기능 도구를 선택하고, 샌드박스 스크립트는 해당 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?`	단계의 형식이 잘못되었거나 지원되지 않는 메서드를 지정함

음성 에이전트 및 세션 레시피

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

실제 함수 도구를 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은 해당 도구를 사용하겠다는 모델의 결정을 나타냅니다. 실제 모델이 에이전트의 지시를 따르거나 도구를 선택한다는 것을 입증하지는 않습니다. 모델 결정 품질은 평가 또는 통합 테스트로 검증하세요.

실행 코드는 세션을 닫기 전에 비동기 도구 출력을 기다리므로, 이 코드 예제는 대기 시간을 제한하기 위해 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()는 대기 중인 기대 조건을 거부하고, 시나리오는 해당 실패를 활성 상태의 테스트 전용 동기화 게이트로 전달합니다. 그러면 실행 코드는 두 콜백을 모두 차단된 상태로 두는 대신 세션 정리를 수행합니다.

expectCall(method, matcher?)은 메서드별로 타입이 지정되며 하나의 전역 순서로 호출을 매칭합니다. 매처는 어설션을 포함할 수 있고, false를 반환하여 호출을 거부할 수 있습니다. 기대 조건이 등록되기 직전에 호출이 발생하면 전송은 다음 기대 조건을 위해 해당 호출을 보관합니다.

runScenario()는 어느 한쪽이 실패하거나, AbortSignal이 시나리오를 취소하거나, 기대 조건이 완료되지 않은 상태에서 실행 코드가 종료되면 종료됩니다. 기대 조건이 남아 있는 상태에서 실행 코드가 종료되면 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()로 마무리하세요.
오류 메시지를 파싱하는 대신 구조화된 오류 필드를 검증하세요.
공급자 와이어 테스트에는 실제 어댑터와 모킹된 네트워크 전송을 사용하세요.

에이전트 실행은 ScriptedModel이 테스트하는 실행 루프를 설명합니다.
모델은 Model 경계와 공급자 어댑터를 설명합니다.
빠른 시작은 샌드박스 기능, 세션, 공급자 수명 주기를 설명합니다.
음성 에이전트 개요는 RealtimeSession을 소개합니다.
전송 방식은 프로덕션 전송과 전송 이벤트를 다룹니다.

범위 및 현재 제한 사항

테스트 모듈에서 의도적으로 제외된 영역

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

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

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

현재 미제공 항목

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

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

이는 현재 API 인터페이스에 대한 설명이며, 향후 릴리스에서 누락된 기능을 추가하겠다는 약속이 아닙니다.