82 lines
2.7 KiB
Python
82 lines
2.7 KiB
Python
"""中转模型列表（cloud/relay.py）单测：只用注入的取数函数，不联网、
|
|
不碰真 key。"""
|
|
|
|
import os
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
try:
|
|
from cloud.relay import RelayError, get_relay_models
|
|
from cloud.relay import _reset_cache_for_tests
|
|
except ImportError:
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
|
from cloud.relay import RelayError, get_relay_models
|
|
from cloud.relay import _reset_cache_for_tests
|
|
|
|
|
|
class RelayTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
_reset_cache_for_tests()
|
|
self.env = mock.patch.dict(os.environ, {}, clear=False)
|
|
self.env.start()
|
|
os.environ.pop("XCM_API_KEY", None)
|
|
|
|
def tearDown(self):
|
|
self.env.stop()
|
|
_reset_cache_for_tests()
|
|
|
|
def test_parses_ids_and_dedups(self):
|
|
calls = []
|
|
|
|
def fetcher(url, headers, timeout):
|
|
calls.append((url, headers))
|
|
return {"data": [{"id": "gpt-6.1-sol"}, {"id": "gpt-6-sol"},
|
|
{"id": "gpt-6.1-sol"}, {"nope": 1}]}
|
|
models, cached, _ = get_relay_models(fetcher=fetcher)
|
|
self.assertEqual(models, ["gpt-6.1-sol", "gpt-6-sol"])
|
|
self.assertFalse(cached)
|
|
self.assertTrue(calls[0][0].endswith("/models"))
|
|
# 注入路径不带 key：响应与请求头里都不应出现密钥字样。
|
|
self.assertNotIn("Authorization", calls[0][1])
|
|
self.assertEqual(len(calls), 1)
|
|
|
|
def test_cache_hit_skips_second_fetch(self):
|
|
calls = []
|
|
|
|
def fetcher(url, headers, timeout):
|
|
calls.append(url)
|
|
return {"data": [{"id": "m1"}]}
|
|
models1, cached1, _ = get_relay_models(fetcher=fetcher)
|
|
models2, cached2, _ = get_relay_models(fetcher=fetcher)
|
|
self.assertEqual(models1, models2)
|
|
self.assertFalse(cached1)
|
|
self.assertTrue(cached2)
|
|
self.assertEqual(len(calls), 1)
|
|
|
|
def test_cache_expires_after_ttl(self):
|
|
calls = []
|
|
|
|
def fetcher(url, headers, timeout):
|
|
calls.append(url)
|
|
return {"data": [{"id": "m1"}]}
|
|
get_relay_models(fetcher=fetcher, now=1000.0)
|
|
_, cached, _ = get_relay_models(fetcher=fetcher, now=1000.0 + 3601)
|
|
self.assertFalse(cached)
|
|
self.assertEqual(len(calls), 2)
|
|
|
|
def test_missing_key_without_fetcher_raises(self):
|
|
with self.assertRaises(RelayError):
|
|
get_relay_models()
|
|
|
|
def test_bad_response_raises(self):
|
|
with self.assertRaises(RelayError):
|
|
get_relay_models(fetcher=lambda u, h, t: {"nope": []})
|
|
with self.assertRaises(RelayError):
|
|
get_relay_models(fetcher=lambda u, h, t: {"data": "x"})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|