143 lines
3.9 KiB
Python
143 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src.services import local_code_review_service as review_module
|
|
from src.services.local_code_review_service import LocalCodeReviewService
|
|
|
|
|
|
class _FakeResponse:
|
|
status_code = 200
|
|
|
|
def json(self) -> dict[str, str]:
|
|
return {"response": "✅ Push 品質正常"}
|
|
|
|
|
|
class _FakeClient:
|
|
def __init__(self, *, fail: bool = False) -> None:
|
|
self.fail = fail
|
|
self.posted_urls: list[str] = []
|
|
|
|
async def post(self, url: str, **kwargs: Any) -> _FakeResponse:
|
|
self.posted_urls.append(url)
|
|
if self.fail:
|
|
raise httpx.TimeoutException("timeout")
|
|
return _FakeResponse()
|
|
|
|
|
|
async def _noop_save(*args: Any, **kwargs: Any) -> None:
|
|
return None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_large_pr_uses_gcp_b_ollama_when_gemini_fallback_disabled(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
review_module.settings,
|
|
"LOCAL_CODE_REVIEW_ALLOW_GEMINI_FALLBACK",
|
|
False,
|
|
)
|
|
monkeypatch.setattr(
|
|
review_module,
|
|
"resolve_ollama_endpoint",
|
|
lambda workload_type: "http://gcp-b:11436",
|
|
)
|
|
|
|
client = _FakeClient()
|
|
service = LocalCodeReviewService()
|
|
|
|
async def _get_http() -> _FakeClient:
|
|
return client
|
|
|
|
async def _fail_gemini(*args: Any, **kwargs: Any) -> None:
|
|
raise AssertionError("Gemini fallback should stay disabled")
|
|
|
|
monkeypatch.setattr(service, "_get_http", _get_http)
|
|
monkeypatch.setattr(service, "_save_to_db", _noop_save)
|
|
monkeypatch.setattr(service, "_review_with_gemini", _fail_gemini)
|
|
|
|
result = await service.review_pr(
|
|
pr_id="pr-1",
|
|
repo="wooo/awoooi",
|
|
title="large diff",
|
|
diff="x" * (60 * 1024),
|
|
)
|
|
|
|
assert result is not None
|
|
assert result["provider"] == "ollama"
|
|
assert client.posted_urls == ["http://gcp-b:11436/api/generate"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ollama_failure_does_not_fall_back_to_gemini_by_default(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
review_module.settings,
|
|
"LOCAL_CODE_REVIEW_ALLOW_GEMINI_FALLBACK",
|
|
False,
|
|
)
|
|
monkeypatch.setattr(
|
|
review_module,
|
|
"resolve_ollama_endpoint",
|
|
lambda workload_type: "http://gcp-b:11436",
|
|
)
|
|
|
|
client = _FakeClient(fail=True)
|
|
service = LocalCodeReviewService()
|
|
|
|
async def _get_http() -> _FakeClient:
|
|
return client
|
|
|
|
async def _fail_gemini(*args: Any, **kwargs: Any) -> None:
|
|
raise AssertionError("Gemini fallback should stay disabled")
|
|
|
|
monkeypatch.setattr(service, "_get_http", _get_http)
|
|
monkeypatch.setattr(service, "_save_to_db", _noop_save)
|
|
monkeypatch.setattr(service, "_review_with_gemini", _fail_gemini)
|
|
|
|
result = await service.review_pr(
|
|
pr_id="pr-2",
|
|
repo="wooo/awoooi",
|
|
title="ollama unavailable",
|
|
diff="small diff",
|
|
)
|
|
|
|
assert result is not None
|
|
assert result["provider"] == "ollama_unavailable"
|
|
assert result["cloud_fallback_skipped"] is True
|
|
assert client.posted_urls == ["http://gcp-b:11436/api/generate"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gemini_fallback_requires_explicit_flag(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
review_module.settings,
|
|
"LOCAL_CODE_REVIEW_ALLOW_GEMINI_FALLBACK",
|
|
True,
|
|
)
|
|
|
|
service = LocalCodeReviewService()
|
|
|
|
async def _gemini_result(*args: Any, **kwargs: Any) -> dict[str, Any]:
|
|
return {"review_text": "ok", "issues_count": 0, "model": "gemini", "provider": "gemini"}
|
|
|
|
monkeypatch.setattr(service, "_save_to_db", _noop_save)
|
|
monkeypatch.setattr(service, "_review_with_gemini", _gemini_result)
|
|
|
|
result = await service.review_pr(
|
|
pr_id="pr-3",
|
|
repo="wooo/awoooi",
|
|
title="large diff explicit cloud",
|
|
diff="x" * (60 * 1024),
|
|
)
|
|
|
|
assert result is not None
|
|
assert result["provider"] == "gemini"
|