"""运行实际评分编排/接口函数，隔离模型、文件存储和数据库，不启动服务。""" import ast import base64 import json import os from pathlib import Path import time import types import unittest import uuid from unittest.mock import AsyncMock, Mock import numpy as np from fastapi import File, Form, HTTPException, Query from fastapi.responses import JSONResponse from models import ModelType ROOT = Path(__file__).resolve().parents[1] def load_function(filename: str, name: str, namespace: dict, class_name: str = "") -> object: """仅编译指定的真实函数，跳过模块顶层的模型加载和资源同步。""" tree = ast.parse((ROOT / filename).read_text()) scope = next(node for node in tree.body if isinstance(node, ast.ClassDef) and node.name == class_name) if class_name else tree node = [node for node in scope.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == name][-1] node.decorator_list = [] module = ast.Module(body=[ast.ImportFrom(module="__future__", names=[ast.alias(name="annotations")], level=0), node], type_ignores=[]) exec(compile(ast.fix_missing_locations(module), filename, "exec"), namespace) return namespace[name] class OptionalFacialAnalysisTest(unittest.IsolatedAsyncioTestCase): def setUp(self) -> None: self.image = np.zeros((224, 224, 3), dtype=np.uint8) self.logger = Mock() namespace = { "np": np, "os": os, "time": time, "uuid": uuid, "json": json, "base64": base64, "logger": self.logger, "ModelType": ModelType, "File": File, "Form": Form, "Query": Query, "HTTPException": HTTPException, "JSONResponse": JSONResponse, "config": types.SimpleNamespace(DRAW_SCORE=False), "OUTPUT_DIR": "/isolated", "IMAGES_DIR": "/isolated", "SAVE_QUALITY": 85, "FACE_SCORE_MAX_IMAGES": 5, "CLIP_AVAILABLE": False, "FEMALE_AGE_ADJUSTMENT_THRESHOLD": 999, "FEMALE_AGE_ADJUSTMENT": 0, "get_date_subfolder": Mock(return_value="day"), "save_image_high_quality": Mock(return_value=True), "_get_save_path": Mock(return_value=("/isolated/result.webp", "result.webp")), "_record_device_info": AsyncMock(), "_record_output_file": AsyncMock(), "_get_region_from_headers": Mock(return_value="CN"), "_log_stage_duration": Mock(), "cv2": types.SimpleNamespace( resize=Mock(return_value=self.image), imdecode=Mock(return_value=self.image), getTextSize=Mock(return_value=((100, 10), 2)), FONT_HERSHEY_SIMPLEX=0, IMREAD_COLOR=1, ), } unavailable = load_function("facial_analyzer.py", "_unavailable_facial_analysis", namespace, "FacialFeatureAnalyzer") namespace["FacialFeatureAnalyzer"] = types.SimpleNamespace(_unavailable_facial_analysis=unavailable) self.unavailable = unavailable self.auxiliary = Mock() self.analyzer = types.SimpleNamespace( _detect_faces=Mock(return_value=[[0, 0, 224, 224]]), _crop_face=Mock(return_value=self.image), facial_analyzer=self.auxiliary, _predict_with_hybrid_model=Mock(return_value={ "beauty_score": 72.87, "beauty_score_100": 72.87, "gender": "Male", "age": "25", "model_used": "hybrid", }), gender_colors={"Male": (255, 165, 0)}, ) self.analyzer.analyze_faces = types.MethodType( load_function("face_analyzer.py", "analyze_faces", namespace, "EnhancedFaceAnalyzer"), self.analyzer) namespace["analyzer"] = self.analyzer load_function("utils.py", "convert_numpy_types", namespace) self.endpoints = [load_function("api_routes.py", "analyze_face", namespace), load_function("api_routes.py", "app_analyze_face", namespace)] async def assert_main_response(self, expected_auxiliary: dict) -> None: for endpoint in self.endpoints: with self.subTest(endpoint=endpoint.__name__): response = await endpoint( request=types.SimpleNamespace(headers={}), file=types.SimpleNamespace(read=AsyncMock(return_value=b"isolated-image")), files=None, images=None, nickname=None, model=ModelType.HYBRID, ) self.assertEqual(200, response.status_code) body = json.loads(response.body) self.assertTrue(body["success"], body) self.assertEqual(1, body["face_count"]) face = body["faces"][0]["face"] self.assertEqual(72.87, face["beauty_score_100"]) self.assertEqual("Male", face["gender"]) self.assertEqual("25", face["age"]) self.assertEqual(expected_auxiliary, face["facial_features"]) self.assertEqual("result.webp", body["faces"][0]["annotated_image_filename"]) async def test_missing_landmarks_still_return_http_200_and_main_score(self) -> None: unavailable = self.unavailable("no_landmarks_detected") self.auxiliary.analyze_facial_features.return_value = unavailable await self.assert_main_response(unavailable) self.assertNotIn("beauty_timeline", unavailable) self.assertNotIn("facial_features", unavailable) async def test_unexpected_auxiliary_exception_never_discards_main_result(self) -> None: self.auxiliary.analyze_facial_features.side_effect = RuntimeError("auxiliary failure") await self.assert_main_response(self.unavailable("auxiliary_analysis_failed")) async def test_successful_auxiliary_data_is_preserved(self) -> None: valid = {"analysis_method": "mediapipe_landmarks", "scores_available": True, "facial_features": {"eyes": 7.2, "nose": 8.1}} self.auxiliary.analyze_facial_features.return_value = valid await self.assert_main_response(valid) async def test_one_auxiliary_failure_does_not_drop_other_faces(self) -> None: self.analyzer._detect_faces.return_value *= 2 valid = {"scores_available": True, "facial_features": {"eyes": 8.5}} self.auxiliary.analyze_facial_features.side_effect = [RuntimeError("first face failure"), valid] result = self.analyzer.analyze_faces(self.image, "isolated", ModelType.HYBRID) self.assertTrue(result["success"]) self.assertEqual(2, result["face_count"]) self.assertEqual(2, len(result["faces"])) self.assertEqual([72.87, 72.87], [face["beauty_score_100"] for face in result["faces"]]) self.assertFalse(result["faces"][0]["facial_features"]["scores_available"]) self.assertEqual(valid, result["faces"][1]["facial_features"]) if __name__ == "__main__": unittest.main()