Files
ewoooc/tests/test_roi_report_service.py
OoO c300e496c5
All checks were successful
CD Pipeline / deploy (push) Successful in 55s
記錄 ROI 月報反饋區塊失敗
2026-05-13 10:00:04 +08:00

154 lines
5.2 KiB
Python

"""
tests/test_roi_report_service.py
─────────────────────────────────────────────────────────────────
Operation Ollama-First v5.0 / Phase 24 — ROI 月報生成器驗證
"""
from datetime import datetime
import logging
from unittest.mock import patch
import pytest
def test_baseline_constants():
"""戰前 baseline 應有 5 個必要欄位"""
from services.roi_report_service import BASELINE
required = {'gemini_monthly_tokens', 'gemini_monthly_cost_usd',
'nim_monthly_tokens', 'ollama_monthly_tokens',
'total_monthly_cost_usd'}
assert required.issubset(BASELINE.keys())
assert BASELINE['gemini_monthly_tokens'] == 50_000_000
assert BASELINE['gemini_monthly_cost_usd'] == 20.0
def test_last_month_range_january():
"""1 月跑時上月應為去年 12 月"""
from services.roi_report_service import _last_month_range
today = datetime(2026, 1, 5)
last_start, this_start = _last_month_range(today)
assert last_start == datetime(2025, 12, 1)
assert this_start == datetime(2026, 1, 1)
def test_last_month_range_normal_month():
from services.roi_report_service import _last_month_range
today = datetime(2026, 5, 15)
last_start, this_start = _last_month_range(today)
assert last_start == datetime(2026, 4, 1)
assert this_start == datetime(2026, 5, 1)
def test_render_roi_report_with_savings():
"""戰役後成功攔截 Gemini → 訊息含「攔截」+ 達標 ✅"""
from services.roi_report_service import render_roi_report
stats = {
'period_label': '2026年04月',
'period_start': datetime(2026, 4, 1),
'period_end': datetime(2026, 5, 1),
'ai_total_tokens': 80_000_000,
'ai_total_cost': 12.0,
'ai_total_calls': 5000,
'ollama_calls': 4000,
'gemini_calls': 800,
'claude_calls': 100,
'nim_calls': 100,
'rag_hit_calls': 200,
'cache_hit_calls': 150,
'gemini_tokens': 35_000_000, # 戰前 50M → 省 15M (30%)
'gemini_cost': 14.0, # 戰前 $20 → 省 $6
'claude_cost': 5.0,
'mcp_total': 180,
'mcp_cache_hits': 50,
'rag_total': 800,
'rag_saved': 250, # 31% RAG 攔截率
'rag_avg_feedback': 4.2,
}
msg = render_roi_report(stats)
assert '2026年04月' in msg
assert '攔截' in msg
assert '15,000,000' in msg or '15M' in msg or '15,000' in msg
assert '$6' in msg or '6.00' in msg
assert '' in msg # Gemini -23.5% 目標達標
def test_render_roi_report_below_target():
"""未達 -23.5% 目標 → 訊息含 ⚠️"""
from services.roi_report_service import render_roi_report
stats = {
'period_label': '2026年04月',
'period_start': datetime(2026, 4, 1),
'period_end': datetime(2026, 5, 1),
'ai_total_tokens': 80_000_000, 'ai_total_cost': 18.0, 'ai_total_calls': 5000,
'ollama_calls': 1000, 'gemini_calls': 3000, 'claude_calls': 500, 'nim_calls': 500,
'rag_hit_calls': 50, 'cache_hit_calls': 30,
'gemini_tokens': 45_000_000, # 戰前 50M → 只省 10% < 23%
'gemini_cost': 18.0, 'claude_cost': 5.0,
'mcp_total': 100, 'mcp_cache_hits': 10,
'rag_total': 200, 'rag_saved': 30, 'rag_avg_feedback': 3.5,
}
msg = render_roi_report(stats)
assert '⚠️' in msg # 未達標標記
def test_render_empty_stats():
"""空 stats → 預設失敗訊息"""
from services.roi_report_service import render_roi_report
assert '⚠️' in render_roi_report({})
def test_render_roi_report_logs_feedback_block_failure(monkeypatch, caplog):
"""反饋趨勢附加區塊失敗時,月報主體仍應產生並留下可追蹤 log。"""
from services.roi_report_service import render_roi_report
import services.feedback_quality_tracker as fqt
def _raise_feedback_error(days=30):
raise RuntimeError("feedback trend unavailable")
monkeypatch.setattr(fqt, "compute_caller_quality_trend", _raise_feedback_error)
caplog.set_level(logging.WARNING, logger="services.roi_report_service")
msg = render_roi_report({
'period_label': '2026年04月',
'gemini_tokens': 35_000_000,
'gemini_cost': 14.0,
'ollama_calls': 4000,
'gemini_calls': 800,
'claude_calls': 100,
'claude_cost': 5.0,
'nim_calls': 100,
'rag_total': 800,
'rag_saved': 250,
'rag_avg_feedback': 4.2,
'mcp_total': 180,
'mcp_cache_hits': 50,
'cache_hit_calls': 150,
})
assert 'ROI 月報 2026年04月' in msg
assert '反饋趨勢查詢失敗' in caplog.text
def test_query_last_month_stats_db_fail_returns_empty(monkeypatch):
"""DB 失敗應回 {} 不 raise"""
from services.roi_report_service import query_last_month_stats
class _BrokenSession:
def execute(self, *a, **kw):
raise RuntimeError('DB connection lost')
def close(self):
pass
import services.roi_report_service as ros
monkeypatch.setattr('database.manager.get_session', lambda: _BrokenSession())
result = ros.query_last_month_stats()
assert result == {}