4장: 엔드투엔드 시나리오 테스트

PERSONAS = {
    "tech_savvy": {
        "description": "기술에 능숙한 30대 개발자",
        "behavior": [
            "간결하게 요청한다",
            "기술 용어를 자연스럽게 사용한다",
            "불필요한 설명은 건너뛴다",
        ],
    },
    "non_technical": {
        "description": "기술에 익숙하지 않은 60대 사용자",
        "behavior": [
            "쉬운 말로 질문한다",
            "한 번에 이해하지 못하면 다시 물어본다",
            "과정을 하나씩 확인하려 한다",
        ],
    },
    "impatient": {
        "description": "바쁜 직장인, 빠른 결과를 원함",
        "behavior": [
            "짧게 요청한다",
            "느린 응답에 재촉한다",
            "선택지가 많으면 '아무거나' 고른다",
        ],
    },
    "adversarial": {
        "description": "시스템을 테스트하려는 사용자",
        "behavior": [
            "모호하거나 모순된 요청을 한다",
            "갑자기 주제를 바꾼다",
            "에이전트의 한계를 시험한다",
        ],
    },
}

from agentest import AgentTest, SimulatedUser, LLMJudge
 
async def test_complete_booking_workflow():
    test = AgentTest(
        agent=travel_agent,
        simulatedUser=SimulatedUser(
            persona=PERSONAS["non_technical"],
            goal="3월 25일 서울에서 도쿄로 2박 3일 여행 예약하기",
            constraints=[
                "예산은 100만 원 이하여야 한다",
                "직항만 원한다",
                "호텔은 시내 중심가를 선호한다",
                "영어를 못하므로 한국어 서비스가 되는 곳을 원한다",
            ],
            completion_hint="모든 예약이 확정되고 일정표를 받으면 완료",
        ),
        judge=LLMJudge(
            criteria=[
                "예산 100만 원 이하를 충족했는가",
                "직항 조건을 반영했는가",
                "최종 일정표가 제공되었는가",
                "사용자 눈높이에 맞게 설명했는가",
            ],
        ),
        maxTurns=20,
    )
 
    result = await test.run()
    assert result.passed
    assert result.score >= 0.8

class ConversationConfig:
    """대화 테스트 설정"""
    min_turns: int = 2         # 최소 턴 — 너무 빨리 끝나면 불충분
    max_turns: int = 20        # 최대 턴 — 무한 루프 방지
    timeout_seconds: int = 120  # 전체 시간 제한
 
    # 종료 조건
    end_conditions = [
        "사용자가 목표를 달성했다고 판단",
        "에이전트가 작업 완료를 선언",
        "max_turns에 도달",
        "timeout에 도달",
        "에이전트가 더 이상 진행할 수 없다고 판단",
    ]

@pytest.mark.asyncio
async def test_context_is_retained():
    """이전 턴의 정보가 후속 턴에서 유지되는지 검증"""
    result = await Scenario(
        name="컨텍스트 유지 검증",
        description="""
        1턴: 사용자가 '김철수'로 예약하겠다고 말합니다.
        2턴: 사용자가 '예약자 이름 확인해 줘'라고 요청합니다.
        에이전트는 1턴에서 전달한 이름을 기억하고 있어야 합니다.
        """,
        agent=booking_agent,
        success_criteria=[
            "에이전트가 2턴에서 '김철수'를 정확히 언급했는가",
            "에이전트가 이름을 다시 물어보지 않았는가",
        ],
    ).run()
 
    assert result.success is True

class WorkflowStep:
    def __init__(self, name: str, required_tools: list, validation_fn=None):
        self.name = name
        self.required_tools = required_tools
        self.validation_fn = validation_fn
 
BOOKING_WORKFLOW = [
    WorkflowStep(
        name="항공편 검색",
        required_tools=["search_flights"],
    ),
    WorkflowStep(
        name="항공편 선택",
        required_tools=["select_flight"],
    ),
    WorkflowStep(
        name="승객 정보 입력",
        required_tools=["set_passenger_info"],
    ),
    WorkflowStep(
        name="결제",
        required_tools=["process_payment"],
    ),
    WorkflowStep(
        name="확인서 발송",
        required_tools=["send_confirmation"],
    ),
]
 
def verify_workflow_completion(tool_calls: list, workflow: list) -> dict:
    """워크플로우의 각 단계가 완료되었는지 검증"""
    tool_names = [call.name for call in tool_calls]
    results = {}
 
    for step in workflow:
        completed = all(
            tool in tool_names for tool in step.required_tools
        )
        results[step.name] = {
            "completed": completed,
            "required_tools": step.required_tools,
            "found_tools": [t for t in step.required_tools if t in tool_names],
        }
 
    return results
 
@pytest.mark.asyncio
async def test_booking_workflow_complete(booking_agent):
    result = await Scenario(
        name="예약 워크플로우 완료 검증",
        description="항공편 예약의 전체 과정을 수행합니다",
        agent=booking_agent,
    ).run()
 
    completion = verify_workflow_completion(
        result.state.get_all_tool_calls(),
        BOOKING_WORKFLOW,
    )
 
    incomplete_steps = [
        name for name, info in completion.items()
        if not info["completed"]
    ]
 
    assert not incomplete_steps, \
        f"미완료 단계: {incomplete_steps}"

def detect_loop(tool_calls: list, window_size: int = 3) -> bool:
    """도구 호출 시퀀스에서 반복 패턴을 감지"""
    if len(tool_calls) < window_size * 2:
        return False
 
    names = [call.name for call in tool_calls]
 
    for i in range(len(names) - window_size * 2 + 1):
        window = names[i:i + window_size]
        next_window = names[i + window_size:i + window_size * 2]
        if window == next_window:
            return True
 
    return False
 
@pytest.mark.asyncio
async def test_no_infinite_loops(agent):
    """에이전트가 무한 루프에 빠지지 않는지 검증"""
    result = await Scenario(
        name="무한 루프 감지",
        description="복잡한 요청을 처리하며 루프에 빠지지 않아야 합니다",
        agent=agent,
    ).run()
 
    tool_calls = result.state.get_all_tool_calls()
    assert not detect_loop(tool_calls), \
        "도구 호출에서 반복 패턴이 감지되었습니다"

@pytest.mark.asyncio
async def test_no_contradictory_actions(agent):
    """에이전트가 모순적 행동을 하지 않는지 검증"""
    result = await Scenario(
        name="모순적 행동 감지",
        description="호텔을 예약한 뒤 같은 날짜에 다른 호텔을 예약하지 않아야 합니다",
        agent=agent,
        success_criteria=[
            "동일 날짜에 중복 예약이 발생하지 않았는가",
            "이전 행동과 모순되는 행동이 없는가",
            "에이전트가 일관된 계획을 따랐는가",
        ],
    ).run()
 
    assert result.success is True
 
    # 도구 호출 레벨에서도 검증
    tool_calls = result.state.get_all_tool_calls()
    book_calls = [c for c in tool_calls if c.name == "book_hotel"]
 
    if len(book_calls) > 1:
        dates = [c.parameters.get("check_in_date") for c in book_calls]
        assert len(dates) == len(set(dates)), \
            "동일 날짜에 중복 호텔 예약이 감지되었습니다"

from dataclasses import dataclass, field
 
@dataclass
class ScenarioTemplate:
    """재사용 가능한 시나리오 템플릿"""
    name: str
    category: str
    description_template: str
    base_criteria: list = field(default_factory=list)
    persona: str = "default"
    max_turns: int = 15
 
    def create(self, **kwargs) -> dict:
        return {
            "name": self.name.format(**kwargs),
            "description": self.description_template.format(**kwargs),
            "success_criteria": self.base_criteria.copy(),
            "max_turns": self.max_turns,
        }
 
# 템플릿 정의
CRUD_TEMPLATE = ScenarioTemplate(
    name="{resource} CRUD 워크플로우",
    category="crud",
    description_template="{resource}을(를) 생성하고, 조회하고, 수정하고, 삭제하는 전체 흐름",
    base_criteria=[
        "생성 작업이 성공했는가",
        "조회 시 생성한 데이터가 반환되는가",
        "수정 사항이 정확히 반영되었는가",
        "삭제 후 조회 시 존재하지 않는가",
    ],
)
 
# 템플릿 사용
booking_scenario = CRUD_TEMPLATE.create(resource="항공편 예약")

import pytest
 
@pytest.fixture
def travel_agent():
    """여행 에이전트 픽스처"""
    agent = create_travel_agent(
        model="claude-sonnet-4-20250514",
        tools=["search_flights", "book_flight", "search_hotels",
               "book_hotel", "send_confirmation"],
        temperature=0.1,  # 테스트 시에는 낮은 temperature
    )
    yield agent
    agent.cleanup()
 
@pytest.fixture
def mock_apis():
    """외부 API 모킹 픽스처"""
    with MockAPIServer() as server:
        server.register("flights_api", mock_flight_data)
        server.register("hotels_api", mock_hotel_data)
        server.register("payment_api", mock_payment_success)
        yield server
 
@pytest.fixture(params=["tech_savvy", "non_technical", "impatient"])
def user_persona(request):
    """다양한 사용자 페르소나를 순회하는 픽스처"""
    return PERSONAS[request.param]

def analyze_e2e_result(result) -> dict:
    """E2E 테스트 결과를 상세히 분석"""
    tool_calls = result.state.get_all_tool_calls()
 
    return {
        "passed": result.success,
        "score": result.score,
        "total_turns": result.turn_count,
        "tool_calls": {
            "total": len(tool_calls),
            "unique_tools": len(set(c.name for c in tool_calls)),
            "by_tool": {
                name: len([c for c in tool_calls if c.name == name])
                for name in set(c.name for c in tool_calls)
            },
        },
        "issues": {
            "loops_detected": detect_loop(tool_calls),
            "excessive_turns": result.turn_count > 15,
            "tool_errors": len([c for c in tool_calls if c.error]),
        },
        "duration_seconds": result.duration,
    }