import os import sys sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) import asyncio from unittest.mock import AsyncMock, patch import pytest from engine_v2.cli_runner import CLIResult # --------------------------------------------------------------------------- # 헬퍼: CLIResult 팩토리 # --------------------------------------------------------------------------- def _ok_result(engine: str, stdout: str, stderr: str = "") -> CLIResult: return CLIResult(stdout=stdout, stderr=stderr, returncode=0, engine=engine) # type: ignore[arg-type] def _err_result(engine: str, stdout: str = "", stderr: str = "", returncode: int = 1) -> CLIResult: return CLIResult(stdout=stdout, stderr=stderr, returncode=returncode, engine=engine) # type: ignore[arg-type] def _timeout_result(engine: str) -> CLIResult: return CLIResult(stdout="", stderr="Timeout", returncode=-1, engine=engine, timed_out=True) # type: ignore[arg-type] # --------------------------------------------------------------------------- # 테스트: call_gemini() — CLI 기반 # --------------------------------------------------------------------------- class TestCallGemini: """call_gemini() CLIRunner 기반 동작 검증 (engine.py redirect 경유)""" @pytest.mark.asyncio async def test_call_gemini_returns_normal_response(self): """정상 응답 시 CLIResult.stdout을 그대로 반환해야 한다""" import engine result = _ok_result("gemini", "Gemini의 응답입니다.") with patch("engine_v2.bot_api.CLIRunner.run_gemini", new=AsyncMock(return_value=result)): res = await engine.call_gemini("테스트 질문입니다.") assert res == "Gemini의 응답입니다." @pytest.mark.asyncio async def test_call_gemini_timeout_returns_error_message(self): """timed_out=True인 CLIResult가 반환되면 타임아웃 관련 메시지를 반환해야 한다""" import engine result = _timeout_result("gemini") with patch("engine_v2.bot_api.CLIRunner.run_gemini", new=AsyncMock(return_value=result)): res = await engine.call_gemini("타임아웃 테스트", timeout=1) assert res is not None assert len(res) > 0 assert any(keyword in res.lower() for keyword in ["timeout", "시간", "초과", "timed out", "타임아웃"]) @pytest.mark.asyncio async def test_call_gemini_api_error_403_returns_error_message(self): """returncode != 0이면 에러 메시지를 반환해야 한다 (403 케이스)""" import engine result = _err_result("gemini", stderr="403 에러", returncode=1) with patch("engine_v2.bot_api.CLIRunner.run_gemini", new=AsyncMock(return_value=result)): res = await engine.call_gemini("403 에러 테스트") assert res is not None assert len(res) > 0 assert "❌" in res @pytest.mark.asyncio async def test_call_gemini_api_error_500_returns_error_message(self): """returncode != 0이면 에러 메시지를 반환해야 한다 (500 케이스)""" import engine result = _err_result("gemini", stderr="500 에러", returncode=1) with patch("engine_v2.bot_api.CLIRunner.run_gemini", new=AsyncMock(return_value=result)): res = await engine.call_gemini("500 에러 테스트") assert res is not None assert len(res) > 0 assert "❌" in res @pytest.mark.asyncio async def test_call_gemini_empty_candidates_returns_fallback_message(self): """stdout이 비어 있으면 fallback 메시지를 반환해야 한다""" import engine result = _ok_result("gemini", "") with patch("engine_v2.bot_api.CLIRunner.run_gemini", new=AsyncMock(return_value=result)): res = await engine.call_gemini("빈 응답 테스트") assert res is not None assert len(res) > 0 assert "⚠️" in res @pytest.mark.asyncio async def test_call_gemini_empty_text_returns_fallback_message(self): """stdout이 빈 문자열이면 fallback 메시지를 반환해야 한다""" import engine result = _ok_result("gemini", "") with patch("engine_v2.bot_api.CLIRunner.run_gemini", new=AsyncMock(return_value=result)): res = await engine.call_gemini("빈 텍스트 테스트") assert res is not None assert len(res) > 0 assert "⚠️" in res @pytest.mark.asyncio async def test_call_gemini_general_exception_returns_error_message(self): """일반 에러(returncode != 0, stderr 있음) 시 에러 메시지를 반환해야 한다""" import engine result = _err_result("gemini", stderr="unexpected error", returncode=-1) with patch("engine_v2.bot_api.CLIRunner.run_gemini", new=AsyncMock(return_value=result)): res = await engine.call_gemini("예외 테스트") assert res is not None assert len(res) > 0 assert "❌" in res # --------------------------------------------------------------------------- # 테스트: call_codex() # --------------------------------------------------------------------------- class TestCallCodex: """call_codex() CLIRunner 기반 동작 검증 (engine.py redirect 경유)""" @pytest.mark.asyncio async def test_call_codex_returns_normal_response(self): """정상 응답 시 CLIResult.stdout을 그대로 반환해야 한다""" import engine result = _ok_result("codex", "Codex의 응답입니다.") with patch("engine_v2.bot_api.CLIRunner.run_codex", new=AsyncMock(return_value=result)): res = await engine.call_codex("코드 리뷰 요청") assert res == "Codex의 응답입니다." @pytest.mark.asyncio async def test_call_codex_passes_prompt_to_subprocess(self): """call_codex()는 프롬프트를 CLIRunner에 첫 번째 인자로 전달해야 한다""" import engine result = _ok_result("codex", "응답") mock_run = AsyncMock(return_value=result) with patch("engine_v2.bot_api.CLIRunner.run_codex", new=mock_run): await engine.call_codex("함수 작성해줘") args, _ = mock_run.call_args assert args[0] == "함수 작성해줘" @pytest.mark.asyncio async def test_call_codex_not_logged_in_returns_auth_error_message(self): """stderr에 login/auth 관련 텍스트가 있으면 로그인 안내 메시지를 반환해야 한다""" import engine result = _err_result("codex", stderr="Error: Not authenticated. Please run `codex login`.", returncode=1) with patch("engine_v2.bot_api.CLIRunner.run_codex", new=AsyncMock(return_value=result)): res = await engine.call_codex("로그인 없이 요청") assert res is not None assert len(res) > 0 assert any(keyword in res.lower() for keyword in ["login", "auth", "로그인", "인증"]) @pytest.mark.asyncio async def test_call_codex_auth_keyword_in_stderr_triggers_login_message(self): """stderr에 'auth' 키워드가 포함되면 로그인 안내 메시지를 반환해야 한다""" import engine result = _err_result("codex", stderr="authentication required", returncode=1) with patch("engine_v2.bot_api.CLIRunner.run_codex", new=AsyncMock(return_value=result)): res = await engine.call_codex("auth 에러 테스트") assert res is not None assert len(res) > 0 assert any(keyword in res.lower() for keyword in ["login", "auth", "로그인", "인증"]) @pytest.mark.asyncio async def test_call_codex_file_not_found_returns_error_message(self): """CLIRunner가 FileNotFoundError 에러 결과를 반환하면 에러 메시지를 반환해야 한다""" import engine result = _err_result("codex", stderr="codex not found in PATH", returncode=-1) with patch("engine_v2.bot_api.CLIRunner.run_codex", new=AsyncMock(return_value=result)): res = await engine.call_codex("명령어 없음 테스트") assert res is not None assert len(res) > 0 @pytest.mark.asyncio async def test_call_codex_timeout_returns_error_message(self): """timed_out=True인 CLIResult가 반환되면 타임아웃 관련 메시지를 반환해야 한다""" import engine result = _timeout_result("codex") with patch("engine_v2.bot_api.CLIRunner.run_codex", new=AsyncMock(return_value=result)): res = await engine.call_codex("타임아웃 테스트") assert res is not None assert len(res) > 0 assert any(keyword in res.lower() for keyword in ["timeout", "시간", "초과", "timed out", "타임아웃"]) @pytest.mark.asyncio async def test_call_codex_timeout_uses_configured_timeout(self): """call_codex()는 timeout 파라미터를 CLIRunner.run_codex에 전달해야 한다""" import engine result = _ok_result("codex", "응답") mock_run = AsyncMock(return_value=result) with patch("engine_v2.bot_api.CLIRunner.run_codex", new=mock_run): await engine.call_codex("타임아웃 값 확인", timeout=30) _, kwargs = mock_run.call_args assert kwargs.get("timeout") == 30 @pytest.mark.asyncio async def test_call_codex_strips_whitespace_from_output(self): """CLIRunner가 이미 strip()한 결과를 반환하므로 정상 응답이 그대로 전달되어야 한다""" import engine result = _ok_result("codex", "코드 응답") with patch("engine_v2.bot_api.CLIRunner.run_codex", new=AsyncMock(return_value=result)): res = await engine.call_codex("공백 제거 테스트") assert res == "코드 응답" @pytest.mark.asyncio async def test_call_codex_empty_output_returns_fallback_message(self): """stdout이 비어 있으면 fallback 메시지를 반환해야 한다""" import engine result = _ok_result("codex", "") with patch("engine_v2.bot_api.CLIRunner.run_codex", new=AsyncMock(return_value=result)): res = await engine.call_codex("빈 응답 테스트") assert res is not None assert len(res) > 0 @pytest.mark.asyncio async def test_call_codex_usage_limit_returns_warning(self): """stderr에 'usage limit' 포함 시 사용량 한도 초과 경고 메시지를 반환해야 한다""" import engine result = _err_result( "codex", stderr="Error: You have reached your usage limit. Please try again later.", returncode=1, ) with patch("engine_v2.bot_api.CLIRunner.run_codex", new=AsyncMock(return_value=result)): res = await engine.call_codex("사용량 한도 테스트") assert res == "⚠️ Codex 사용량 한도 초과. 잠시 후 다시 시도해주세요." @pytest.mark.asyncio async def test_call_codex_hit_your_returns_warning(self): """stderr에 'hit your' 포함 시 사용량 한도 초과 경고 메시지를 반환해야 한다""" import engine result = _err_result("codex", stderr="You have hit your monthly quota.", returncode=1) with patch("engine_v2.bot_api.CLIRunner.run_codex", new=AsyncMock(return_value=result)): res = await engine.call_codex("hit your 에러 테스트") assert res == "⚠️ Codex 사용량 한도 초과. 잠시 후 다시 시도해주세요." @pytest.mark.asyncio async def test_call_codex_usage_limit_no_stderr_exposure(self): """usage limit 에러 시 stderr 원문이 반환 메시지에 포함되지 않아야 한다""" import engine result = _err_result( "codex", stderr="Error: usage limit reached. token=secret-api-key-12345", returncode=1, ) with patch("engine_v2.bot_api.CLIRunner.run_codex", new=AsyncMock(return_value=result)): res = await engine.call_codex("stderr 노출 방지 테스트") assert "secret-api-key-12345" not in res assert "token=" not in res @pytest.mark.asyncio async def test_call_codex_error_filters_error_lines_only(self): """stderr에 여러 줄이 있고 일부만 'error' 포함 시 error 줄만 필터링되어 반환되어야 한다""" import engine result = _err_result( "codex", stderr="info: starting process\nError: connection refused\ndebug: retrying\n", returncode=1, ) with patch("engine_v2.bot_api.CLIRunner.run_codex", new=AsyncMock(return_value=result)): res = await engine.call_codex("error 필터링 테스트") assert "Error: connection refused" in res assert "info: starting process" not in res assert "debug: retrying" not in res @pytest.mark.asyncio async def test_call_codex_error_no_error_lines_returns_generic(self): """stderr에 'error' 포함 줄이 없으면 '상세 내용 없음' 메시지를 반환해야 한다""" import engine result = _err_result( "codex", stderr="warning: something happened\ninfo: process exited\n", returncode=1, ) with patch("engine_v2.bot_api.CLIRunner.run_codex", new=AsyncMock(return_value=result)): res = await engine.call_codex("error 줄 없음 테스트") assert "상세 내용 없음" in res @pytest.mark.asyncio async def test_call_codex_error_filtered_max_300_chars(self): """필터링된 에러가 300자 초과 시 300자로 잘려야 한다""" import engine long_error_line = "Error: " + "x" * 400 result = _err_result("codex", stderr=long_error_line, returncode=1) with patch("engine_v2.bot_api.CLIRunner.run_codex", new=AsyncMock(return_value=result)): res = await engine.call_codex("300자 제한 테스트") prefix = "❌ Codex CLI 에러 (exit 1): " error_part = res[len(prefix) :] assert len(error_part) <= 300 @pytest.mark.asyncio async def test_call_codex_auth_error_no_stderr_exposure(self): """auth 에러 메시지에서도 stderr 전문이 노출되지 않아야 한다""" import engine result = _err_result("codex", stderr="auth failed: token=super-secret-system-prompt-data", returncode=1) with patch("engine_v2.bot_api.CLIRunner.run_codex", new=AsyncMock(return_value=result)): res = await engine.call_codex("auth stderr 노출 방지 테스트") assert "super-secret-system-prompt-data" not in res assert "token=" not in res @pytest.mark.asyncio async def test_call_codex_default_model_is_mini(self): """model 파라미터 없이 호출하면 CLIRunner에 gpt-5.1-codex-mini가 전달되어야 한다""" import engine result = _ok_result("codex", "ok") mock_run = AsyncMock(return_value=result) with patch("engine_v2.bot_api.CLIRunner.run_codex", new=mock_run): await engine.call_codex("기본 모델 테스트") _, kwargs = mock_run.call_args assert kwargs.get("model") == "gpt-5.1-codex-mini" @pytest.mark.asyncio async def test_call_codex_custom_model_passed_to_cmd(self): """model='gpt-5.2-codex'로 호출하면 CLIRunner에 gpt-5.2-codex가 전달되어야 한다""" import engine result = _ok_result("codex", "ok") mock_run = AsyncMock(return_value=result) with patch("engine_v2.bot_api.CLIRunner.run_codex", new=mock_run): await engine.call_codex("커스텀 모델 테스트", model="gpt-5.2-codex") _, kwargs = mock_run.call_args assert kwargs.get("model") == "gpt-5.2-codex" @pytest.mark.asyncio async def test_call_codex_model_not_hardcoded(self): """기본 호출 시 CLIRunner에 gpt-5.2-codex가 전달되지 않아야 한다 (하드코딩 제거 확인)""" import engine result = _ok_result("codex", "ok") mock_run = AsyncMock(return_value=result) with patch("engine_v2.bot_api.CLIRunner.run_codex", new=mock_run): await engine.call_codex("하드코딩 제거 확인") _, kwargs = mock_run.call_args assert kwargs.get("model") != "gpt-5.2-codex" class TestCallClaude: """call_claude() CLIRunner 기반 동작 검증 (engine.py redirect 경유)""" @pytest.mark.asyncio async def test_call_claude_returns_normal_response(self): """정상 응답 시 CLIResult.stdout을 그대로 반환해야 한다""" import engine result = _ok_result("claude", "Claude의 응답입니다.") with patch("engine_v2.bot_api.CLIRunner.run_claude", new=AsyncMock(return_value=result)): res = await engine.call_claude("테스트 질문") assert res == "Claude의 응답입니다." @pytest.mark.asyncio async def test_call_claude_default_no_allowed_tools(self): """code_analysis=False(기본값)면 CLIRunner에 code_analysis=False가 전달되어야 한다""" import engine result = _ok_result("claude", "ok") mock_run = AsyncMock(return_value=result) with patch("engine_v2.bot_api.CLIRunner.run_claude", new=mock_run): await engine.call_claude("일반 질문") _, kwargs = mock_run.call_args assert kwargs.get("code_analysis") is False @pytest.mark.asyncio async def test_call_claude_code_analysis_adds_allowed_tools(self): """code_analysis=True면 CLIRunner에 code_analysis=True가 전달되어야 한다""" import engine result = _ok_result("claude", "code result") mock_run = AsyncMock(return_value=result) with patch("engine_v2.bot_api.CLIRunner.run_claude", new=mock_run): await engine.call_claude("코드 분석", code_analysis=True) _, kwargs = mock_run.call_args assert kwargs.get("code_analysis") is True @pytest.mark.asyncio async def test_call_claude_code_analysis_changes_cwd(self): """code_analysis=True면 CLIRunner 내부에서 cwd가 /home/jay/workspace로 설정되어야 한다 (파라미터 전달 검증)""" import engine result = _ok_result("claude", "ok") mock_run = AsyncMock(return_value=result) with patch("engine_v2.bot_api.CLIRunner.run_claude", new=mock_run): await engine.call_claude("코드 분석", code_analysis=True) _, kwargs = mock_run.call_args # code_analysis=True가 전달되면 CLIRunner 내부에서 cwd 결정 assert kwargs.get("code_analysis") is True @pytest.mark.asyncio async def test_call_claude_default_cwd_is_tmp(self): """code_analysis=False(기본값)이면 CLIRunner에 code_analysis=False가 전달되어야 한다""" import engine result = _ok_result("claude", "ok") mock_run = AsyncMock(return_value=result) with patch("engine_v2.bot_api.CLIRunner.run_claude", new=mock_run): await engine.call_claude("일반 질문") _, kwargs = mock_run.call_args assert kwargs.get("code_analysis") is False @pytest.mark.asyncio async def test_call_claude_code_analysis_no_write_tools(self): """code_analysis=True에서 code_analysis 파라미터가 True로 전달되어야 한다""" import engine result = _ok_result("claude", "ok") mock_run = AsyncMock(return_value=result) with patch("engine_v2.bot_api.CLIRunner.run_claude", new=mock_run): await engine.call_claude("코드 분석", code_analysis=True) _, kwargs = mock_run.call_args assert kwargs.get("code_analysis") is True @pytest.mark.asyncio async def test_call_claude_timeout_returns_error_message(self): """타임아웃 시 에러 메시지를 반환해야 한다""" import engine result = _timeout_result("claude") with patch("engine_v2.bot_api.CLIRunner.run_claude", new=AsyncMock(return_value=result)): res = await engine.call_claude("타임아웃 테스트") assert "시간 초과" in res or "timeout" in res.lower() @pytest.mark.asyncio async def test_call_claude_removes_claudecode_env(self): """CLAUDECODE 환경변수 제거가 CLIRunner 레벨에서 처리되어야 한다 (정상 응답 반환 확인)""" import engine result = _ok_result("claude", "ok") mock_run = AsyncMock(return_value=result) os.environ["CLAUDECODE"] = "test-value" try: with patch("engine_v2.bot_api.CLIRunner.run_claude", new=mock_run): res = await engine.call_claude("env 테스트") assert res == "ok" finally: os.environ.pop("CLAUDECODE", None) @pytest.mark.asyncio async def test_call_claude_nonzero_returncode_empty_output_filters_errors(self): """returncode != 0이고 stdout 비어있을 때 stderr에서 error 줄만 필터링하여 반환해야 한다""" import engine result = _err_result( "claude", stdout="", stderr="info: starting\nError: authentication failed\ndebug: done\n", returncode=1, ) with patch("engine_v2.bot_api.CLIRunner.run_claude", new=AsyncMock(return_value=result)): res = await engine.call_claude("에러 필터링 테스트") assert "Error: authentication failed" in res assert "info: starting" not in res assert "debug: done" not in res @pytest.mark.asyncio async def test_call_claude_nonzero_returncode_empty_output_no_error_lines(self): """returncode != 0이고 stdout 비어있고 error 줄도 없으면 '상세 내용 없음'을 반환해야 한다""" import engine result = _err_result( "claude", stdout="", stderr="warning: something odd\ninfo: exiting\n", returncode=2, ) with patch("engine_v2.bot_api.CLIRunner.run_claude", new=AsyncMock(return_value=result)): res = await engine.call_claude("error 줄 없음 테스트") assert "상세 내용 없음" in res @pytest.mark.asyncio async def test_call_claude_nonzero_returncode_with_output_returns_output(self): """returncode != 0이더라도 stdout에 출력이 있으면 그 출력을 반환해야 한다""" import engine result = _err_result( "claude", stdout="partial output from claude", stderr="Error: some stderr error", returncode=1, ) with patch("engine_v2.bot_api.CLIRunner.run_claude", new=AsyncMock(return_value=result)): res = await engine.call_claude("출력 있는 에러 테스트") assert res == "partial output from claude"