314 lines
11 KiB
Python
314 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from gitea_codex_bot.config import get_settings
|
|
from gitea_codex_bot.services.gitea import PullRequestContext
|
|
from gitea_codex_bot.types import ParsedCommand
|
|
from gitea_codex_bot.workers.container_runner import (
|
|
CONTAINER_CODEX_HOME,
|
|
RESULT_END_MARKER,
|
|
RESULT_START_MARKER,
|
|
_build_docker_command,
|
|
_build_install_and_run_command,
|
|
_extract_result_meta_from_codex_stdout,
|
|
_load_codex_auth_json_b64,
|
|
_load_repo_review_config_from_gitea,
|
|
_parse_review_result_from_stdout_artifact,
|
|
_resolve_codex_auth_json_path,
|
|
run_review_ephemeral,
|
|
)
|
|
|
|
|
|
def _sample_pr() -> PullRequestContext:
|
|
return PullRequestContext(
|
|
repo="acme/repo",
|
|
pr_number=1,
|
|
base_ref="main",
|
|
base_sha="b" * 40,
|
|
head_ref="feature",
|
|
head_sha="a" * 40,
|
|
clone_url="https://gitea.test/acme/repo.git",
|
|
html_url="https://gitea.test/acme/repo/pulls/1",
|
|
is_fork=False,
|
|
)
|
|
|
|
|
|
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 "GITEA_TOKEN" in cmd
|
|
assert "GITEA_GIT_USERNAME" 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
|
|
assert "GITEA_TOKEN" in env_items
|
|
assert "GITEA_GIT_USERNAME" in env_items
|
|
|
|
|
|
def test_build_install_command_chatgpt_mode_sets_git_checkout_and_review(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()
|
|
pr = _sample_pr()
|
|
|
|
command = _build_install_and_run_command(settings, pr=pr)
|
|
|
|
assert 'printf "%s" "$CODEX_AUTH_JSON_B64" | base64 -d > /root/.codex/auth.json' in command
|
|
assert "git -c http.extraHeader=" in command
|
|
assert f"clone --no-tags --depth 80 {pr.clone_url} /work/repo" in command
|
|
assert f"fetch --no-tags origin {pr.base_ref} {pr.head_ref}" in command
|
|
assert f"git checkout --detach {pr.head_sha}" in command
|
|
assert "resolved_head=\"$(git rev-parse HEAD)\"" in command
|
|
assert "unset GITEA_TOKEN auth_b64" in command
|
|
assert f"codex exec review --base {pr.base_sha}" in command
|
|
assert "--output-schema /tmp/codex-review-schema.json" in command
|
|
assert "-o /tmp/codex-review-result.json" in command
|
|
assert f'echo "{RESULT_START_MARKER}"' in command
|
|
assert f'echo "{RESULT_END_MARKER}"' in command
|
|
|
|
|
|
def test_build_install_command_can_disable_reasoning_effort_flag() -> None:
|
|
settings = get_settings()
|
|
pr = _sample_pr()
|
|
|
|
command = _build_install_and_run_command(settings, pr=pr, 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_load_repo_review_config_from_gitea_when_missing() -> None:
|
|
class _Gitea:
|
|
def get_file_content(self, *_args, **_kwargs):
|
|
return None
|
|
|
|
cfg = _load_repo_review_config_from_gitea(_Gitea(), "acme/repo", "a" * 40)
|
|
|
|
assert cfg.configured is False
|
|
assert cfg.enabled is True
|
|
|
|
|
|
def test_load_repo_review_config_from_gitea_when_present() -> None:
|
|
class _Gitea:
|
|
def get_file_content(self, *_args, **_kwargs):
|
|
return "enabled: false\nreview:\n default_mode: tests\n"
|
|
|
|
cfg = _load_repo_review_config_from_gitea(_Gitea(), "acme/repo", "a" * 40)
|
|
|
|
assert cfg.configured is True
|
|
assert cfg.enabled is False
|
|
assert cfg.default_mode == "tests"
|
|
|
|
|
|
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()
|
|
|
|
class _FakeGiteaClient:
|
|
def __init__(self, _settings) -> None:
|
|
pass
|
|
|
|
def get_pull_request(self, *_args, **_kwargs):
|
|
return _sample_pr()
|
|
|
|
def get_file_content(self, *_args, **_kwargs):
|
|
return None
|
|
|
|
monkeypatch.setattr("gitea_codex_bot.workers.container_runner.GiteaClient", _FakeGiteaClient)
|
|
monkeypatch.setattr(
|
|
"gitea_codex_bot.workers.container_runner.subprocess.run",
|
|
lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("docker unavailable")),
|
|
)
|
|
|
|
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()
|
|
|
|
class _FakeGiteaClient:
|
|
def __init__(self, _settings) -> None:
|
|
pass
|
|
|
|
def get_pull_request(self, *_args, **_kwargs):
|
|
return _sample_pr()
|
|
|
|
def get_file_content(self, *_args, **_kwargs):
|
|
return None
|
|
|
|
monkeypatch.setattr("gitea_codex_bot.workers.container_runner.GiteaClient", _FakeGiteaClient)
|
|
monkeypatch.setattr(
|
|
"gitea_codex_bot.workers.container_runner.subprocess.run",
|
|
lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("docker unavailable")),
|
|
)
|
|
|
|
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()
|
|
|
|
class _FakeGiteaClient:
|
|
def __init__(self, _settings) -> None:
|
|
pass
|
|
|
|
def get_pull_request(self, *_args, **_kwargs):
|
|
return _sample_pr()
|
|
|
|
def get_file_content(self, *_args, **_kwargs):
|
|
return None
|
|
|
|
monkeypatch.setattr("gitea_codex_bot.workers.container_runner.GiteaClient", _FakeGiteaClient)
|
|
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": (
|
|
'{"type":"response.started","model":"gpt-5.3-codex"}\n'
|
|
f"{RESULT_START_MARKER}\n"
|
|
'{"verdict":"correct","confidence":0.9,"summary":"ok","findings":[],"markdown_comment":"ok"}\n'
|
|
f"{RESULT_END_MARKER}\n"
|
|
),
|
|
"stderr": "",
|
|
},
|
|
)()
|
|
|
|
monkeypatch.setattr("gitea_codex_bot.workers.container_runner.subprocess.run", _fake_run)
|
|
|
|
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_review_result_from_stdout_artifact() -> None:
|
|
stdout = (
|
|
"noise\n"
|
|
f"{RESULT_START_MARKER}\n"
|
|
'{"verdict":"correct","confidence":0.9,"summary":"ok","findings":[],"markdown_comment":"ok"}\n'
|
|
f"{RESULT_END_MARKER}\n"
|
|
)
|
|
parsed = _parse_review_result_from_stdout_artifact(stdout)
|
|
assert parsed["verdict"] == "correct"
|
|
assert parsed["summary"] == "ok"
|
|
|
|
|
|
def test_parse_review_result_from_stdout_artifact_fails_without_markers() -> None:
|
|
with pytest.raises(RuntimeError):
|
|
_parse_review_result_from_stdout_artifact("no markers here")
|
|
|
|
|
|
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
|