优化

backend/tests/unit/test_colg_official.py

@@ -0,0 +1,65 @@
+import asyncio
+
+from api.routes import colg
+
+
+class FakeResponse:
+    def __init__(self, region):
+        self.region = region
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, traceback):
+        return False
+
+    def raise_for_status(self):
+        return None
+
+    async def json(self, content_type=None):
+        await asyncio.sleep(0)
+        return {
+            'data': {
+                'godDealRank': [{'price': '100'}, {'price': '101'}],
+                'godTopDealRank': [{'price': '200'}, {'price': '201'}],
+            }
+        }
+
+
+class FakeSession:
+    def __init__(self, calls, **kwargs):
+        self.calls = calls
+        self.kwargs = kwargs
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, traceback):
+        return False
+
+    def post(self, url, data):
+        self.calls.append((url, data))
+        return FakeResponse(data['serverNo'])
+
+
+def test_fetch_official_gold_uses_async_http_with_isolated_forms(monkeypatch):
+    calls = []
+    sessions = []
+
+    def create_session(**kwargs):
+        session = FakeSession(calls, **kwargs)
+        sessions.append(session)
+        return session
+
+    monkeypatch.setattr(colg.aiohttp, 'ClientSession', create_session)
+
+    result = asyncio.run(colg._fetch_official_gold())
+
+    assert [data['serverNo'] for _, data in calls] == list(colg.OFFICIAL_GOLD_REGIONS)
+    assert all(url == colg.OFFICIAL_GOLD_URL for url, _ in calls)
+    assert len({id(data) for _, data in calls}) == len(colg.OFFICIAL_GOLD_REGIONS)
+    assert [item['region'] for item in result] == [f'跨{region}' for region in colg.OFFICIAL_GOLD_REGIONS]
+    assert result[0]['goldDealRank'] == [100, 101]
+    assert result[0]['goldTopDealRank'] == [200, 201]
+    assert sessions[0].kwargs['timeout'].total == 10
+    assert 'serverNo' not in colg.payload_helper

backend/tests/unit/test_redis_cache.py

@@ -0,0 +1,64 @@
+import asyncio
+import json
+
+from api.core.Redis import get_redis_info, get_redis_info_async
+from config.main import config
+from database.connect import AsyncRedisPool, RedisPool
+
+
+class FakeRedis:
+    def __init__(self, value=None, fail_write=False):
+        self.value = value
+        self.fail_write = fail_write
+        self.set_calls = []
+
+    def get(self, key):
+        return self.value
+
+    def set(self, key, value, **kwargs):
+        if self.fail_write:
+            raise RuntimeError('Redis unavailable')
+        self.set_calls.append((key, value, kwargs))
+
+
+class FakeAsyncRedis(FakeRedis):
+    async def get(self, key):
+        return self.value
+
+    async def set(self, key, value, **kwargs):
+        super().set(key, value, **kwargs)
+
+
+def test_shared_pool_keeps_resp2_compatibility():
+    assert RedisPool.connection_kwargs['protocol'] == 2
+    assert AsyncRedisPool.connection_kwargs['protocol'] == 2
+
+
+def test_sync_cache_sets_value_and_ttl_in_one_command(monkeypatch):
+    monkeypatch.setattr(config, 'DEBUG_MODE', False)
+    redis = FakeRedis()
+
+    result = get_redis_info(redis, 'key', lambda: {'value': 1}, expire=60, without_gzip=True)
+
+    assert result == {'value': 1}
+    assert redis.set_calls == [('key', json.dumps(result), {'ex': 60})]
+
+
+def test_sync_cache_write_failure_does_not_fail_request(monkeypatch):
+    monkeypatch.setattr(config, 'DEBUG_MODE', False)
+    redis = FakeRedis(fail_write=True)
+
+    assert get_redis_info(redis, 'key', lambda: {'value': 1}, without_gzip=True) == {'value': 1}
+
+
+def test_async_cache_awaits_async_redis_io(monkeypatch):
+    monkeypatch.setattr(config, 'DEBUG_MODE', False)
+    redis = FakeAsyncRedis()
+
+    async def load():
+        return {'value': 1}
+
+    result = asyncio.run(get_redis_info_async(redis, 'key', load, expire=60, without_gzip=True))
+
+    assert result == {'value': 1}
+    assert redis.set_calls == [('key', json.dumps(result), {'ex': 60})]