File size: 8,447 Bytes
d2ae3d7 fcd684b d2ae3d7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 | import asyncio
from datetime import datetime
import unittest
from unittest.mock import AsyncMock, patch
import httpx
from fastapi import APIRouter, FastAPI, Request
from client_source import current_client_source, resolve_client_source
from upload_routes import FaceAnalysisFormRoute
from test_ranking_consistency import database
class SourceIsolationTest(unittest.IsolatedAsyncioTestCase):
def setUp(self):
database._RANK_CACHE.clear()
database._LEADERBOARD_CACHE.clear()
async def test_rank_sql_and_cache_are_partitioned_for_both_scopes(self):
fetch = AsyncMock(side_effect=[{'t': 20, 'r': 4}, {'t': 10, 'r': 2},
{'t': 3, 'r': 1}, {'t': 2, 'r': 1}])
with patch.object(database, 'fetch_one', fetch):
android = await database.get_face_score_rank(88, 'CN', client_source='android')
wechat = await database.get_face_score_rank(88, 'CN', client_source='wechat')
cached = await database.get_face_score_rank(88, 'CN', client_source='android')
self.assertEqual(20, android['global_total'])
self.assertEqual(3, wechat['global_total'])
self.assertEqual(android, cached)
self.assertEqual(4, fetch.await_count)
for index, call in enumerate(fetch.await_args_list):
self.assertIn('client_source = %s', call.args[0])
self.assertEqual('android' if index < 2 else 'wechat', call.args[1][-1])
self.assertNotIn('nickname LIKE', call.args[0])
async def test_leaderboard_cache_and_device_enrichment_do_not_cross_sources(self):
async def fetch(sql, params):
if 'tpl_app_user_devices' in sql:
return [{'device_id': 'same', 'display_name': 'Android name'}]
return [{'id': 1 if params[-1] == 'android' else 2, 'score': 88,
'nickname': 'android_release_same', 'region': 'CN'}]
query = AsyncMock(side_effect=fetch)
with patch.object(database, 'fetch_all', query):
a = await database.get_face_score_leaderboard('CN', 50, 'same', 'android')
w = await database.get_face_score_leaderboard('CN', 50, 'same', 'wechat')
w2 = await database.get_face_score_leaderboard('CN', 50, 'same', 'wechat')
self.assertEqual(1, a['items'][0]['record_id'])
self.assertEqual(2, w['items'][0]['record_id'])
self.assertEqual(w, w2)
self.assertTrue(a['items'][0]['is_current_user'])
self.assertFalse(w['items'][0]['is_current_user'])
self.assertNotEqual('Android name', w['items'][0].get('display_name'))
self.assertEqual(3, query.await_count)
self.assertIn('candidate.client_source = %s', query.await_args_list[0].args[0])
self.assertIn('candidate.client_source = %s', query.await_args_list[2].args[0])
async def test_leaderboard_file_path_survives_serialization_and_cache(self):
from models import LeaderboardResponse
for source in ('android', 'wechat'):
for region in (None, 'CN'):
with self.subTest(source=source, region=region):
path = f'face/{source}/portrait.webp'
query = AsyncMock(return_value=[{'id': 7, 'score': 88,
'region': 'CN', 'file_path': path}])
with patch.object(database, 'fetch_all', query):
fresh = await database.get_face_score_leaderboard(region, 50, client_source=source)
cached = await database.get_face_score_leaderboard(region, 50, client_source=source)
self.assertEqual(1, query.await_count)
self.assertIn('candidate.file_path', query.await_args.args[0])
for result in (fresh, cached):
payload = LeaderboardResponse(**result).model_dump()
self.assertEqual(path, payload['items'][0]['file_path'])
async def test_image_and_device_writes_include_source(self):
execute = AsyncMock()
with patch.object(database, 'execute', execute):
for source in ('android', 'wechat'):
token = current_client_source.set(source)
try:
await database.upsert_image_record(file_path='face.webp', category='face', nickname='same',
score=88, is_cropped_face=True, size_bytes=1, last_modified=datetime.now(), bos_uploaded=True)
finally:
current_client_source.reset(token)
self.assertEqual(source, execute.await_args.args[1][-1])
self.assertIn("IF(client_source = 'unknown'", execute.await_args.args[0])
await database.upsert_device_record(device_id='device')
self.assertEqual('android', execute.await_args.args[1][-1])
async def test_concurrent_requests_and_background_tasks_keep_provenance(self):
app = FastAPI()
router = APIRouter(prefix='/facescore', route_class=FaceAnalysisFormRoute)
tasks = []
seen = []
@router.get('/app/rank')
@router.get('/wechat/rank')
async def endpoint(request: Request):
async def record():
await asyncio.sleep(.01)
seen.append(current_client_source.get())
tasks.append(asyncio.create_task(record()))
await asyncio.sleep(0)
return {'source': current_client_source.get()}
app.include_router(router)
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url='http://test') as client:
results = await asyncio.gather(client.get('/facescore/app/rank'), client.get('/facescore/wechat/rank'))
await asyncio.gather(*tasks)
self.assertEqual(['android', 'wechat'], [r.json()['source'] for r in results])
self.assertCountEqual(['android', 'wechat'], seen)
self.assertEqual('unknown', current_client_source.get())
async def test_admin_filters_use_source_instead_of_region_or_nickname(self):
fetch_all = AsyncMock(return_value=[])
fetch_one = AsyncMock(return_value={'total': 0})
with patch.object(database, 'fetch_all', fetch_all), patch.object(database, 'fetch_one', fetch_one):
await database.fetch_paged_image_records(category=None, nickname=None,
platform='miniprogram', offset=0, limit=20)
await database.count_image_records(category=None, nickname=None, platform='miniprogram')
await database.fetch_image_region_stats(category=None, nickname=None, platform='miniprogram')
for call in [*fetch_all.await_args_list, *fetch_one.await_args_list]:
self.assertIn('client_source = %s', call.args[0])
self.assertNotIn('region = %s', call.args[0])
self.assertEqual('wechat', call.args[1][0])
async def test_request_exception_restores_source_context(self):
app = FastAPI()
router = APIRouter(prefix='/facescore', route_class=FaceAnalysisFormRoute)
@router.get('/app/failure')
async def failure():
self.assertEqual('android', current_client_source.get())
raise RuntimeError('expected failure')
app.include_router(router)
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url='http://test') as client:
with self.assertRaisesRegex(RuntimeError, 'expected failure'):
await client.get('/facescore/app/failure')
self.assertEqual('unknown', current_client_source.get())
def test_routes_override_client_headers_and_queries(self):
for path, headers, expected in [
('/facescore/app/analyze', {}, 'android'),
('/facescore/analyze', {'x-device-id': 'spoof'}, 'wechat'),
('/facescore/rank', {'x-device-id': 'legacy'}, 'android'),
('/facescore/rank', {}, 'wechat'),
('/facescore/wechat/leaderboard', {'x-device-id': 'spoof'}, 'wechat'),
('/facescore/web/analyze', {}, 'web'), ('/facescore/ios/analyze', {}, 'ios')]:
request = Request({'type': 'http', 'path': path, 'query_string': b'client_source=android',
'headers': [(k.encode(), v.encode()) for k, v in headers.items()]})
self.assertEqual(expected, resolve_client_source(request))
if __name__ == '__main__':
unittest.main()
|