import unittest from unittest.mock import AsyncMock, patch from test_ranking_consistency import database from fastapi import FastAPI from fastapi.testclient import TestClient import profile_routes class UserProfileApiTest(unittest.TestCase): def setUp(self): app = FastAPI() app.include_router(profile_routes.profile_router, prefix="/facescore") self.client = TestClient(app) self.headers = {"x-device-id": "0123456789abcdef"} def test_new_user_is_empty_and_does_not_create_record(self): with patch.object(profile_routes, "get_user_profile", AsyncMock(return_value={"display_name": "", "nickname_initialized": False})) as read: response = self.client.get("/facescore/app/profile", headers=self.headers) self.assertEqual(200, response.status_code) self.assertEqual({"success": True, "display_name": "", "nickname_initialized": False}, response.json()) read.assert_awaited_once_with("0123456789abcdef") def test_save_trim_clear_and_unicode_limit(self): for name, expected in [(" 小星星 ", "小星星"), ("", ""), (" ", ""), ("😀" * 12, "😀" * 12), ("昵称" * 6, "昵称" * 6)]: with self.subTest(name=name), patch.object(profile_routes, "set_user_display_name", AsyncMock()) as write: response = self.client.put("/facescore/app/profile", headers=self.headers, json={"display_name": name}) self.assertEqual(200, response.status_code) self.assertEqual(expected, response.json()["display_name"]) self.assertTrue(response.json()["nickname_initialized"]) write.assert_awaited_once_with("0123456789abcdef", expected, device_type=None, device_model=None, os_version=None, app_version=None, timezone=None, region=None, language=None) def test_first_profile_save_writes_complete_device_in_one_statement(self): # 贯穿 HTTP -> 实际数据库函数，仅隔离最终数据库连接。 headers = {**self.headers, "x-device-type": "Android", "x-device-model": "Google Pixel 9", "x-os-version": "Android 15", "x-app-version": "1.8.11", "x-timezone": "Asia/Shanghai", "x-region": "CN", "Accept-Language": "zh-CN", "x-client-source": "wechat"} with patch.object(database, "execute", AsyncMock()) as write: response = self.client.put("/facescore/app/profile", headers=headers, json={"display_name": "GlowBoss4826"}) self.assertEqual(200, response.status_code) write.assert_awaited_once() sql, params = write.await_args.args self.assertEqual(("0123456789abcdef", "GlowBoss4826", "Android", "Google Pixel 9", "Android 15", "1.8.11", "Asia/Shanghai", "CN", "zh-CN"), params) self.assertIn("'android'", sql) # 来源仍由接口确定，不能被请求头伪造。 self.assertNotIn("tpl_app_processed_images", sql) def test_region_fallback_and_blank_headers(self): for extra, expected in [({"x-region": " US ", "cf-ipcountry": "JP"}, "US"), ({"x-region": " ", "cf-ipcountry": "JP"}, "JP"), ({"x-country-code": "DE", "x-geo-country": "FR"}, "DE"), ({"x-geo-country": "FR"}, "FR"), ({}, None)]: with self.subTest(extra=extra), patch.object(profile_routes, "set_user_display_name", AsyncMock()) as write: response = self.client.put("/facescore/app/profile", headers={**self.headers, **extra, "x-device-model": " ", "x-app-version": " 1.8.11 "}, json={"display_name": ""}) self.assertEqual(200, response.status_code) self.assertEqual(expected, write.await_args.kwargs["region"]) self.assertIsNone(write.await_args.kwargs["device_model"]) self.assertEqual("1.8.11", write.await_args.kwargs["app_version"]) self.assertEqual("", write.await_args.args[1]) def test_device_headers_fit_database_columns(self): fields = {"x-device-type": ("device_type", 32), "x-device-model": ("device_model", 64), "x-os-version": ("os_version", 32), "x-app-version": ("app_version", 32), "x-timezone": ("timezone", 64), "x-region": ("region", 16), "Accept-Language": ("language", 32)} with patch.object(profile_routes, "set_user_display_name", AsyncMock()) as write: response = self.client.put("/facescore/app/profile", headers={**self.headers, **{key: "x" * 200 for key in fields}}, json={"display_name": "GlowBoss4826"}) self.assertEqual(200, response.status_code) for field, limit in fields.values(): self.assertEqual("x" * limit, write.await_args.kwargs[field]) def test_invalid_input_never_writes(self): for body in [{}, {"display_name": None}, {"display_name": 12}, {"display_name": "😀" * 13}, {"display_name": "x\ny"}, {"display_name": "\u202eevil"}, {"display_name": "\u0000"}]: with self.subTest(body=body), patch.object(profile_routes, "set_user_display_name", AsyncMock()) as write: response = self.client.put("/facescore/app/profile", headers=self.headers, json=body) self.assertEqual(422, response.status_code) write.assert_not_awaited() def test_initialize_returns_actual_name_including_manual_clear(self): for actual in ["GlowBoss4826", "AlreadySet", ""]: state = {"display_name": actual, "nickname_initialized": True} with self.subTest(actual=actual), \ patch.object(profile_routes, "get_user_profile", AsyncMock(return_value=state)), \ patch.object(database, "execute", AsyncMock()) as write: response = self.client.post("/facescore/app/profile/initialize", headers={**self.headers, "x-app-version": "1.8.11", "x-region": "CN"}, json={"display_name": "GlowBoss4826"}) self.assertEqual(200, response.status_code) self.assertEqual({"success": True, **state}, response.json()) write.assert_awaited_once() sql, params = write.await_args.args self.assertIn("IF(display_name_initialized = 0 AND display_name = '', VALUES(display_name), display_name)", sql) self.assertEqual("1.8.11", params[5]) self.assertEqual("CN", params[7]) def test_initialize_rejects_empty_developer_and_invalid_names(self): for candidate in ["", "99", "9527", "小帅9527", "GlowBoss0000", "toolongtitle1234", "Name1234\n"]: with self.subTest(candidate=candidate), patch.object(profile_routes, "set_user_display_name", AsyncMock()) as write: response = self.client.post("/facescore/app/profile/initialize", headers=self.headers, json={"display_name": candidate}) self.assertEqual(422, response.status_code) write.assert_not_awaited() def test_initialize_requires_valid_device(self): for headers in [{}, {"x-device-id": "unknown"}, {"x-device-id": " "}]: with patch.object(profile_routes, "set_user_display_name", AsyncMock()) as write: response = self.client.post("/facescore/app/profile/initialize", headers=headers, json={"display_name": "GlowBoss4826"}) self.assertIn(response.status_code, (400, 422)) write.assert_not_awaited() def test_initialize_failed_read_after_write_is_safe_to_retry(self): state = {"display_name": "OtherName", "nickname_initialized": True} with patch.object(profile_routes, "get_user_profile", AsyncMock(side_effect=[RuntimeError("offline"), state])), \ patch.object(database, "execute", AsyncMock()) as write: first = self.client.post("/facescore/app/profile/initialize", headers=self.headers, json={"display_name": "GlowBoss4826"}) second = self.client.post("/facescore/app/profile/initialize", headers=self.headers, json={"display_name": "GlowBoss4826"}) self.assertEqual(503, first.status_code) self.assertEqual("OtherName", second.json()["display_name"]) self.assertEqual(2, write.await_count) for call in write.await_args_list: self.assertIn("IF(display_name_initialized = 0 AND display_name = ''", call.args[0]) def test_initialize_write_failure_never_reports_success(self): with patch.object(database, "execute", AsyncMock(side_effect=RuntimeError("offline"))), \ patch.object(profile_routes, "get_user_profile", AsyncMock()) as read: response = self.client.post("/facescore/app/profile/initialize", headers=self.headers, json={"display_name": "GlowBoss4826"}) self.assertEqual(503, response.status_code) read.assert_not_awaited() def test_missing_or_invalid_device_is_rejected(self): for headers in [{}, {"x-device-id": "unknown"}, {"x-device-id": " "}, {"x-device-id": "x" * 129}]: with patch.object(profile_routes, "set_user_display_name", AsyncMock()) as write: response = self.client.put("/facescore/app/profile", headers=headers, json={"display_name": "hello"}) self.assertIn(response.status_code, (400, 422)) write.assert_not_awaited() def test_database_failure_does_not_report_save_success(self): with patch.object(profile_routes, "set_user_display_name", AsyncMock(side_effect=RuntimeError("offline"))): response = self.client.put("/facescore/app/profile", headers=self.headers, json={"display_name": "hello"}) self.assertEqual(503, response.status_code) with patch.object(profile_routes, "get_user_profile", AsyncMock(side_effect=RuntimeError("offline"))): response = self.client.get("/facescore/app/profile", headers=self.headers) self.assertEqual(503, response.status_code) def test_device_and_nickname_write_failure_returns_error(self): with patch.object(database, "execute", AsyncMock(side_effect=RuntimeError("offline"))) as write: response = self.client.put("/facescore/app/profile", headers={**self.headers, "x-device-model": "Google Pixel 9"}, json={"display_name": "GlowBoss4826"}) self.assertEqual(503, response.status_code) write.assert_awaited_once() class DisplayNameDatabaseTest(unittest.IsolatedAsyncioTestCase): def setUp(self): database._LEADERBOARD_CACHE.clear() async def test_save_only_changes_device_profile(self): with patch.object(database, "execute", AsyncMock()) as write: await database.set_user_display_name("device", "小星星") sql, params = write.await_args.args self.assertEqual(("device", "小星星") + (None,) * 7, params) self.assertNotIn("tpl_app_processed_images", sql) self.assertNotIn("nickname", sql) self.assertIn("ON DUPLICATE KEY UPDATE display_name", sql) async def test_missing_device_fields_do_not_clear_existing_profile(self): with patch.object(database, "execute", AsyncMock()) as write: await database.set_user_display_name("device", "", app_version="1.8.11", region="") sql, params = write.await_args.args self.assertEqual(("device", "", None, None, None, "1.8.11", None, "", None), params) for field in ("device_type", "device_model", "os_version", "app_version", "timezone", "region", "language"): self.assertIn(f"{field} = COALESCE(NULLIF(VALUES({field}), ''), {field})", sql) self.assertIn("display_name = VALUES(display_name)", sql) # 清空昵称仍有效。 self.assertIn("IF(client_source = 'unknown', VALUES(client_source), client_source)", sql) async def test_device_telemetry_preserves_display_name(self): with patch.object(database, "execute", AsyncMock()) as write: await database.upsert_device_record(device_id="device", app_version="1.8.6") self.assertNotIn("display_name", write.await_args.args[0]) async def test_missing_profile_reads_empty(self): with patch.object(database, "fetch_one", AsyncMock(return_value=None)): self.assertEqual("", await database.get_user_display_name("device")) async def test_profile_distinguishes_unset_named_and_cleared(self): for row, expected in [(None, False), ({"display_name": "", "display_name_initialized": 0}, False), ({"display_name": "", "display_name_initialized": 1}, True), ({"display_name": "OldName", "display_name_initialized": 0}, True)]: with self.subTest(row=row), patch.object(database, "fetch_one", AsyncMock(return_value=row)): profile = await database.get_user_profile("device") self.assertEqual(expected, profile["nickname_initialized"]) async def test_manual_clear_marks_initialized_and_auto_guard_precedes_marker(self): with patch.object(database, "execute", AsyncMock()) as write: await database.set_user_display_name("device", "") await database.set_user_display_name("device", "GlowBoss4826", initialize_only=True) manual, automatic = [call.args[0] for call in write.await_args_list] self.assertIn("display_name = VALUES(display_name)", manual) self.assertIn("display_name_initialized = 1", manual) self.assertLess(automatic.index("display_name = IF("), automatic.index("display_name_initialized = 1")) async def test_cached_rank_gets_live_names_for_all_records_and_other_viewers(self): rows = [{"id": 1, "score": 90, "nickname": "android_release_device", "updated_at": "2026-09-24 10:00:00"}, {"id": 2, "score": 89, "nickname": "android_device", "updated_at": "2026-09-23 10:00:00"}, {"id": 3, "score": 88, "nickname": "android_release_other"}] with patch.object(database, "fetch_all", AsyncMock(side_effect=[rows, [{"device_id": "device", "display_name": "old"}], [{"device_id": "device", "display_name": "new"}], [{"device_id": "device", "display_name": ""}]])) as fetch: first = await database.get_face_score_leaderboard(None, 50, "device") second = await database.get_face_score_leaderboard(None, 50, "other") third = await database.get_face_score_leaderboard(None, 50, "device") self.assertEqual(["old", "old", ""], [i["display_name"] for i in first["items"]]) self.assertEqual(["new", "new", ""], [i["display_name"] for i in second["items"]]) self.assertEqual(["", "", ""], [i["display_name"] for i in third["items"]]) self.assertFalse(second["items"][0]["is_current_user"]) self.assertTrue(second["items"][2]["is_current_user"]) self.assertEqual(4, fetch.await_count) # 排名仅查一次；昵称每次实时读取。 for result in (second, third): self.assertEqual([1, 2, 3], [i["rank"] for i in result["items"]]) self.assertEqual(first["items"][0]["updated_at"], result["items"][0]["updated_at"]) if __name__ == "__main__": unittest.main()