94 lines
3.2 KiB
Python
94 lines
3.2 KiB
Python
# apps/api/tests/test_ai_router_feedback.py | 2026-04-20 @ Asia/Taipei
|
||
"""Task A8: AIRouter.feedback_from_aider_events read-only aggregation test."""
|
||
import pytest
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
from src.services.ai_router import AIRouter
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_feedback_aggregates_by_model(monkeypatch):
|
||
stats = [
|
||
{"repo": "awoooi", "model": "elephant-alpha", "total": 10,
|
||
"errors": 2, "success_rate": 0.8},
|
||
{"repo": "awoooi", "model": "gemini-pro", "total": 5,
|
||
"errors": 0, "success_rate": 1.0},
|
||
]
|
||
|
||
class FakeRepo:
|
||
def __init__(self, sess): pass
|
||
async def model_stats_since(self, days): return stats
|
||
|
||
class FakeSession:
|
||
async def __aenter__(self): return self
|
||
async def __aexit__(self, *a): return False
|
||
|
||
monkeypatch.setattr("src.services.ai_router.get_session_factory",
|
||
lambda: (lambda: FakeSession()), raising=False)
|
||
monkeypatch.setattr("src.repositories.aider_event_repository.AiderEventRepository",
|
||
FakeRepo)
|
||
|
||
r = AIRouter()
|
||
out = await r.feedback_from_aider_events(days=7)
|
||
assert out["elephant-alpha"] == 0.8
|
||
assert out["gemini-pro"] == 1.0
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_feedback_filters_by_repo(monkeypatch):
|
||
stats = [
|
||
{"repo": "awoooi", "model": "elephant-alpha", "total": 5,
|
||
"errors": 1, "success_rate": 0.8},
|
||
{"repo": "other-repo", "model": "elephant-alpha", "total": 3,
|
||
"errors": 3, "success_rate": 0.0},
|
||
]
|
||
|
||
class FakeRepo:
|
||
def __init__(self, sess): pass
|
||
async def model_stats_since(self, days): return stats
|
||
|
||
class FakeSession:
|
||
async def __aenter__(self): return self
|
||
async def __aexit__(self, *a): return False
|
||
|
||
monkeypatch.setattr("src.services.ai_router.get_session_factory",
|
||
lambda: (lambda: FakeSession()), raising=False)
|
||
monkeypatch.setattr("src.repositories.aider_event_repository.AiderEventRepository",
|
||
FakeRepo)
|
||
|
||
r = AIRouter()
|
||
out = await r.feedback_from_aider_events(repo="awoooi", days=7)
|
||
assert out == {"elephant-alpha": 0.8} # other-repo 過濾掉
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_feedback_returns_empty_on_db_failure(monkeypatch):
|
||
def fail_sf():
|
||
raise RuntimeError("DB unavailable")
|
||
|
||
monkeypatch.setattr("src.services.ai_router.get_session_factory",
|
||
fail_sf, raising=False)
|
||
|
||
r = AIRouter()
|
||
out = await r.feedback_from_aider_events(days=7)
|
||
assert out == {} # 降級為空 dict，caller 不該崩
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_feedback_handles_empty_stats(monkeypatch):
|
||
class FakeRepo:
|
||
def __init__(self, sess): pass
|
||
async def model_stats_since(self, days): return []
|
||
|
||
class FakeSession:
|
||
async def __aenter__(self): return self
|
||
async def __aexit__(self, *a): return False
|
||
|
||
monkeypatch.setattr("src.services.ai_router.get_session_factory",
|
||
lambda: (lambda: FakeSession()), raising=False)
|
||
monkeypatch.setattr("src.repositories.aider_event_repository.AiderEventRepository",
|
||
FakeRepo)
|
||
|
||
r = AIRouter()
|
||
out = await r.feedback_from_aider_events()
|
||
assert out == {}
|