hair/tests/test_api.py

"""接口集成测试（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) 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 test_health_no_token_200(client):
    r = client.get("/health")
    assert r.status_code == 200
    assert r.json()["status"] == "ok"


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")}
    r = client.post(GROW, headers=H, files=files, data={"gender": "female"})
    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"])[:3] == b"\xff\xd8\xff"        # JPEG
    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_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_female(client):
    files = {"image_file": ("frontal.jpg", open(fixture("frontal.jpg"), "rb"), "application/octet-stream")}
    body = client.post(HLGEN, headers=H, files=files, data={"gender": "female"}).json()
    assert body["code"] == 0, body
    d = body["data"]
    assert [x["order"] for x in d["hairline_images"]] == [1, 2, 3, 4, 5]
    assert base64.b64decode(d["hairline_images"][0]["image_base64"])[:3] == b"\xff\xd8\xff"  # JPEG
    c = d["best_hairline_center_point"]
    assert 0 <= c["x"] <= 682 and 0 <= c["y"] <= 811        # 落在原图范围内
    assert "image_url" not in d["hairline_images"][0]


# 接口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())
    assert data["hairline_source"] in ("segmentation", "estimated")
    # 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"