41 lines
1.4 KiB
Python
41 lines
1.4 KiB
Python
"""monitor 调度链路测试：api_key base64 解码等关键行为。"""
|
|
|
|
import base64
|
|
|
|
from app.monitor import _split
|
|
|
|
|
|
def _row(**overrides):
|
|
row = {
|
|
"id": 1, "platform_id": 2, "name": "a",
|
|
"api_key": base64.b64encode(b"sk-123").decode(),
|
|
"threshold": 0, "enabled": 1, "alert_armed": 1,
|
|
"last_balance": None, "last_status": "pending", "last_error": "",
|
|
"last_check_at": None, "note": "",
|
|
"provider_id": "deepseek", "platform_name": "P", "currency": "USD", "icon": "",
|
|
"interval_seconds": None, "retry_count": None, "timeout_seconds": None,
|
|
"platform_enabled": 1, "platform_note": "",
|
|
}
|
|
row.update(overrides)
|
|
return row
|
|
|
|
|
|
class TestSplit:
|
|
def test_api_key_decoded(self):
|
|
"""监控链路必须使用解码后的明文 key，否则请求永远 401。"""
|
|
account, platform = _split(_row())
|
|
assert account["api_key"] == "sk-123"
|
|
|
|
def test_platform_fields_mapped(self):
|
|
account, platform = _split(_row())
|
|
assert platform["name"] == "P"
|
|
assert platform["id"] == 2
|
|
assert platform["enabled"] == 1
|
|
assert platform["provider_id"] == "deepseek"
|
|
assert platform["currency"] == "USD"
|
|
|
|
def test_plain_key_passthrough(self):
|
|
"""未编码的 key（历史数据）原样使用，不抛错。"""
|
|
account, _ = _split(_row(api_key="sk-plain"))
|
|
assert account["api_key"] == "sk-plain"
|