446 lines
16 KiB
Python
446 lines
16 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.services.repo_config import RepoReviewConfig
|
|
from gitea_codex_bot.types import ParsedCommand
|
|
from gitea_codex_bot.workers.container_runner import (
|
|
CONTAINER_CODEX_HOME,
|
|
RESULT_END_MARKER,
|
|
RESULT_START_MARKER,
|
|
_apply_repo_default_review_mode,
|
|
_build_docker_command,
|
|
_build_exec_review_prompt,
|
|
_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 _sample_fork_pr() -> PullRequestContext:
|
|
return PullRequestContext(
|
|
repo="acme/repo",
|
|
pr_number=2,
|
|
base_ref="main",
|
|
base_sha="c" * 40,
|
|
head_ref="feature",
|
|
head_sha="d" * 40,
|
|
clone_url="https://gitea.test/fork/repo.git",
|
|
base_clone_url="https://gitea.test/acme/repo.git",
|
|
head_clone_url="https://gitea.test/fork/repo.git",
|
|
html_url="https://gitea.test/acme/repo/pulls/2",
|
|
is_fork=True,
|
|
)
|
|
|
|
|
|
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,
|
|
review_prompt="review: security --full",
|
|
result_start_marker=f"{RESULT_START_MARKER}_x",
|
|
result_end_marker=f"{RESULT_END_MARKER}_x",
|
|
)
|
|
|
|
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 "fetch_required() {" in command
|
|
assert f"fetch_required origin {pr.head_ref} {pr.head_sha} head" in command
|
|
assert f"fetch_required \"$base_remote\" {pr.base_ref} {pr.base_sha} base" in command
|
|
assert "base_remote=origin" 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 (
|
|
"codex exec --sandbox danger-full-access --json --output-schema /tmp/codex-review-schema.json "
|
|
"-o /tmp/codex-review-result.json"
|
|
) in command
|
|
assert "review: security --full" in command
|
|
assert "--output-schema /tmp/codex-review-schema.json" in command
|
|
assert "-o /tmp/codex-review-result.json" in command
|
|
assert "npm install -g @openai/codex@latest" in command
|
|
assert "codex --version >/tmp/codex-version.log" in command
|
|
assert " - " not in command
|
|
assert f'echo "{RESULT_START_MARKER}_x"' in command
|
|
assert f'echo "{RESULT_END_MARKER}_x"' in command
|
|
|
|
|
|
def test_build_install_command_does_not_include_reasoning_effort_flag() -> None:
|
|
settings = get_settings()
|
|
pr = _sample_pr()
|
|
|
|
command = _build_install_and_run_command(
|
|
settings,
|
|
pr=pr,
|
|
review_prompt="review: tests",
|
|
result_start_marker=f"{RESULT_START_MARKER}_x",
|
|
result_end_marker=f"{RESULT_END_MARKER}_x",
|
|
)
|
|
|
|
assert "--reasoning-effort" not in command
|
|
|
|
|
|
def test_build_install_command_uses_upstream_remote_for_fork_pr_base_fetch() -> None:
|
|
settings = get_settings()
|
|
pr = _sample_fork_pr()
|
|
|
|
command = _build_install_and_run_command(
|
|
settings,
|
|
pr=pr,
|
|
review_prompt="review: tests",
|
|
result_start_marker=f"{RESULT_START_MARKER}_x",
|
|
result_end_marker=f"{RESULT_END_MARKER}_x",
|
|
)
|
|
|
|
assert "base_remote=upstream" in command
|
|
assert f"git remote add upstream {pr.base_clone_url}" in command
|
|
assert f"fetch_required origin {pr.head_ref} {pr.head_sha} head" in command
|
|
assert f"fetch_required \"$base_remote\" {pr.base_ref} {pr.base_sha} base" 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_single_attempt_success(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.uuid.uuid4",
|
|
lambda: type("U", (), {"hex": "abc123def4567890abc123def4567890"})(),
|
|
)
|
|
calls: list[list[str]] = []
|
|
|
|
def _fake_run(cmd, *args, **kwargs):
|
|
calls.append(cmd)
|
|
return type(
|
|
"Completed",
|
|
(),
|
|
{
|
|
"returncode": 0,
|
|
"stdout": (
|
|
'{"type":"response.started","model":"gpt-5.3-codex"}\n'
|
|
f"{RESULT_START_MARKER}_abc123def4567890abc123def4567890\n"
|
|
'{"verdict":"correct","confidence":0.9,"summary":"ok","findings":[],"markdown_comment":"ok"}\n'
|
|
f"{RESULT_END_MARKER}_abc123def4567890abc123def4567890\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) == 1
|
|
first_shell = calls[0][-1]
|
|
assert "--reasoning-effort" not in first_shell
|
|
|
|
|
|
def test_parse_review_result_from_stdout_artifact() -> None:
|
|
stdout = (
|
|
"noise\n"
|
|
f"{RESULT_START_MARKER}_test\n"
|
|
'{"verdict":"correct","confidence":0.9,"summary":"ok","findings":[],"markdown_comment":"ok"}\n'
|
|
f"{RESULT_END_MARKER}_test\n"
|
|
)
|
|
parsed = _parse_review_result_from_stdout_artifact(
|
|
stdout,
|
|
result_start_marker=f"{RESULT_START_MARKER}_test",
|
|
result_end_marker=f"{RESULT_END_MARKER}_test",
|
|
)
|
|
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",
|
|
result_start_marker=f"{RESULT_START_MARKER}_x",
|
|
result_end_marker=f"{RESULT_END_MARKER}_x",
|
|
)
|
|
|
|
|
|
def test_build_exec_review_prompt_strips_mention_and_command() -> None:
|
|
prompt = _build_exec_review_prompt(
|
|
ParsedCommand(name="review", raw="@codex review security --full\nfocus session handling"),
|
|
RepoReviewConfig(),
|
|
_sample_pr(),
|
|
)
|
|
assert prompt.startswith("review: security --full\nfocus session handling")
|
|
assert "Compare exactly these commits:" in prompt
|
|
|
|
|
|
def test_build_exec_review_prompt_falls_back_when_no_extra_text() -> None:
|
|
prompt = _build_exec_review_prompt(ParsedCommand(name="rerun", raw="@codex rerun"), RepoReviewConfig(), _sample_pr())
|
|
assert prompt.startswith("review: review this pull request and report introduced issues.")
|
|
|
|
|
|
def test_build_exec_review_prompt_disables_test_execution_by_default() -> None:
|
|
prompt = _build_exec_review_prompt(ParsedCommand(name="review", raw="@codex review"), RepoReviewConfig(), _sample_pr())
|
|
assert "Do not run tests, benchmarks, or other executables." in prompt
|
|
|
|
|
|
def test_build_exec_review_prompt_allows_test_execution_for_tests_mode() -> None:
|
|
prompt = _build_exec_review_prompt(
|
|
ParsedCommand(name="review", raw="@codex review tests", mode="tests", mode_explicit=True),
|
|
RepoReviewConfig(),
|
|
_sample_pr(),
|
|
)
|
|
assert "Tests may be executed for this run" in prompt
|
|
|
|
|
|
def test_apply_repo_default_review_mode_uses_full_when_not_configured() -> None:
|
|
command = ParsedCommand(name="review", raw="@codex review")
|
|
cfg = RepoReviewConfig()
|
|
_apply_repo_default_review_mode(command, cfg)
|
|
assert command.mode == "full"
|
|
|
|
|
|
def test_apply_repo_default_review_mode_for_review_command() -> None:
|
|
command = ParsedCommand(name="review", raw="@codex review")
|
|
cfg = RepoReviewConfig(default_mode="tests")
|
|
_apply_repo_default_review_mode(command, cfg)
|
|
assert command.mode == "tests"
|
|
|
|
|
|
def test_parse_review_result_from_stdout_artifact_uses_end_marker_after_start() -> None:
|
|
stdout = (
|
|
f"{RESULT_START_MARKER}_a\n"
|
|
'{"verdict":"correct","confidence":0.9,"summary":"contains marker text __CODEX_REVIEW_RESULT_END___a","findings":[],"markdown_comment":"ok"}\n'
|
|
f"{RESULT_END_MARKER}_a\n"
|
|
)
|
|
parsed = _parse_review_result_from_stdout_artifact(
|
|
stdout,
|
|
result_start_marker=f"{RESULT_START_MARKER}_a",
|
|
result_end_marker=f"{RESULT_END_MARKER}_a",
|
|
)
|
|
assert parsed["verdict"] == "correct"
|
|
|
|
|
|
def test_parse_review_result_from_stdout_artifact_handles_inline_end_marker() -> None:
|
|
stdout = (
|
|
"noise\n"
|
|
f"{RESULT_START_MARKER}_a\n"
|
|
'{"verdict":"correct","confidence":0.9,"summary":"ok","findings":[],"markdown_comment":"ok"}'
|
|
f"{RESULT_END_MARKER}_a\n"
|
|
)
|
|
parsed = _parse_review_result_from_stdout_artifact(
|
|
stdout,
|
|
result_start_marker=f"{RESULT_START_MARKER}_a",
|
|
result_end_marker=f"{RESULT_END_MARKER}_a",
|
|
)
|
|
assert parsed["verdict"] == "correct"
|
|
assert parsed["summary"] == "ok"
|
|
|
|
|
|
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
|