from __future__ import annotations from pathlib import Path import pytest from gitea_codex_bot.config import get_settings from gitea_codex_bot.workers.container_runner import ( CONTAINER_CODEX_HOME, _build_docker_command, _build_install_and_run_command, _extract_result_meta_from_codex_stdout, _load_codex_auth_json_b64, _parse_codex_exec_stdout, _resolve_codex_auth_json_path, run_review_ephemeral, ) def test_build_docker_command_api_key_mode_uses_openai_env() -> None: settings = get_settings() cmd = _build_docker_command(settings, container_name="codex-review-test", install_and_run="echo ok") assert "OPENAI_API_KEY" in cmd assert "OPENAI_ORG_ID" in cmd assert "OPENAI_PROJECT_ID" in cmd assert "--mount" not in cmd def test_build_docker_command_chatgpt_mode_mounts_auth_json( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: auth_file = tmp_path / "custom-auth.json" auth_file.write_text('{"auth_mode":"chatgpt"}', encoding="utf-8") monkeypatch.setenv("CODEX_AUTH_MODE", "chatgpt") monkeypatch.setenv("CODEX_AUTH_JSON_PATH", str(auth_file)) get_settings.cache_clear() settings = get_settings() cmd = _build_docker_command(settings, container_name="codex-review-test", install_and_run="echo ok") env_items = {value for index, value in enumerate(cmd) if index > 0 and cmd[index - 1] == "-e"} assert "OPENAI_API_KEY" not in cmd assert f"CODEX_HOME={CONTAINER_CODEX_HOME}" in env_items assert "CODEX_AUTH_JSON_B64" in env_items def test_build_install_command_chatgpt_mode_copies_auth_json(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: auth_file = tmp_path / "auth.json" auth_file.write_text("{}", encoding="utf-8") monkeypatch.setenv("CODEX_AUTH_MODE", "chatgpt") monkeypatch.setenv("CODEX_AUTH_JSON_PATH", str(auth_file)) get_settings.cache_clear() settings = get_settings() command = _build_install_and_run_command(settings) assert 'printf "%s" "$CODEX_AUTH_JSON_B64" | base64 -d > /root/.codex/auth.json' in command assert "codex exec --skip-git-repo-check --json -m gpt-5.3-codex" in command assert f"--reasoning-effort {settings.openai_reasoning_effort}" in command def test_build_install_command_includes_configured_reasoning_effort(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("OPENAI_REASONING_EFFORT", "medium") get_settings.cache_clear() settings = get_settings() command = _build_install_and_run_command(settings) assert "--reasoning-effort medium" in command def test_build_install_command_can_disable_reasoning_effort_flag() -> None: settings = get_settings() command = _build_install_and_run_command(settings, include_reasoning_effort=False) assert "--reasoning-effort" not in command def test_chatgpt_mode_requires_existing_auth_json(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: missing = tmp_path / "missing-auth.json" monkeypatch.setenv("CODEX_AUTH_MODE", "chatgpt") monkeypatch.setenv("CODEX_AUTH_JSON_PATH", str(missing)) get_settings.cache_clear() settings = get_settings() with pytest.raises(FileNotFoundError): _resolve_codex_auth_json_path(settings) def test_load_codex_auth_json_b64_roundtrip(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: auth_file = tmp_path / "auth.json" auth_file.write_text('{"auth_mode":"chatgpt","access_token":"abc"}', encoding="utf-8") monkeypatch.setenv("CODEX_AUTH_MODE", "chatgpt") monkeypatch.setenv("CODEX_AUTH_JSON_PATH", str(auth_file)) get_settings.cache_clear() settings = get_settings() encoded = _load_codex_auth_json_b64(settings) assert encoded def test_run_review_ephemeral_chatgpt_does_not_fallback_to_api_key_path( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: auth_file = tmp_path / "auth.json" auth_file.write_text('{"auth_mode":"chatgpt"}', encoding="utf-8") monkeypatch.setenv("CODEX_AUTH_MODE", "chatgpt") monkeypatch.setenv("CODEX_AUTH_JSON_PATH", str(auth_file)) get_settings.cache_clear() settings = get_settings() monkeypatch.setattr( "gitea_codex_bot.workers.container_runner.prepare_review_prompt", lambda *_args, **_kwargs: ("prompt", {"diff": ""}, object()), ) monkeypatch.setattr("gitea_codex_bot.workers.container_runner.GiteaClient", lambda _settings: object()) monkeypatch.setattr( "gitea_codex_bot.workers.container_runner.subprocess.run", lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("docker unavailable")), ) from gitea_codex_bot.types import ParsedCommand result, _repo_cfg = run_review_ephemeral( settings, repo="acme/repo", pr_number=1, command=ParsedCommand(name="review", raw="@codex review"), ) assert result["verdict"] == "has_issues" assert "ChatGPT auth runner failed" in result["summary"] def test_run_review_ephemeral_api_key_mode_does_not_fallback_to_host(monkeypatch: pytest.MonkeyPatch) -> None: get_settings.cache_clear() settings = get_settings() monkeypatch.setattr( "gitea_codex_bot.workers.container_runner.prepare_review_prompt", lambda *_args, **_kwargs: ("prompt", {"diff": ""}, object()), ) monkeypatch.setattr("gitea_codex_bot.workers.container_runner.GiteaClient", lambda _settings: object()) monkeypatch.setattr( "gitea_codex_bot.workers.container_runner.subprocess.run", lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("docker unavailable")), ) from gitea_codex_bot.types import ParsedCommand result, _repo_cfg = run_review_ephemeral( settings, repo="acme/repo", pr_number=1, command=ParsedCommand(name="review", raw="@codex review"), ) assert result["verdict"] == "has_issues" assert "API-key auth runner failed" in result["summary"] def test_run_review_ephemeral_retries_without_reasoning_effort_when_unsupported(monkeypatch: pytest.MonkeyPatch) -> None: get_settings.cache_clear() settings = get_settings() monkeypatch.setattr( "gitea_codex_bot.workers.container_runner.prepare_review_prompt", lambda *_args, **_kwargs: ("prompt", {"diff": ""}, object()), ) monkeypatch.setattr("gitea_codex_bot.workers.container_runner.GiteaClient", lambda _settings: object()) calls: list[list[str]] = [] def _fake_run(cmd, *args, **kwargs): calls.append(cmd) if len(calls) == 1: return type( "Completed", (), { "returncode": 2, "stdout": "", "stderr": "error: unexpected argument '--reasoning-effort' found", }, )() return type( "Completed", (), { "returncode": 0, "stdout": '{"verdict":"correct","confidence":0.9,"summary":"ok","findings":[]}\n', "stderr": "", }, )() monkeypatch.setattr("gitea_codex_bot.workers.container_runner.subprocess.run", _fake_run) from gitea_codex_bot.types import ParsedCommand result, _repo_cfg = run_review_ephemeral( settings, repo="acme/repo", pr_number=1, command=ParsedCommand(name="review", raw="@codex review"), ) assert result["verdict"] == "correct" assert len(calls) == 2 first_shell = calls[0][-1] second_shell = calls[1][-1] assert "--reasoning-effort" in first_shell assert "--reasoning-effort" not in second_shell def test_parse_codex_exec_stdout_from_stream_item_text_json() -> None: stdout = '\n'.join( [ '{"type":"thread.started","thread_id":"abc"}', '{"type":"item.completed","item":{"type":"agent_message","text":"{\\"verdict\\":\\"correct\\",\\"confidence\\":0.9,\\"summary\\":\\"ok\\",\\"findings\\":[]}"}}', ] ) parsed = _parse_codex_exec_stdout(stdout) assert parsed["verdict"] == "correct" assert parsed["summary"] == "ok" def test_parse_codex_exec_stdout_from_fenced_json_text() -> None: stdout = '\n'.join( [ '{"type":"thread.started","thread_id":"abc"}', '{"type":"item.completed","item":{"type":"agent_message","text":"Here is the result:\\n```json\\n{\\"verdict\\":\\"has_issues\\",\\"confidence\\":0.8,\\"summary\\":\\"x\\",\\"findings\\":[]}\\n```"}}', ] ) parsed = _parse_codex_exec_stdout(stdout) assert parsed["verdict"] == "has_issues" assert parsed["summary"] == "x" def test_extract_result_meta_from_codex_stdout_collects_model_and_usage() -> None: settings = get_settings() stdout = '\n'.join( [ '{"type":"response.started","model":"gpt-5.3-codex"}', '{"type":"response.completed","response":{"usage":{"input_tokens":101,"output_tokens":22,"total_tokens":123}}}', ] ) meta = _extract_result_meta_from_codex_stdout(stdout, settings) assert meta["model"] == "gpt-5.3-codex" assert meta["usage"]["input_tokens"] == 101 assert meta["usage"]["output_tokens"] == 22 assert meta["usage"]["total_tokens"] == 123