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