| 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): |
| |
| 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() |
|
|
