"""接口集成测试(FastAPI TestClient):错误码 + 鉴权 + 正常用例结构。""" import base64 import json import pytest from fastapi.testclient import TestClient from conftest import fixture import app as app_module # 测试自带固定密码,避免依赖 worker_config.json 的实际值(中间件运行期读模块全局) app_module.ACCEPT_PASSWORDS = ["testpass"] URL = "/api/v1/face/measure" H = {"X-Internal-Token": "testpass"} @pytest.fixture(scope="module") def client(): # with 触发 lifespan:加载模型单例(detector + 尽力加载 segmenter) with TestClient(app_module.app, client=("127.0.0.1", 50000)) as c: yield c def _post(client, fixture_name=None, headers=H, data=None, extra_files=None): files = {} if fixture_name: files["image_file"] = (fixture_name, open(fixture(fixture_name), "rb"), "application/octet-stream") if extra_files: files.update(extra_files) return client.post(URL, headers=headers, files=files or None, data=data) def test_auth_missing_token_401(client): r = _post(client, "frontal.jpg", headers={}) assert r.status_code == 401 def _patch_health_ok(monkeypatch): """让 /health 深检确定性通过:不依赖本机磁盘水位与 8188/8801 服务存活。""" monkeypatch.setattr(app_module, "_HEALTH_CFG", { **app_module._HEALTH_CFG, "min_free_gb": 0.0, "check_deps": True}) async def _deps_ok(): return True, {"comfyui": True, "change_hair": True} monkeypatch.setattr(app_module, "_check_deps", _deps_ok) def test_health_no_token_200(client, monkeypatch): _patch_health_ok(monkeypatch) r = client.get("/health") assert r.status_code == 200 assert r.json()["status"] == "ok" def test_health_disk_low_503(client, monkeypatch): # 阈值调到不可能的 99999G → 必然触发磁盘水位 503 monkeypatch.setattr(app_module, "_HEALTH_CFG", { **app_module._HEALTH_CFG, "min_free_gb": 99999.0, "check_deps": False}) r = client.get("/health") assert r.status_code == 503 body = r.json() assert body["status"] == "degraded" assert body["reason"].startswith("disk_free_below") def test_health_dep_down_503(client, monkeypatch): monkeypatch.setattr(app_module, "_HEALTH_CFG", { **app_module._HEALTH_CFG, "min_free_gb": 0.0, "check_deps": True}) async def _deps_down(): return False, {"comfyui": False, "change_hair": True} monkeypatch.setattr(app_module, "_check_deps", _deps_down) r = client.get("/health") assert r.status_code == 503 body = r.json() assert body["status"] == "degraded" assert "comfyui" in body["reason"] def test_health_ok_carries_detail(client, monkeypatch): _patch_health_ok(monkeypatch) r = client.get("/health") assert r.status_code == 200 body = r.json() assert "disk_free_gb" in body assert body["deps"] == {"comfyui": True, "change_hair": True} def test_tcp_probe_real_socket(): # 用进程内真实 socket 验证探测本身(曾因 open_connection 返回元组误判全部不可达) import asyncio async def scenario(): server = await asyncio.start_server(lambda r, w: None, host="127.0.0.1", port=0) port = server.sockets[0].getsockname()[1] up = await app_module._tcp_probe("127.0.0.1", port) server.close() await server.wait_closed() # 端口 1 无监听,本机 connect 立即被拒 down = await app_module._tcp_probe("127.0.0.1", 1) return up, down up, down = asyncio.run(scenario()) assert up is True assert down is False def test_param_none_provided_1007(client): r = client.post(URL, headers=H) assert r.json()["code"] == 1007 def test_param_multiple_provided_1007(client): r = _post(client, "frontal.jpg", data={"image_url": "http://example.com/x.jpg"}) assert r.json()["code"] == 1007 def test_no_face_1001(client): r = _post(client, "landscape.jpg") assert r.json()["code"] == 1001 def test_corrupt_1008(client): r = _post(client, "corrupt.bin") assert r.json()["code"] == 1008 GROW = "/api/v1/hair/grow" def test_grow_missing_gender_1004(client): files = {"image_file": ("frontal.jpg", open(fixture("frontal.jpg"), "rb"), "application/octet-stream")} r = client.post(GROW, headers=H, files=files) assert r.json()["code"] == 1004 # mock ComfyUI 输出:一张合法 PNG(worker 会把它重编码成 JPG) import cv2 as _cv2 import numpy as _np _PNG_1x1 = _cv2.imencode(".png", _np.full((8, 8, 3), 200, _np.uint8))[1].tobytes() def test_grow_female_returns_5(client, monkeypatch): # mock ComfyUI:不依赖 8182、不跑 Flux,只验证管线接线 + grown 字段 import hairline.comfyui as comfy monkeypatch.setattr(comfy, "run", lambda *a, **k: _PNG_1x1) files = {"image_file": ("frontal.jpg", open(fixture("frontal.jpg"), "rb"), "application/octet-stream")} # hair_style 必填(cb1989c 起):指定全部 5 种发型以验证完整管线 r = client.post(GROW, headers=H, files=files, data={"gender": "female", "hair_style": "1,2,3,4,5"}) body = r.json() assert body["code"] == 0, body results = body["data"]["results"] assert [x["hairline_type"] for x in results] == ["ellipse", "flower", "heart", "straight", "wave"] assert [x["order"] for x in results] == [1, 2, 3, 4, 5] assert base64.b64decode(results[0]["image_base64"])[:8] == b"\x89PNG\r\n\x1a\n" # 透明 PNG(发际线曲线) assert base64.b64decode(results[0]["grown_image_base64"])[:3] == b"\xff\xd8\xff" # JPEG(生发图) assert "image_url" not in results[0] GROWB = "/api/v1/hair/grow-b" def test_growb_missing_marked_1007(client): r = client.post(GROWB, headers=H) # 一张图都没传 assert r.json()["code"] == 1007 def test_growb_no_line_1001(client): files = {"marked_image_file": ("m.jpg", open(fixture("frontal.jpg"), "rb"), "application/octet-stream")} r = client.post(GROWB, headers=H, files=files) # 无划线 → 拒识 assert r.json()["code"] == 1001 def test_growb_success(client, monkeypatch): import hairline.comfyui as comfy monkeypatch.setattr(comfy, "run", lambda *a, **k: _PNG_1x1) files = {"marked_image_file": ("m.jpg", open(fixture("marked_hairline.jpg"), "rb"), "application/octet-stream")} body = client.post(GROWB, headers=H, files=files).json() # 只传划线图一张 assert body["code"] == 0, body d = body["data"] assert d["hairline_type"] == "custom" assert base64.b64decode(d["hair_growth_image_base64"])[:3] == b"\xff\xd8\xff" # JPEG assert "best_hairline_image_base64" not in d # 已去掉该字段 assert "best_hairline_image_url" not in d HLGEN = "/api/v1/hairline/generate" def test_hairline5_restored_schema_has_only_historical_parameters(client): schema = client.get("/openapi.json", headers=H).json() ref = schema["paths"][HLGEN]["post"]["requestBody"]["content"]["multipart/form-data"]["schema"]["$ref"] fields = schema["components"]["schemas"][ref.rsplit("/", 1)[-1]]["properties"] assert set(fields) == {"image_file", "image_url", "image_base64", "gender", "hair_style", "use_mask", "prompt", "generate_grow_image"} assert fields["use_mask"]["default"] is True assert fields["generate_grow_image"]["default"] is True assert fields["prompt"]["default"] == "填充遮罩区域的头发,人物增加一点磨皮" @pytest.mark.parametrize("gender", ["female", "male"]) def test_hairline5_restored_three_levels_and_original_coordinate_contract(client, gender): from pathlib import Path photo = Path(fixture("frontal.jpg")).read_bytes() body = client.post(HLGEN, headers=H, files={"image_file": ("frontal.jpg", photo, "image/jpeg")}, data={"gender": gender, "hair_style": "1,3", "generate_grow_image": "false"}).json() assert body["code"] == 0, body data = body["data"] assert set(data) == {"hairline_images", "best_hairline_center_point", "high_hairline_center_point", "low_hairline_center_point", "face_measure"} expected = _cv2.imdecode(_np.frombuffer(photo, _np.uint8), _cv2.IMREAD_COLOR).shape[:2] for key in ("best_hairline_center_point", "high_hairline_center_point", "low_hairline_center_point"): assert set(data[key]) == {"x", "y"} assert isinstance(data[key]["x"], int) and isinstance(data[key]["y"], int) for item in data["hairline_images"]: assert set(item) == {"hairline_type", "image_middle_base64", "image_high_base64", "image_low_base64", "grown_image_base64", "order"} assert item["grown_image_base64"] is None for level in ("middle", "high", "low"): image = _cv2.imdecode(_np.frombuffer(base64.b64decode(item[f"image_{level}_base64"]), _np.uint8), _cv2.IMREAD_UNCHANGED) assert image.shape == (*expected, 4) def test_hairline_gen_missing_gender_1004(client): files = {"image_file": ("frontal.jpg", open(fixture("frontal.jpg"), "rb"), "application/octet-stream")} assert client.post(HLGEN, headers=H, files=files).json()["code"] == 1004 def test_hairline_gen_missing_hairstyle_1007(client): # hair_style 已改为必填(同接口2):只给 gender、不给 hair_style → 1007 files = {"image_file": ("frontal.jpg", open(fixture("frontal.jpg"), "rb"), "application/octet-stream")} assert client.post(HLGEN, headers=H, files=files, data={"gender": "female"}).json()["code"] == 1007 def test_hairline_gen_female(client, monkeypatch): # mock ComfyUI:不依赖 8182,只验证三档叠图 + 生发字段接线 import hairline.comfyui as comfy monkeypatch.setattr(comfy, "run", lambda *a, **k: _PNG_1x1) files = {"image_file": ("frontal.jpg", open(fixture("frontal.jpg"), "rb"), "application/octet-stream")} body = client.post(HLGEN, headers=H, files=files, data={"gender": "female", "hair_style": "1,3"}).json() assert body["code"] == 0, body d = body["data"] imgs = d["hairline_images"] # 只返回选中发型,order = 发型序号 assert [x["order"] for x in imgs] == [1, 3] assert [x["hairline_type"] for x in imgs] == ["ellipse", "heart"] # 三档叠图为透明 PNG(发际线曲线,含 alpha),生发图为 JPEG(完整人像) for k in ("image_middle_base64", "image_high_base64", "image_low_base64"): assert base64.b64decode(imgs[0][k])[:8] == b"\x89PNG\r\n\x1a\n", k assert base64.b64decode(imgs[0]["grown_image_base64"])[:3] == b"\xff\xd8\xff" assert "image_middle_url" not in imgs[0] c = d["best_hairline_center_point"] assert 0 <= c["x"] <= 682 and 0 <= c["y"] <= 811 # 落在原图范围内 # face_measure:复用接口1测量数值(容错,失败为 null;本用例正常 → 必须有完整结构) fm = d["face_measure"] assert fm is not None, "face_measure 不应为 null(正常正面照)" assert set(["face_total_height_cm", "four_courts", "seven_eyes", "landmarks", "hairline_source", "head_pose"]).issubset(fm.keys()) # 不应含标注图字段(接口5 只要数值,不要画线图) assert "annotated_image_base64" not in fm assert "annotated_image_url" not in fm # 子结构 assert set(["top_court_cm", "upper_court_cm", "middle_court_cm", "lower_court_cm", "ratios"]).issubset(fm["four_courts"].keys()) assert set(["eye_width_cm", "face_width_cm", "inter_eye_distance_cm", "ratios"]).issubset(fm["seven_eyes"].keys()) # 七眼 eye1~eye7 键必须存在(eye1/eye7 在耳朵不可见时可为 null) assert set([f"eye{i}" for i in range(1, 8)]).issubset(fm["seven_eyes"].keys()) assert set(["hair_top", "hairline", "brow_center", "nose_bottom", "chin_tip"]).issubset(fm["landmarks"].keys()) assert fm["hairline_source"] in ("segmentation", "estimated", "corrected") assert set(["yaw", "pitch", "roll"]).issubset(fm["head_pose"].keys()) HLGEN52 = "/api/v1/hairline/generate-52" def test_hairline52_legacy_form_parameters_and_defaults_are_preserved(client): schema = client.get("/openapi.json", headers=H).json() body = schema["paths"][HLGEN52]["post"]["requestBody"]["content"]["multipart/form-data"]["schema"] body = schema["components"]["schemas"][body["$ref"].rsplit("/", 1)[-1]] fields = body["properties"] legacy = {"image_file", "image_url", "image_base64", "gender", "hair_style", "use_mask", "prompt", "generate_grow_image"} assert legacy.issubset(fields) assert not set(body.get("required", [])) - legacy for key in ("image_file", "image_url", "image_base64", "gender", "hair_style"): assert any(choice.get("type") == "string" for choice in fields[key].get("anyOf", [fields[key]])) assert fields["use_mask"]["type"] == "boolean" and fields["use_mask"]["default"] is True assert fields["generate_grow_image"]["type"] == "boolean" and fields["generate_grow_image"]["default"] is True assert fields["prompt"]["default"] == "填充遮罩区域的头发,人物增加一点磨皮" assert fields["normalize_image"]["default"] is False @pytest.mark.parametrize("image_kind", ["file", "base64", "url"]) def test_hairline52_legacy_request_keeps_fields_and_original_coordinates(client, monkeypatch, image_kind): import contextlib from pathlib import Path from types import SimpleNamespace import httpx import hairline.comfyui as comfy raw = Path(fixture("frontal.jpg")).read_bytes() calls = [] def generate(*args, **kwargs): calls.append(kwargs) return _PNG_1x1 monkeypatch.setattr(comfy, "run", generate) form = {"gender": "male", "hair_style": "1", "use_mask": "false", "prompt": "legacy prompt"} kwargs = {"headers": H, "data": form} if image_kind == "file": kwargs["files"] = {"image_file": ("frontal.jpg", raw, "image/jpeg")} elif image_kind == "base64": form["image_base64"] = "data:image/jpeg;base64," + base64.b64encode(raw).decode() else: url = "https://legacy.example.test/frontal.jpg" form["image_url"] = url response = httpx.Response(200, content=raw, request=httpx.Request("GET", url)) def get_image(actual_url): assert actual_url == url return response monkeypatch.setattr(httpx, "Client", lambda **kw: contextlib.nullcontext(SimpleNamespace(get=get_image))) # No new parameters: old clients keep their original-size overlays. body = client.post(HLGEN52, **kwargs).json() assert body["code"] == 0, body assert {"code", "message", "request_id", "data"}.issubset(body) data = body["data"] assert {"hairline_images", "best_hairline_center_point", "high_hairline_center_point", "low_hairline_center_point", "face_measure"}.issubset(data) assert data["coordinate_space"] == "original_image" and data["framing"] is None assert data["input_image_base64"] == data["original_image_base64"] original = _cv2.imdecode(_np.frombuffer(raw, _np.uint8), _cv2.IMREAD_COLOR) expected = original.shape[:2] assert (data["height"], data["width"]) == expected item = data["hairline_images"][0] assert {"hairline_type", "image_middle_base64", "image_high_base64", "image_low_base64", "grown_image_base64", "order"}.issubset(item) assert item["image_high_base64"] is None and item["image_low_base64"] is None assert data["high_hairline_center_point"] is None and data["low_hairline_center_point"] is None for key in ("image_middle_base64", "grown_image_base64"): image = _cv2.imdecode(_np.frombuffer(base64.b64decode(item[key]), _np.uint8), _cv2.IMREAD_UNCHANGED) assert image.shape[:2] == expected assert len(calls) == 1 and calls[0]["prompt"] == "legacy prompt" @pytest.mark.parametrize("normalize_image", [True, False]) def test_hairline52_height_compare_uses_same_new_rule(client, monkeypatch, normalize_image): import hairline.comfyui as comfy def unexpected_generation(*args, **kwargs): pytest.fail("Position comparison must not invoke ComfyUI by default") monkeypatch.setattr(comfy, "run", unexpected_generation) with open(fixture("frontal.jpg"), "rb") as photo: response = client.post("/api/v1/debug/hairline-height-compare", files={"image_file": ("frontal.jpg", photo, "image/jpeg")}, data={"gender": "male", "hair_style": "1,3", "normalize_image": str(normalize_image).lower()}) data = response.json() assert data["code"] == 0, data data = data["data"] assert [r["order"] for r in data["results"]] == [1, 3] assert set(data["guide_landmarks"]) == {"9", "150", "152", "162", "251", "389"} assert data["guide_landmarks"]["9"]["y"] < data["guide_landmarks"]["152"]["y"] expected_size = (data["height"], data["width"]) base = _cv2.imdecode(_np.frombuffer(base64.b64decode(data["input_base64"]), _np.uint8), _cv2.IMREAD_COLOR) assert base.shape[:2] == expected_size assert data["coordinate_space"] == ("processed_image" if normalize_image else "original_image") if normalize_image: assert data["framing"]["enabled"] is True assert data["framing"]["axis_x_error_px"] < 1e-8 assert data["guide_landmarks"]["9"]["x"] == pytest.approx(data["guide_landmarks"]["152"]["x"], abs=1e-8) else: assert data["framing"] is None original = _cv2.imread(fixture("frontal.jpg")) assert base.shape == original.shape for result in data["results"]: assert result["height_placement"]["target_error_px"] <= 1 assert result["grown_image_base64"] is None for key in ("old_overlay_base64", "new_overlay_base64"): assert base64.b64decode(result[key])[:8] == b"\x89PNG\r\n\x1a\n" overlay = _cv2.imdecode(_np.frombuffer(base64.b64decode(result[key]), _np.uint8), _cv2.IMREAD_UNCHANGED) assert overlay.shape[:2] == expected_size def test_hairline52_gen_missing_gender_1004(client): files = {"image_file": ("frontal.jpg", open(fixture("frontal.jpg"), "rb"), "application/octet-stream")} assert client.post(HLGEN52, headers=H, files=files).json()["code"] == 1004 def test_hairline52_gen_missing_hairstyle_1007(client): # hair_style 已改为必填(同接口2):只给 gender、不给 hair_style → 1007 files = {"image_file": ("frontal.jpg", open(fixture("frontal.jpg"), "rb"), "application/octet-stream")} assert client.post(HLGEN52, headers=H, files=files, data={"gender": "female"}).json()["code"] == 1007 def test_hairline52_gen_female(client, monkeypatch): # mock ComfyUI:验证处理后底图、发际线和生发图使用相同尺寸与坐标。 import hairline.comfyui as comfy monkeypatch.setattr(comfy, "run", lambda *a, **k: _PNG_1x1) files = {"image_file": ("frontal.jpg", open(fixture("frontal.jpg"), "rb"), "application/octet-stream")} body = client.post(HLGEN52, headers=H, files=files, data={"gender": "female", "hair_style": "1,3", "normalize_image": "true"}).json() assert body["code"] == 0, body d = body["data"] assert d["coordinate_space"] == "processed_image" assert d["framing"]["enabled"] is True assert d["framing"]["axis_x_error_px"] < 1e-8 expected_size = (d["height"], d["width"]) for value in (d["input_image_base64"], d["hairline_images"][0]["image_middle_base64"], d["hairline_images"][0]["grown_image_base64"]): image = _cv2.imdecode(_np.frombuffer(base64.b64decode(value), _np.uint8), _cv2.IMREAD_UNCHANGED) assert image.shape[:2] == expected_size imgs = d["hairline_images"] # 只返回选中发型,order = 发型序号 assert [x["order"] for x in imgs] == [1, 3] assert [x["hairline_type"] for x in imgs] == ["ellipse", "heart"] # Only the designed middle curve is generated; deprecated fields stay null. assert base64.b64decode(imgs[0]["image_middle_base64"])[:8] == b"\x89PNG\r\n\x1a\n" assert all(x["image_high_base64"] is None and x["image_low_base64"] is None for x in imgs) assert d["high_hairline_center_point"] is None and d["low_hairline_center_point"] is None rule = d["height_rule"] assert rule["target_upper_height_px"] == pytest.approx( (rule["middle_height_px"] + rule["lower_height_px"]) / 2) for x in imgs: assert x["height_placement"]["new_center"]["y"] == pytest.approx(rule["target_center"]["y"], abs=1) assert base64.b64decode(imgs[0]["grown_image_base64"])[:3] == b"\xff\xd8\xff" assert "image_middle_url" not in imgs[0] c = d["best_hairline_center_point"] assert 0 <= c["x"] < d["width"] and 0 <= c["y"] < d["height"] # face_measure:复用接口1测量数值(容错,失败为 null;本用例正常 → 必须有完整结构) fm = d["face_measure"] assert fm is not None, "face_measure 不应为 null(正常正面照)" assert set(["face_total_height_cm", "four_courts", "seven_eyes", "landmarks", "hairline_source", "head_pose"]).issubset(fm.keys()) # 不应含标注图字段(接口52 只要数值,不要画线图) assert "annotated_image_base64" not in fm assert "annotated_image_url" not in fm # 子结构 assert set(["top_court_cm", "upper_court_cm", "middle_court_cm", "lower_court_cm", "ratios"]).issubset(fm["four_courts"].keys()) assert set(["eye_width_cm", "face_width_cm", "inter_eye_distance_cm", "ratios"]).issubset(fm["seven_eyes"].keys()) # 七眼 eye1~eye7 键必须存在(eye1/eye7 在耳朵不可见时可为 null) assert set([f"eye{i}" for i in range(1, 8)]).issubset(fm["seven_eyes"].keys()) assert set(["hair_top", "hairline", "brow_center", "nose_bottom", "chin_tip"]).issubset(fm["landmarks"].keys()) assert fm["hairline_source"] in ("segmentation", "estimated", "corrected") assert set(["yaw", "pitch", "roll"]).issubset(fm["head_pose"].keys()) # 接口4(用户特征)已迁到网关本机实现(直接调豆包),不再在 worker; # 其测试随实现一起在网关侧做,worker 这边不再覆盖。 def test_success_structure(client): r = _post(client, "frontal.jpg") body = r.json() assert body["code"] == 0, body data = body["data"] # 业务字段对齐文档 assert set(["face_total_height_cm", "four_courts", "seven_eyes", "landmarks", "hairline_source", "head_pose", "annotated_image_base64"]).issubset(data.keys()) assert set(["top_court_cm", "upper_court_cm", "middle_court_cm", "lower_court_cm", "ratios"]).issubset(data["four_courts"].keys()) assert set(["eye_width_cm", "face_width_cm", "inter_eye_distance_cm", "ratios"]).issubset(data["seven_eyes"].keys()) # 七眼:从左到右 eye1~eye7(eye1/eye7 在耳朵不可见时可为 null,但键必须存在) assert set([f"eye{i}" for i in range(1, 8)]).issubset(data["seven_eyes"].keys()) # eye3/eye5 为左右眼宽、eye4 为两眼间距,与 eye_width_cm/inter_eye_distance_cm 语义一致 assert data["seven_eyes"]["eye3"] is not None assert data["seven_eyes"]["eye4"] is not None assert data["seven_eyes"]["eye5"] is not None assert data["hairline_source"] in ("segmentation", "estimated", "corrected") # base64 解码为合法 PNG(非 URL) assert "annotated_image_url" not in data png = base64.b64decode(data["annotated_image_base64"]) assert png[:8] == b"\x89PNG\r\n\x1a\n"