| """运行实际评分编排/接口函数，隔离模型、文件存储和数据库，不启动服务。""" |
| 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() |
|
|
