핵심 결론부터 확인하세요

Mock Testing은 선택이 아닌 필수입니다. AI API 호출 비용은 GPT-4.1 기준 $8/1M 토큰, Claude Sonnet 4.5 기준 $15/1M 토큰에 달합니다. 개발 단계에서 실제 API를 호출하면 하루에 수십 달러가 순식간에 사라질 수 있습니다. 이 튜토리얼에서는 HolySheep AI를 포함한 모든 AI API에서 안전하게 Mock Testing을 구현하는 방법을 단계별로 설명드리겠습니다.

저는 3년 넘게 AI API 통합 프로젝트를 진행하면서 수천 달러의 비용을 절감한 경험이 있습니다. 이 글에서 소개하는 방법들을 적용하면 개발 환경에서 실제 API 비용을 100% 절감하면서도 프로덕션 환경과 동일한 코드 구조를 유지할 수 있습니다.

왜 AI API Mock Testing이 중요한가?

AI API 게이트웨이 비교표

서비스 결제 방식 GPT-4.1 Claude Sonnet 4.5 Gemini 2.5 Flash DeepSeek V3.2 적합한 팀
HolySheep AI 로컬 결제 (해외 카드 불필요) $8/MTok $15/MTok $2.50/MTok $0.42/MTok 글로벌 서비스 필요 팀, 스타트업
OpenAI 공식 해외 신용카드 필수 $8/MTok - - - OpenAI 전용 프로젝트
Anthropic 공식 해외 신용카드 필수 - $15/MTok - - Claude 전용 프로젝트
Google Vertex AI 해외 신용카드 필수 - - $2.50/MTok - Enterprise GCP 사용자

HolySheep AI는 단일 API 키로 모든 주요 모델을 통합 지원하며, 해외 신용카드 없이 로컬 결제가 가능합니다. 개발 단계에서 Mock Testing 후 동일 코드베이스로 실제 API切り替え가无缝합니다.

Python 기반 Mock Testing 구현

1. 기본 Mock 클라이언트 클래스

import json
import time
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, field

@dataclass
class MockMessage:
    role: str
    content: str

@dataclass
class MockChoice:
    index: int
    message: MockMessage
    finish_reason: str

@dataclass
class MockUsage:
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int

@dataclass
class MockResponse:
    id: str
    object: str = "chat.completion"
    created: int = field(default_factory=lambda: int(time.time()))
    model: str = "mock-model"
    choices: List[MockChoice] = field(default_factory=list)
    usage: Optional[MockUsage] = None

class AIAPIMockClient:
    """AI API Mock Testing을 위한 범용 클라이언트"""
    
    def __init__(self, base_url: str = "https://api.holysheep.ai/v1"):
        self.base_url = base_url
        self.response_delay_ms: int = 0
        self.custom_responses: Dict[str, str] = {}
        self.error_rate: float = 0.0
        self._call_count: int = 0
    
    def set_response_delay(self, delay_ms: int) -> "AIAPIMockClient":
        """응답 지연 시간 설정 (밀리초 단위)"""
        self.response_delay_ms = delay_ms
        return self
    
    def set_custom_response(self, prompt_pattern: str, response: str) -> "AIAPIMockClient":
        """특정 프롬프트 패턴에 대한 커스텀 응답 설정"""
        self.custom_responses[prompt_pattern] = response
        return self
    
    def set_error_rate(self, rate: float) -> "AIAPIMockClient":
        """에러 발생 확률 설정 (0.0 ~ 1.0)"""
        if not 0.0 <= rate <= 1.0:
            raise ValueError("error_rate must be between 0.0 and 1.0")
        self.error_rate = rate
        return self
    
    def chat_completions(
        self,
        model: str,
        messages: List[Dict[str, str]],
        temperature: float = 0.7,
        max_tokens: int = 1000,
        **kwargs
    ) -> Dict[str, Any]:
        """
        Mock Chat Completions API
        
        실제 HolySheep AI API 응답 형식과 100% 호환
        """
        self._call_count += 1
        
        # 응답 지연 시뮬레이션
        if self.response_delay_ms > 0:
            time.sleep(self.response_delay_ms / 1000.0)
        
        # 마지막 사용자 메시지 추출
        last_message = messages[-1]["content"] if messages else ""
        
        # 커스텀 응답 확인
        response_content = self._get_response_content(last_message, model)
        
        # Mock 응답 생성
        response = {
            "id": f"mock-{self._call_count}-{int(time.time())}",
            "object": "chat.completion",
            "created": int(time.time()),
            "model": model,
            "choices": [{
                "index": 0,
                "message": {
                    "role": "assistant",
                    "content": response_content
                },
                "finish_reason": "stop"
            }],
            "usage": {
                "prompt_tokens": len(str(messages)) // 4,
                "completion_tokens": len(response_content) // 4,
                "total_tokens": (len(str(messages)) + len(response_content)) // 4
            }
        }
        
        return response
    
    def _get_response_content(self, prompt: str, model: str) -> str:
        """프롬프트에 맞는 응답 콘텐츠 반환"""
        for pattern, response in self.custom_responses.items():
            if pattern.lower() in prompt.lower():
                return response
        
        # 기본 Mock 응답
        return f"[Mock Response for: {prompt[:50]}... | Model: {model}]"

사용 예시

if __name__ == "__main__": client = AIAPIMockClient() # 150ms 지연 설정 (실제 API 평균 지연 시간 시뮬레이션) client.set_response_delay(150) # 커스텀 응답 설정 client.set_custom_response(" translate ", "번역 결과: Hello World") client.set_custom_response(" summarize ", "요약: 이 문장의 핵심 내용은...") # API 호출 response = client.chat_completions( model="gpt-4.1", messages=[{"role": "user", "content": "Please translate to Korean: Hello World"}] ) print(f"Response ID: {response['id']}") print(f"Content: {response['choices'][0]['message']['content']}") print(f"Total Tokens: {response['usage']['total_tokens']}")

2. 환경별 Mock/실제 API 자동 전환

import os
from typing import Optional
from abc import ABC, abstractmethod

class BaseAIClient(ABC):
    """AI API 클라이언트 추상 기본 클래스"""
    
    @abstractmethod
    def chat_completions(self, model: str, messages: list, **kwargs):
        pass

class MockAIClient(BaseAIClient):
    """Mock 환경용 클라이언트 - HolySheep AI 포맷 호환"""
    
    def __init__(self):
        from collections import defaultdict
        self._responses = defaultdict(lambda: {
            "id": "mock-response",
            "object": "chat.completion",
            "created": 1234567890,
            "model": "mock",
            "choices": [{"index": 0, "message": {"role": "assistant", "content": "Mock response"}, "finish_reason": "stop"}],
            "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}
        })
    
    def add_response(self, model: str, content: str) -> None:
        """모델별 고정 응답 추가"""
        import time
        self._responses[model] = {
            "id": f"mock-{model}-{int(time.time() * 1000)}",
            "object": "chat.completion",
            "created": int(time.time()),
            "model": model,
            "choices": [{
                "index": 0,
                "message": {"role": "assistant", "content": content},
                "finish_reason": "stop"
            }],
            "usage": {"prompt_tokens": 20, "completion_tokens": len(content.split()), "total_tokens": 20 + len(content.split())}
        }
    
    def chat_completions(self, model: str, messages: list, **kwargs):
        return self._responses[model]

class HolySheepAIClient(BaseAIClient):
    """HolySheep AI 실제 API 클라이언트"""
    
    BASE_URL = "https://api.holysheep.ai/v1"
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self._session = None
    
    def _get_session(self):
        if self._session is None:
            import requests
            self._session = requests.Session()
            self._session.headers.update({
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            })
        return self._session
    
    def chat_completions(self, model: str, messages: list, **kwargs):
        import requests
        import time
        
        start_time = time.time()
        
        try:
            response = self._get_session().post(
                f"{self.BASE_URL}/chat/completions",
                json={
                    "model": model,
                    "messages": messages,
                    **{k: v for k, v in kwargs.items() if v is not None}
                },
                timeout=30
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            latency_ms = int((time.time() - start_time) * 1000)
            raise RuntimeError(f"HolySheep AI API Error (Latency: {latency_ms}ms): {str(e)}")

class AIAgentFactory:
    """환경에 따라 Mock/실제 API 자동 선택"""
    
    @staticmethod
    def create_client(env: Optional[str] = None) -> BaseAIClient:
        """
        환경에 맞는 AI 클라이언트 생성
        
        Args:
            env: 'development', 'testing', 'production' 또는 None (자동 감지)
        
        Returns:
            BaseAIClient 인스턴스
        """
        if env is None:
            env = os.getenv("AI_API_ENV", "development")
        
        if env in ("development", "testing", "local", "ci"):
            print(f"[Mock Mode] {env} 환경에서 Mock API 사용")
            return MockAIClient()
        elif env == "production":
            api_key = os.getenv("HOLYSHEEP_API_KEY")
            if not api_key:
                raise ValueError("HOLYSHEEP_API_KEY 환경 변수가 설정되지 않았습니다.")
            print(f"[Production Mode] HolySheep AI API 사용")
            return HolySheepAIClient(api_key)
        else:
            raise ValueError(f"알 수 없는 환경: {env}")

테스트 코드

if __name__ == "__main__": # Mock 모드 테스트 mock_client = AIAgentFactory.create_client("testing") mock_client.add_response("gpt-4.1", "이것은 Mock 응답입니다.") result = mock_client.chat_completions( model="gpt-4.1", messages=[{"role": "user", "content": "테스트 메시지"}] ) print(f"Mock Result: {result['choices'][0]['message']['content']}") # 환경 변수 설정 시뮬레이션 os.environ["AI_API_ENV"] = "production" os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY" # 실제 API 전환 테스트 (실제 호출은 주석 처리) # real_client = AIAgentFactory.create_client("production") # result = real_client.chat_completions(model="gpt-4.1", messages=[...])

Node.js/TypeScript Mock Testing 구현

/**
 * AI API Mock Testing - TypeScript 구현
 * HolySheep AI API 응답 포맷 100% 호환
 */

interface AIMessage {
  role: 'system' | 'user' | 'assistant';
  content: string;
}

interface ChatCompletionRequest {
  model: string;
  messages: AIMessage[];
  temperature?: number;
  max_tokens?: number;
  top_p?: number;
  frequency_penalty?: number;
  presence_penalty?: number;
  stop?: string | string[];
}

interface ChatCompletionChoice {
  index: number;
  message: AIMessage;
  finish_reason: string;
}

interface Usage {
  prompt_tokens: number;
  completion_tokens: number;
  total_tokens: number;
}

interface ChatCompletionResponse {
  id: string;
  object: 'chat.completion';
  created: number;
  model: string;
  choices: ChatCompletionChoice[];
  usage: Usage;
}

type AIAuthFunction = (
  req: ChatCompletionRequest
) => ChatCompletionResponse | Promise;

class MockAIClient {
  private responses: Map = new Map();
  private delayMs: number = 0;
  private callCount: number = 0;
  private authFunctions: AIAuthFunction[] = [];

  setDelay(delayMs: number): MockAIClient {
    this.delayMs = delayMs;
    return this;
  }

  setResponse(model: string, response: string): MockAIClient {
    this.responses.set(model, response);
    return this;
  }

  addAuthFunction(fn: AIAuthFunction): MockAIClient {
    this.authFunctions.push(fn);
    return this;
  }

  async chatCompletions(
    request: ChatCompletionRequest
  ): Promise {
    this.callCount++;
    
    // 지연 시간 시뮬레이션
    if (this.delayMs > 0) {
      await new Promise(resolve => setTimeout(resolve, this.delayMs));
    }

    // 커스텀 인증 함수 우선 실행
    for (const authFn of this.authFunctions) {
      const result = await authFn(request);
      if (result) {
        return {
          ...result,
          id: mock-${this.callCount}-${Date.now()},
          created: Math.floor(Date.now() / 1000)
        };
      }
    }

    // 기본 Mock 응답 생성
    const lastMessage = request.messages[request.messages.length - 1]?.content || '';
    const mockContent = this.responses.get(request.model) || 
      [Mock] Request: ${lastMessage.substring(0, 50)}...;

    return {
      id: mock-${this.callCount}-${Date.now()},
      object: 'chat.completion',
      created: Math.floor(Date.now() / 1000),
      model: request.model,
      choices: [{
        index: 0,
        message: {
          role: 'assistant',
          content: mockContent
        },
        finish_reason: 'stop'
      }],
      usage: {
        prompt_tokens: Math.floor(lastMessage.length / 4),
        completion_tokens: Math.floor(mockContent.length / 4),
        total_tokens: Math.floor((lastMessage.length + mockContent.length) / 4)
      }
    };
  }

  getCallCount(): number {
    return this.callCount;
  }

  reset(): void {
    this.callCount = 0;
    this.responses.clear();
    this.authFunctions = [];
  }
}

// HolySheep AI 실제 API 클라이언트
class HolySheepAIClient {
  private baseUrl = 'https://api.holysheep.ai/v1';
  private apiKey: string;

  constructor(apiKey: string) {
    this.apiKey = apiKey;
  }

  async chatCompletions(
    request: ChatCompletionRequest,
    timeoutMs: number = 30000
  ): Promise {
    const startTime = performance.now();
    
    const response = await fetch(${this.baseUrl}/chat/completions, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Authorization': Bearer ${this.apiKey}
      },
      body: JSON.stringify(request),
      signal: AbortSignal.timeout(timeoutMs)
    });

    const latencyMs = Math.round(performance.now() - startTime);
    
    if (!response.ok) {
      const error = await response.text();
      throw new Error(
        HolySheep AI Error [${response.status}] [Latency: ${latencyMs}ms]: ${error}
      );
    }

    console.log([HolySheep AI] Latency: ${latencyMs}ms | Model: ${request.model});
    return response.json();
  }
}

// 환경별 클라이언트 팩토리
function createAIClient(env?: string): MockAIClient | HolySheepAIClient {
  const environment = env || process.env.AI_API_ENV || 'development';
  
  switch (environment) {
    case 'development':
    case 'testing':
    case 'local':
    case 'ci':
      console.log([Mock Mode] ${environment} 환경);
      return new MockAIClient();
    case 'production':
      const apiKey = process.env.HOLYSHEEP_API_KEY;
      if (!apiKey) {
        throw new Error('HOLYSHEEP_API_KEY 환경 변수가 필요합니다.');
      }
      console.log('[Production Mode] HolySheep AI API 사용');
      return new HolySheepAIClient(apiKey);
    default:
      throw new Error(알 수 없는 환경: ${environment});
  }
}

// 사용 예시
async function main() {
  const client = createAIClient('testing');
  
  if (client instanceof MockAIClient) {
    client
      .setDelay(120) // 120ms 지연 시뮬레이션
      .setResponse('gpt-4.1', 'Mock: 안녕하세요! 무엇을 도와드릴까요?')
      .setResponse('claude-sonnet-4.5', 'Mock: Hello! How can I assist you?');
  }

  const response = await client.chatCompletions({
    model: 'gpt-4.1',
    messages: [
      { role: 'user', content: '인사해 주세요' }
    ],
    temperature: 0.7,
    max_tokens: 500
  });

  console.log('Response ID:', response.id);
  console.log('Content:', response.choices[0].message.content);
  console.log('Tokens:', response.usage.total_tokens);
}

main().catch(console.error);

export { MockAIClient, HolySheepAIClient, createAIClient };
export type { ChatCompletionRequest, ChatCompletionResponse };

에지 케이스 및 시나리오별 Mock 전략

"""
고급 Mock Testing 시나리오
에러 처리, 레이트 리밋, 타임아웃 등 다양한 상황 시뮬레이션
"""

import random
import time
from typing import Callable, Optional
from dataclasses import dataclass
from enum import Enum

class MockErrorType(Enum):
    RATE_LIMIT = "rate_limit"
    TIMEOUT = "timeout"
    SERVER_ERROR = "server_error"
    AUTH_ERROR = "auth_error"
    INVALID_REQUEST = "invalid_request"
    NETWORK_ERROR = "network_error"

@dataclass
class MockError(Exception):
    error_type: MockErrorType
    message: str
    status_code: int = 429
    
    def to_openai_format(self) -> dict:
        return {
            "error": {
                "message": self.message,
                "type": self.error_type.value,
                "code": self.error_type.value,
                "status": self.status_code
            }
        }

class AdvancedMockClient:
    """고급 Mock Testing 클라이언트 - 다양한 에러 시나리오 지원"""
    
    def __init__(self):
        self.call_history: list = []
        self.error_config: dict = {}
        self.rate_limit_remaining: int = 100
        self.rate_limit_reset: float = 0
        self.latency_ms: int = 0
    
    def configure_error_scenario(
        self,
        error_type: MockErrorType,
        probability: float,
        message: Optional[str] = None
    ) -> "AdvancedMockClient":
        """에러 발생 시나리오 설정"""
        if not 0.0 <= probability <= 1.0:
            raise ValueError("probability must be between 0.0 and 1.0")
        
        self.error_config[error_type] = {
            "probability": probability,
            "message": message or f"Mock {error_type.value} occurred"
        }
        return self
    
    def set_rate_limit(self, remaining: int, reset_timestamp: float) -> "AdvancedMockClient":
        """레이트 리밋 설정"""
        self.rate_limit_remaining = remaining
        self.rate_limit_reset = reset_timestamp
        return self
    
    def set_latency(self, ms: int) -> "AdvancedMockClient":
        """평균 지연 시간 설정 (실제 API 지연 시뮬레이션)"""
        self.latency_ms = ms
        return self
    
    def add_to_history(self, request: dict, response: dict) -> None:
        """호출 이력 기록"""
        self.call_history.append({
            "timestamp": time.time(),
            "request": request,
            "response": response
        })
    
    def chat_completions(self, model: str, messages: list, **kwargs):
        """Mock Chat Completions with advanced scenarios"""
        
        # 지연 시간 적용
        if self.latency_ms > 0:
            time.sleep(self.latency_ms / 1000.0)
        
        # 요청 기록
        request = {"model": model, "messages": messages, **kwargs}
        
        # 레이트 리밋 체크
        if self.rate_limit_remaining <= 0:
            error = MockError(
                error_type=MockErrorType.RATE_LIMIT,
                message="Rate limit exceeded. Try again later.",
                status_code=429
            )
            self.add_to_history(request, error.to_openai_format())
            raise error
        
        # 에러 시나리오 체크
        for error_type, config in self.error_config.items():
            if random.random() < config["probability"]:
                status_codes = {
                    MockErrorType.RATE_LIMIT: 429,
                    MockErrorType.TIMEOUT: 408,
                    MockErrorType.SERVER_ERROR: 500,
                    MockErrorType.AUTH_ERROR: 401,
                    MockErrorType.INVALID_REQUEST: 400,
                    MockErrorType.NETWORK_ERROR: 503
                }
                error = MockError(
                    error_type=error_type,
                    message=config["message"],
                    status_code=status_codes.get(error_type, 500)
                )
                self.add_to_history(request, error.to_openai_format())
                raise error
        
        # 정상 응답 생성
        self.rate_limit_remaining -= 1
        last_message = messages[-1]["content"] if messages else ""
        
        response = {
            "id": f"mock-adv-{len(self.call_history)}",
            "object": "chat.completion",
            "created": int(time.time()),
            "model": model,
            "choices": [{
                "index": 0,
                "message": {
                    "role": "assistant",
                    "content": f"[Advanced Mock] Processed: {last_message[:30]}..."
                },
                "finish_reason": "stop"
            }],
            "usage": {
                "prompt_tokens": 10,
                "completion_tokens": 5,
                "total_tokens": 15
            },
            "x-ratelimit-remaining": str(self.rate_limit_remaining),
            "x-ratelimit-reset": str(int(self.rate_limit_reset or time.time() + 3600))
        }
        
        self.add_to_history(request, response)
        return response
    
    def get_history(self) -> list:
        """호출 이력 반환"""
        return self.call_history
    
    def get_statistics(self) -> dict:
        """통계 정보 반환"""
        return {
            "total_calls": len(self.call_history),
            "rate_limit_remaining": self.rate_limit_remaining,
            "configured_errors": list(self.error_config.keys())
        }


사용 예시

if __name__ == "__main__": client = AdvancedMockClient() # 시나리오 1: 10% 확률로 레이트 리밋 발생 client.configure_error_scenario( MockErrorType.RATE_LIMIT, probability=0.1, message="Rate limit exceeded after 100 requests" ) # 시나리오 2: 5% 확률로 서버 에러 발생 client.configure_error_scenario( MockErrorType.SERVER_ERROR, probability=0.05, message="Internal server error during processing" ) # 지연 시간: 평균 200ms (실제 GPT-4 평균 지연 시간) client.set_latency(200) # 레이트 리밋: 5회 호출 후 제한 client.set_rate_limit(remaining=5, reset_timestamp=time.time() + 3600) # 테스트 실행 for i in range(10): try: result = client.chat_completions( model="gpt-4.1", messages=[{"role": "user", "content": f"Test message {i}"}] ) print(f"[{i+1}] Success: {result['choices'][0]['message']['content'][:50]}") print(f" Rate Limit Remaining: {result.get('x-ratelimit-remaining', 'N/A')}") except MockError as e: print(f"[{i+1}] Error ({e.error_type.value}): {e.message}") # 통계 출력 stats = client.get_statistics() print(f"\n=== Statistics ===") print(f"Total Calls: {stats['total_calls']}") print(f"Rate Limit Remaining: {stats['rate_limit_remaining']}") print(f"Configured Errors: {stats['configured_errors']}")

자주 발생하는 오류와 해결책

오류 1: Mock 응답 형식이 실제 API와 호환되지 않음

# ❌ 잘못된 접근 - 불완전한 Mock 응답
def bad_mock():
    return {"result": "Hello World"}  # OpenAI 형식 아님

✅ 올바른 접근 - HolySheep AI / OpenAI 호환 형식

def correct_mock(): return { "id": "chatcmpl-123", "object": "chat.completion", "created": 1677652288, "model": "gpt-4.1", "choices": [{ "index": 0, "message": { "role": "assistant", "content": "Hello World" }, "finish_reason": "stop" }], "usage": { "prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15 } }

Stream 응답 Mock도 동일하게 적용

def correct_stream_mock(): return { "id": "chatcmpl-123", "object": "chat.completion.chunk", "created": 1677652288, "model": "gpt-4.1", "choices": [{ "index": 0, "delta": { "content": "Hello" }, "finish_reason": None }] }

오류 2: 환경 변수 누락으로 프로덕션에서 Mock 모드 유지

import os
import sys

❌ 위험한 접근 - 기본값이 Mock이면 프로덕션에서도 Mock 사용 가능

def bad_client(): env = os.getenv("AI_ENV", "mock") # 잘못된 기본값 if env == "mock": return MockClient() return RealClient()

✅ 안전한 접근 - 명시적 환경 설정 강제

def safe_client(): env = os.getenv("AI_API_ENV") if env is None: print("⚠️ AI_API_ENV가 설정되지 않았습니다.") print(" development: Mock API 사용") print(" production: HolySheep AI API 사용") sys.exit(1) clients = { "development": MockClient, "testing": MockClient, "production": lambda: HolySheepAIClient(os.environ["HOLYSHEEP_API_KEY"]) } client_class = clients.get(env) if client_class is None: raise ValueError(f"알 수 없는 환경: {env}. 허용값: {list(clients.keys())}") return client_class()

.env.example 파일에 반드시 포함해야 할 내용

DOTENV_CONTENT = """

.env.example - 복사하여 .env로 이름 변경 후填写

AI_API_ENV=development # development, testing, production

HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY # production에서만 필요

"""

오류 3: 비동기 코드에서 Mock 지연 시간 누락

import asyncio
import time

❌ 비동기 함수에서 time.sleep() 블로킹 발생

async def bad_async_mock(): time.sleep(1) # 이벤트 루프 블로킹! return {"content": "result"}

✅ asyncio.sleep() 사용 - 논블로킹 지연

async def good_async_mock(): await asyncio.sleep(1) # 다른 태스크 실행 가능 return {"content": "result"}

실제 테스트

async def test_mock_performance(): client = AsyncMockClient() start = time.time() tasks = [client.chat_completions(f"request-{i}") for i in range(10)] results = await asyncio.gather(*tasks) # 동시 실행 elapsed = time.time() - start # 10개 요청이 1초(지연) 만에 완료되어야 함 assert elapsed < 2.0, f"너무 오래 걸림: {elapsed:.2f}초" print(f"10개 동시 요청 완료: {elapsed*1000:.0f}ms")

async Mock 클라이언트 예시

class AsyncMockClient: async def chat_completions(self, request_id: str): await asyncio.sleep(0.1) # 100ms 지연 return { "id": request_id, "choices": [{"message": {"content": f"Response for {request_id}"}}] } if __name__ == "__main__": asyncio.run(test_mock_performance())

오류 4: 레이트 리밋 시뮬레이션 누락

import time
from collections import deque

✅ 레이트 리밋을 포함한 Mock 클라이언트

class RateLimitedMockClient: def __init__(self, rpm: int = 60, tpm: int = 100000): self.rpm = rpm self.tpm = tpm self.request_timestamps: deque = deque(maxlen=rpm) self.total_tokens_used: int = 0 self.window_start: float = time.time() def _check_rate_limit(self, estimated_tokens: int): current_time = time.time() # RPM 체크 (1분 윈도우) while self.request_timestamps and \ current_time - self.request_timestamps[0] > 60: self.request_timestamps.popleft() if len(self.request_timestamps) >= self.rpm: wait_time = 60 - (current_time - self.request_timestamps[0]) raise Exception(f"RPM_limit_exceeded: {len(self.request_timestamps)}/{self.rpm}. Wait {wait_time:.1f}s") # TPM 체크 (1분 윈도우) if self.total_tokens_used + estimated_tokens > self.tpm: if current_time - self.window_start >= 60: self.total_tokens_used = 0 self.window_start = current_time else: raise Exception(f"TPM_limit_exceeded: {self.total_tokens_used}/{self.tpm}") self.request_timestamps.append(current_time) self.total_tokens_used += estimated_tokens def chat_completions(self, messages: list, **kwargs): estimated_tokens = sum(len(str(m)) for m in messages) // 4 self._check_rate_limit(estimated_tokens) return { "id": f"mock-{len(self.request_timestamps)}", "usage": { "prompt_tokens": estimated_tokens, "completion_tokens": 20, "total_tokens": estimated_tokens + 20 } }

테스트

client = RateLimitedMockClient(rpm=5) # 분당 5회 제한 for i in range(7): try: result = client.chat_completions([{"role": "user", "content":