gitea-codex/tests/test_container_runner.py

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,
    _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


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")),
    )

    def _api_fallback_should_not_run(*_args, **_kwargs):
        raise AssertionError("API-key fallback should not run in chatgpt mode")

    monkeypatch.setattr("gitea_codex_bot.workers.container_runner.run_review_for_pr", _api_fallback_should_not_run)

    from gitea_codex_bot.types import ParsedCommand

    result = 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_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"