147 lines
5.8 KiB
Python
147 lines
5.8 KiB
Python
"""
|
||
tests/test_caller_registry.py
|
||
─────────────────────────────────────────────────────────────────
|
||
Operation Ollama-First v5.0 / Phase 16 — caller_registry 集中管理驗證
|
||
|
||
驗證面：
|
||
T1. CALLER_REGISTRY 含 ADR-028 列舉的 30+ caller
|
||
T2. is_known_caller 對白名單回 True / 未知回 False
|
||
T3. assert_known_caller(strict=False) → log warning 不 raise
|
||
T4. assert_known_caller(strict=True) → 未知 raise ValueError
|
||
T5. list_callers_by_service 分組正確
|
||
T6. ai_call_logger 整合：未知 caller 不阻擋（log warning）
|
||
"""
|
||
|
||
import logging
|
||
|
||
import pytest
|
||
|
||
|
||
def test_registry_contains_core_callers():
|
||
"""ADR-028 白名單核心 caller 必在 registry"""
|
||
from services.llm_caller_registry import CALLER_REGISTRY
|
||
|
||
must_have = {
|
||
# Hermes
|
||
'hermes_analyst', 'hermes_intent',
|
||
# OpenClaw
|
||
'openclaw_daily', 'openclaw_weekly', 'openclaw_monthly',
|
||
'openclaw_meta', 'openclaw_qa',
|
||
'openclaw_qa_gemini_fallback', 'openclaw_qa_nim',
|
||
'openclaw_daily_gemini_fallback', 'openclaw_weekly_gemini_fallback',
|
||
'openclaw_monthly_gemini_fallback', 'openclaw_meta_gemini_fallback',
|
||
'openclaw_daily_insight_gemini_fallback',
|
||
'openclaw_daily_nim', 'openclaw_weekly_nim',
|
||
'openclaw_monthly_nim', 'openclaw_meta_nim',
|
||
'openclaw_daily_insight_nim',
|
||
# MCP
|
||
'mcp_l1_grounding', 'mcp_collector',
|
||
# Code Review
|
||
'code_review_hermes', 'code_review_openclaw', 'code_review_elephant',
|
||
# NemoTron / EA
|
||
'nemotron_dispatch', 'ea_engine',
|
||
# PPT
|
||
'ppt_gemini', 'ppt_ollama', 'ppt_nim', 'ppt_vision',
|
||
# KM Embedding
|
||
'km_embedding_worker', 'km_embedding_realtime',
|
||
# Sales / Trend
|
||
'sales_copy', 'trend_match', 'trend_qa', 'product_insights',
|
||
# Bot
|
||
'openclaw_bot_main', 'openclaw_bot_gemini', 'openclaw_bot_nim',
|
||
'openclaw_bot_image', 'openclaw_bot_image_gemini',
|
||
}
|
||
|
||
missing = must_have - CALLER_REGISTRY
|
||
assert not missing, f"registry 缺 {len(missing)} 個關鍵 caller: {missing}"
|
||
|
||
|
||
def test_is_known_caller():
|
||
from services.llm_caller_registry import is_known_caller
|
||
|
||
assert is_known_caller('hermes_analyst') is True
|
||
assert is_known_caller('openclaw_qa') is True
|
||
assert is_known_caller('openclaw_qa_gemini_fallback') is True
|
||
assert is_known_caller('openclaw_qa_nim') is True
|
||
assert is_known_caller('not_a_real_caller') is False
|
||
assert is_known_caller('') is False
|
||
assert is_known_caller('GeMiNi_Caller_Wrong_Case') is False
|
||
|
||
|
||
def test_assert_known_caller_strict_false_only_warns(caplog):
|
||
"""strict=False（預設）→ 不在 registry 只 log warning"""
|
||
from services.llm_caller_registry import assert_known_caller
|
||
|
||
with caplog.at_level(logging.WARNING):
|
||
assert_known_caller('totally_made_up_caller', strict=False)
|
||
# 應有 warning
|
||
warnings = [r for r in caplog.records if r.levelno >= logging.WARNING]
|
||
assert len(warnings) >= 1
|
||
assert 'unknown caller' in warnings[0].message.lower() or \
|
||
'totally_made_up_caller' in warnings[0].message
|
||
|
||
|
||
def test_assert_known_caller_strict_true_raises():
|
||
from services.llm_caller_registry import assert_known_caller
|
||
|
||
with pytest.raises(ValueError, match='unknown caller'):
|
||
assert_known_caller('definitely_not_real', strict=True)
|
||
|
||
|
||
def test_assert_known_caller_passes_for_real_caller():
|
||
"""合法 caller → 不 raise / 不 warn"""
|
||
from services.llm_caller_registry import assert_known_caller
|
||
# 不該 raise
|
||
assert_known_caller('hermes_analyst', strict=True)
|
||
assert_known_caller('openclaw_qa', strict=False)
|
||
|
||
|
||
def test_list_callers_by_service_structure():
|
||
from services.llm_caller_registry import list_callers_by_service, CALLER_REGISTRY
|
||
|
||
grouped = list_callers_by_service()
|
||
expected_groups = {'hermes', 'openclaw', 'openclaw_bot', 'mcp', 'code_review',
|
||
'ppt', 'tg_bot', 'km_embedding', 'sales_trend', 'misc'}
|
||
assert expected_groups.issubset(grouped.keys())
|
||
|
||
# 每組至少 1 個
|
||
for group, callers in grouped.items():
|
||
assert isinstance(callers, list)
|
||
assert len(callers) >= 1, f"group '{group}' 是空的"
|
||
|
||
# hermes 至少含 hermes_analyst
|
||
assert 'hermes_analyst' in grouped['hermes']
|
||
|
||
|
||
def test_ai_call_logger_integration_does_not_block_unknown_caller(caplog):
|
||
"""ai_call_logger 收到 unknown caller 應 log warning 但不阻擋 context manager"""
|
||
import os
|
||
os.environ['AI_CALL_LOGGING_ENABLED'] = 'false' # 跳過 DB 寫入
|
||
try:
|
||
from services.ai_call_logger import log_ai_call
|
||
|
||
with caplog.at_level(logging.WARNING):
|
||
with log_ai_call(
|
||
caller='unknown_test_caller_xyz', # 故意不在 registry
|
||
provider='gcp_ollama',
|
||
model='hermes3:latest',
|
||
) as ctx:
|
||
ctx.set_tokens(input=100, output=50)
|
||
# context manager 不 raise，正常結束
|
||
# 應該有 warning（caller 不在 registry）
|
||
warnings = [r for r in caplog.records if r.levelno >= logging.WARNING]
|
||
assert any('unknown_test_caller' in r.message.lower() or
|
||
'unknown caller' in r.message.lower()
|
||
for r in warnings), \
|
||
f"應 warn unknown caller，實際 warnings: {[r.message for r in warnings]}"
|
||
finally:
|
||
os.environ.pop('AI_CALL_LOGGING_ENABLED', None)
|
||
|
||
|
||
def test_registry_is_immutable_frozenset():
|
||
"""CALLER_REGISTRY 是 frozenset 不可變動"""
|
||
from services.llm_caller_registry import CALLER_REGISTRY
|
||
|
||
assert isinstance(CALLER_REGISTRY, frozenset)
|
||
with pytest.raises(AttributeError):
|
||
CALLER_REGISTRY.add('attempted_mutation') # type: ignore
|