duobibi-app-mock/duobibi(mock)backend/duobibi-server/tests/test_parse_image.py

"""新接口的端到端测试。会真实跑到 mock_extractor,因此每个 case 会 sleep 1.5-2.5s。
我们 monkey-patch 把 sleep 拿掉,加快测试。"""
from __future__ import annotations

import io
import json
from unittest.mock import patch

import pytest
from fastapi.testclient import TestClient

from app.main import app

client = TestClient(app)


@pytest.fixture(autouse=True)
def _no_sleep():
    """禁用 mock_extractor 的真实异步延迟,测试运行时间从 ~10s 降到 < 1s。"""
    async def _instant(_seconds):
        return None
    with patch("app.mock_extractor.asyncio.sleep", _instant):
        yield


def _fake_image(payload: bytes = b"fake jpeg bytes 0123456789") -> tuple:
    return ("image", ("test.jpg", io.BytesIO(payload), "image/jpeg"))


def test_parse_image_basic() -> None:
    r = client.post(
        "/api/v1/parse-image",
        files=[_fake_image()],
        data={"clusters": "[]"},
    )
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["title"]
    assert body["price"] > 0
    assert body["typical_price"] > 0
    assert body["source_app"] in {"淘宝", "京东", "拼多多", "抖音", "美团", "饿了么"}
    assert body["cluster_id"] is None  # 空 clusters 必然新建


def test_parse_image_same_bytes_same_product() -> None:
    """同样字节 → 同样商品(便于真机重放)。允许价格因抖动小幅不同。"""
    r1 = client.post("/api/v1/parse-image", files=[_fake_image(b"AAAA")], data={"clusters": "[]"})
    r2 = client.post("/api/v1/parse-image", files=[_fake_image(b"AAAA")], data={"clusters": "[]"})
    assert r1.status_code == 200
    assert r2.status_code == 200
    assert r1.json()["title"] == r2.json()["title"]
    assert r1.json()["source_app"] == r2.json()["source_app"]


def test_parse_image_cluster_match() -> None:
    """如果 clusters 里有 title 与 mock 选中商品共享前 4 字符,应命中该 cluster。"""
    # 先发一次拿到 picked title
    r1 = client.post("/api/v1/parse-image", files=[_fake_image(b"BBBB")], data={"clusters": "[]"})
    picked_title = r1.json()["title"]
    cluster_id = "uuid-fake-1"
    clusters = [{"id": cluster_id, "title": picked_title}]
    r2 = client.post(
        "/api/v1/parse-image",
        files=[_fake_image(b"BBBB")],
        data={"clusters": json.dumps(clusters)},
    )
    assert r2.status_code == 200
    assert r2.json()["cluster_id"] == cluster_id


def test_parse_image_no_image() -> None:
    r = client.post(
        "/api/v1/parse-image",
        files=[("image", ("empty.jpg", io.BytesIO(b""), "image/jpeg"))],
        data={"clusters": "[]"},
    )
    assert r.status_code == 422
    assert r.json()["detail"] == "no_image"


def test_parse_image_missing_image_field() -> None:
    r = client.post("/api/v1/parse-image", data={"clusters": "[]"})
    assert r.status_code == 422  # FastAPI 字段校验失败也返回 422


def test_parse_image_invalid_clusters_json() -> None:
    r = client.post(
        "/api/v1/parse-image",
        files=[_fake_image()],
        data={"clusters": "not a json"},
    )
    assert r.status_code == 422
    assert r.json()["detail"] == "invalid_clusters"


def test_parse_image_clusters_not_array() -> None:
    r = client.post(
        "/api/v1/parse-image",
        files=[_fake_image()],
        data={"clusters": '{"id":"x","title":"y"}'},
    )
    assert r.status_code == 422
    assert r.json()["detail"] == "invalid_clusters"


def test_parse_image_too_large() -> None:
    big = b"x" * (5 * 1024 * 1024 + 1)
    r = client.post(
        "/api/v1/parse-image",
        files=[("image", ("big.jpg", io.BytesIO(big), "image/jpeg"))],
        data={"clusters": "[]"},
    )
    assert r.status_code == 422
    assert r.json()["detail"] == "image_too_large"


def test_parse_text_basic() -> None:
    r = client.post(
        "/api/v1/parse-text",
        json={
            "title": "iPhone 15 Pro Max 256GB",
            "price": 9999.0,
            "source_app": "淘宝",
            "clusters": [],
        },
    )
    assert r.status_code == 200
    body = r.json()
    assert body["title"] == "iPhone 15 Pro Max 256GB"
    assert body["price"] == 9999.0
    assert body["source_app"] == "淘宝"
    assert body["typical_price"] > 0
    assert body["cluster_id"] is None


def test_parse_text_cluster_match() -> None:
    """同样 title 前 4 字符的 cluster 应命中。"""
    cluster_id = "uuid-text-1"
    r = client.post(
        "/api/v1/parse-text",
        json={
            "title": "iPhone 15 Pro Max 1TB 暮光紫",
            "price": 11999.0,
            "source_app": "京东",
            "clusters": [{"id": cluster_id, "title": "iPhone 15 Pro 256GB"}],
        },
    )
    assert r.status_code == 200
    assert r.json()["cluster_id"] == cluster_id


def test_parse_text_empty_title() -> None:
    r = client.post(
        "/api/v1/parse-text",
        json={"title": "  ", "price": 9.9, "source_app": "淘宝", "clusters": []},
    )
    assert r.status_code == 422
    assert r.json()["detail"] == "empty_title"


def test_parse_text_invalid_price() -> None:
    r = client.post(
        "/api/v1/parse-text",
        json={"title": "x", "price": 0, "source_app": "淘宝", "clusters": []},
    )
    assert r.status_code == 422
    assert r.json()["detail"] == "invalid_price"


def test_parse_text_negative_price() -> None:
    r = client.post(
        "/api/v1/parse-text",
        json={"title": "x", "price": -1, "source_app": "淘宝", "clusters": []},
    )
    assert r.status_code == 422


def test_parse_image_invalid_cluster_schema() -> None:
    """clusters 数组里某个对象缺 id/title 字段 → Pydantic ValidationError 必须返回 422 而非 500"""
    r = client.post(
        "/api/v1/parse-image",
        files=[_fake_image()],
        data={"clusters": '[{"id":"only-id-no-title"}]'},
    )
    assert r.status_code == 422
    assert r.json()["detail"] == "invalid_clusters"


def test_concurrent_parse_image_does_not_block_event_loop(_no_sleep) -> None:
    """禁用 sleep 后并发应该极快;若 event loop 被阻塞会显著退化。
    主要保护 mock_extract_image 永远是 async + await。"""
    import time as _t
    start = _t.time()
    for _ in range(10):
        r = client.post("/api/v1/parse-image", files=[_fake_image()], data={"clusters": "[]"})
        assert r.status_code == 200
    # 10 次串行(TestClient 是同步客户端)且 sleep 被 mock 掉,应远 < 1s
    assert _t.time() - start < 2.0