332 lines
14 KiB
Python
332 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
tests/test_anthropic_service.py
|
||
services/anthropic_service.py 單元測試 — Operation Ollama-First v5.0 Phase 7
|
||
|
||
測試紀律（Phase 7 spec）：
|
||
- generate 正常路徑：cache_creation + cache_read 解析正確
|
||
- generate prompt cache：cache_system=True 時 system 加 cache_control
|
||
- ANTHROPIC_API_KEY 未設定時 is_available() == False
|
||
- SDK ImportError 時不爆（log.error，is_available()=False）
|
||
- SDK 例外時 generate 回 success=False 不 raise
|
||
- cache_hit property 邏輯（cache_read_tokens > 0 → True）
|
||
- check_connection 正常與失敗
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import sys
|
||
import logging
|
||
from types import SimpleNamespace
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import pytest
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Helpers：仿造 anthropic SDK Response 結構
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def _fake_anthropic_response(
|
||
text: str = "ok",
|
||
input_tokens: int = 100,
|
||
output_tokens: int = 50,
|
||
cache_creation_input_tokens: int = 0,
|
||
cache_read_input_tokens: int = 0,
|
||
model: str = "claude-opus-4-7",
|
||
):
|
||
"""模擬 anthropic.types.Message"""
|
||
block = SimpleNamespace(text=text, type="text")
|
||
usage = SimpleNamespace(
|
||
input_tokens=input_tokens,
|
||
output_tokens=output_tokens,
|
||
cache_creation_input_tokens=cache_creation_input_tokens,
|
||
cache_read_input_tokens=cache_read_input_tokens,
|
||
)
|
||
return SimpleNamespace(content=[block], usage=usage, model=model)
|
||
|
||
|
||
@pytest.fixture
|
||
def mock_sdk(monkeypatch):
|
||
"""模擬 anthropic SDK：注入 fake module 到 sys.modules，
|
||
讓 AnthropicService._init_client 走 import anthropic 成功路徑。"""
|
||
fake_anthropic = MagicMock()
|
||
fake_client = MagicMock()
|
||
fake_anthropic.Anthropic.return_value = fake_client
|
||
monkeypatch.setitem(sys.modules, 'anthropic', fake_anthropic)
|
||
return fake_anthropic, fake_client
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# is_available() 測試
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def test_is_available_false_when_no_api_key(monkeypatch):
|
||
"""ANTHROPIC_API_KEY 未設 → is_available() False"""
|
||
monkeypatch.delenv('ANTHROPIC_API_KEY', raising=False)
|
||
# reload 確保模組層 ANTHROPIC_API_KEY 重新讀取
|
||
import importlib
|
||
import services.anthropic_service as svc_mod
|
||
importlib.reload(svc_mod)
|
||
svc = svc_mod.AnthropicService(api_key='')
|
||
assert svc.is_available() is False
|
||
|
||
|
||
def test_is_available_true_when_sdk_ready(mock_sdk):
|
||
"""API key + SDK 都有 → is_available() True"""
|
||
from services.anthropic_service import AnthropicService
|
||
svc = AnthropicService(api_key='sk-ant-test')
|
||
assert svc.is_available() is True
|
||
|
||
|
||
def test_is_available_logs_cost_throttle_failure(mock_sdk, monkeypatch, caplog):
|
||
"""cost_throttle 檢查故障時仍維持 Claude 可用，但必須留下診斷 log。"""
|
||
from services.anthropic_service import AnthropicService
|
||
import services.cost_throttle_service as cost_throttle_service
|
||
|
||
def _raise_cost_throttle_error(provider):
|
||
raise RuntimeError("cost throttle unavailable")
|
||
|
||
monkeypatch.setattr(
|
||
cost_throttle_service,
|
||
"is_provider_throttled",
|
||
_raise_cost_throttle_error,
|
||
)
|
||
caplog.set_level(logging.WARNING, logger="services.anthropic_service")
|
||
|
||
svc = AnthropicService(api_key='sk-ant-test')
|
||
|
||
assert svc.is_available() is True
|
||
assert "cost_throttle check failed" in caplog.text
|
||
|
||
|
||
def test_is_available_false_on_import_error(monkeypatch):
|
||
"""SDK 未安裝（ImportError）→ is_available() False，不 raise"""
|
||
# 移除 anthropic 模組讓 import 失敗
|
||
monkeypatch.setitem(sys.modules, 'anthropic', None)
|
||
from services.anthropic_service import AnthropicService
|
||
svc = AnthropicService(api_key='sk-ant-test')
|
||
assert svc.is_available() is False
|
||
|
||
|
||
def test_is_available_false_on_init_exception(monkeypatch):
|
||
"""SDK 初始化拋例外 → is_available() False，不 raise"""
|
||
fake_anthropic = MagicMock()
|
||
fake_anthropic.Anthropic.side_effect = RuntimeError("auth failed")
|
||
monkeypatch.setitem(sys.modules, 'anthropic', fake_anthropic)
|
||
|
||
from services.anthropic_service import AnthropicService
|
||
svc = AnthropicService(api_key='sk-ant-test')
|
||
assert svc.is_available() is False
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# generate() 正常路徑
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def test_generate_success_basic(mock_sdk):
|
||
"""generate 正常路徑：tokens 正確解析"""
|
||
fake_anthropic, fake_client = mock_sdk
|
||
fake_client.messages.create.return_value = _fake_anthropic_response(
|
||
text="hello world",
|
||
input_tokens=120,
|
||
output_tokens=40,
|
||
)
|
||
|
||
from services.anthropic_service import AnthropicService
|
||
svc = AnthropicService(api_key='sk-ant-test')
|
||
|
||
resp = svc.generate(prompt="say hi", max_tokens=100)
|
||
assert resp.success is True
|
||
assert resp.content == "hello world"
|
||
assert resp.input_tokens == 120
|
||
assert resp.output_tokens == 40
|
||
assert resp.cache_creation_tokens == 0
|
||
assert resp.cache_read_tokens == 0
|
||
assert resp.cache_hit is False
|
||
assert resp.duration_ms >= 0
|
||
assert resp.error is None
|
||
|
||
|
||
def test_generate_with_cache_creation_and_read(mock_sdk):
|
||
"""generate 解析 cache_creation_input_tokens / cache_read_input_tokens"""
|
||
fake_anthropic, fake_client = mock_sdk
|
||
fake_client.messages.create.return_value = _fake_anthropic_response(
|
||
cache_creation_input_tokens=500,
|
||
cache_read_input_tokens=2000,
|
||
)
|
||
|
||
from services.anthropic_service import AnthropicService
|
||
svc = AnthropicService(api_key='sk-ant-test')
|
||
|
||
resp = svc.generate(prompt="reuse", system_prompt="stable system")
|
||
assert resp.success is True
|
||
assert resp.cache_creation_tokens == 500
|
||
assert resp.cache_read_tokens == 2000
|
||
assert resp.cache_hit is True
|
||
|
||
|
||
def test_generate_with_cache_system_adds_cache_control(mock_sdk):
|
||
"""cache_system=True 時 system_prompt 帶 ephemeral cache_control"""
|
||
fake_anthropic, fake_client = mock_sdk
|
||
fake_client.messages.create.return_value = _fake_anthropic_response()
|
||
|
||
from services.anthropic_service import AnthropicService
|
||
svc = AnthropicService(api_key='sk-ant-test')
|
||
|
||
svc.generate(
|
||
prompt="user",
|
||
system_prompt="my-system",
|
||
cache_system=True,
|
||
)
|
||
_, kwargs = fake_client.messages.create.call_args
|
||
assert isinstance(kwargs['system'], list)
|
||
assert kwargs['system'][0]['type'] == 'text'
|
||
assert kwargs['system'][0]['text'] == 'my-system'
|
||
assert kwargs['system'][0]['cache_control'] == {"type": "ephemeral"}
|
||
|
||
|
||
def test_generate_without_cache_system_uses_string(mock_sdk):
|
||
"""cache_system=False 時 system 為純字串"""
|
||
fake_anthropic, fake_client = mock_sdk
|
||
fake_client.messages.create.return_value = _fake_anthropic_response()
|
||
|
||
from services.anthropic_service import AnthropicService
|
||
svc = AnthropicService(api_key='sk-ant-test')
|
||
|
||
svc.generate(prompt="user", system_prompt="my-system", cache_system=False)
|
||
_, kwargs = fake_client.messages.create.call_args
|
||
assert kwargs['system'] == 'my-system'
|
||
|
||
|
||
def test_generate_without_system_prompt(mock_sdk):
|
||
"""無 system_prompt 時 kwargs 不含 system"""
|
||
fake_anthropic, fake_client = mock_sdk
|
||
fake_client.messages.create.return_value = _fake_anthropic_response()
|
||
|
||
from services.anthropic_service import AnthropicService
|
||
svc = AnthropicService(api_key='sk-ant-test')
|
||
|
||
svc.generate(prompt="user")
|
||
_, kwargs = fake_client.messages.create.call_args
|
||
assert 'system' not in kwargs
|
||
|
||
|
||
def test_generate_passes_temperature_and_max_tokens(mock_sdk):
|
||
"""temperature / max_tokens / model 正確傳給 SDK"""
|
||
fake_anthropic, fake_client = mock_sdk
|
||
fake_client.messages.create.return_value = _fake_anthropic_response()
|
||
|
||
from services.anthropic_service import AnthropicService
|
||
svc = AnthropicService(api_key='sk-ant-test')
|
||
|
||
svc.generate(
|
||
prompt="x", model="claude-sonnet-4-6",
|
||
max_tokens=2048, temperature=0.5,
|
||
)
|
||
_, kwargs = fake_client.messages.create.call_args
|
||
assert kwargs['model'] == 'claude-sonnet-4-6'
|
||
assert kwargs['max_tokens'] == 2048
|
||
assert kwargs['temperature'] == 0.5
|
||
assert kwargs['messages'] == [{"role": "user", "content": "x"}]
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# generate() 失敗路徑
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def test_generate_returns_failure_when_no_client(monkeypatch):
|
||
"""無 API key → generate 回 success=False 不 raise"""
|
||
monkeypatch.delenv('ANTHROPIC_API_KEY', raising=False)
|
||
from services.anthropic_service import AnthropicService
|
||
svc = AnthropicService(api_key='')
|
||
|
||
resp = svc.generate(prompt="x")
|
||
assert resp.success is False
|
||
assert resp.content == ""
|
||
assert "not initialized" in (resp.error or "")
|
||
|
||
|
||
def test_generate_handles_sdk_exception(mock_sdk):
|
||
"""SDK 拋例外 → generate 回 success=False，error 含 type+msg"""
|
||
fake_anthropic, fake_client = mock_sdk
|
||
fake_client.messages.create.side_effect = RuntimeError("rate limit")
|
||
|
||
from services.anthropic_service import AnthropicService
|
||
svc = AnthropicService(api_key='sk-ant-test')
|
||
|
||
resp = svc.generate(prompt="x")
|
||
assert resp.success is False
|
||
assert "RuntimeError" in (resp.error or "")
|
||
assert "rate limit" in (resp.error or "")
|
||
assert resp.duration_ms >= 0
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# ClaudeResponse cache_hit property
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def test_cache_hit_property():
|
||
from services.anthropic_service import ClaudeResponse
|
||
assert ClaudeResponse(success=True, content="x", model="m",
|
||
cache_read_tokens=0).cache_hit is False
|
||
assert ClaudeResponse(success=True, content="x", model="m",
|
||
cache_read_tokens=1).cache_hit is True
|
||
assert ClaudeResponse(success=True, content="x", model="m",
|
||
cache_read_tokens=10000).cache_hit is True
|
||
|
||
|
||
def test_total_tokens_property():
|
||
from services.anthropic_service import ClaudeResponse
|
||
r = ClaudeResponse(success=True, content="x", model="m",
|
||
input_tokens=100, output_tokens=50)
|
||
assert r.total_tokens == 150
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# check_connection
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def test_check_connection_success(mock_sdk):
|
||
fake_anthropic, fake_client = mock_sdk
|
||
fake_client.messages.create.return_value = _fake_anthropic_response(text="pong")
|
||
|
||
from services.anthropic_service import AnthropicService
|
||
svc = AnthropicService(api_key='sk-ant-test')
|
||
assert svc.check_connection() is True
|
||
|
||
|
||
def test_check_connection_fail_no_client(monkeypatch):
|
||
monkeypatch.delenv('ANTHROPIC_API_KEY', raising=False)
|
||
from services.anthropic_service import AnthropicService
|
||
svc = AnthropicService(api_key='')
|
||
assert svc.check_connection() is False
|
||
|
||
|
||
def test_check_connection_fail_on_sdk_error(mock_sdk):
|
||
fake_anthropic, fake_client = mock_sdk
|
||
fake_client.messages.create.side_effect = RuntimeError("boom")
|
||
|
||
from services.anthropic_service import AnthropicService
|
||
svc = AnthropicService(api_key='sk-ant-test')
|
||
assert svc.check_connection() is False
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# COST_TABLE 整合（確認 ai_call_logger 認得 claude 模型）
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def test_cost_table_has_claude_models():
|
||
from services.ai_call_logger import COST_TABLE, _calc_cost
|
||
assert 'claude-opus-4-7' in COST_TABLE
|
||
assert 'claude-sonnet-4-6' in COST_TABLE
|
||
assert 'claude-haiku-4-5' in COST_TABLE
|
||
# opus 1M in/1M out 應為 15 + 75 = 90 USD
|
||
assert abs(_calc_cost('claude-opus-4-7', 1_000_000, 1_000_000) - 90.0) < 1e-6
|
||
# haiku 1M in/1M out 應為 0.8 + 4.0 = 4.8 USD
|
||
assert abs(_calc_cost('claude-haiku-4-5', 1_000_000, 1_000_000) - 4.8) < 1e-6
|