xsl
1c51f3eaab
撤回52号网关路由、图片字段改名和mock改动，gateway目录恢复a3fddbb。worker保持base64和null响应，更新接口说明和回归测试。 验证：worker API、IP限制与网关兼容性共54项测试通过。
63 lines
2.7 KiB
Python
63 lines
2.7 KiB
Python
"""Keep the gateway's historical image-field conversion unchanged."""
|
|
import asyncio
|
|
import json
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
|
|
import httpx
|
|
import pytest
|
|
from starlette.requests import Request
|
|
|
|
from gateway.forward import rewrite_base64_to_url
|
|
|
|
|
|
def test_nullable_image_fields_keep_original_names(tmp_path):
|
|
value = {"hairline_images": [{"image_high_base64": None, "image_low_base64": None,
|
|
"grown_image_base64": None}],
|
|
"high_hairline_center_point": None, "face_measure": None}
|
|
actual = rewrite_base64_to_url(value, "https://example.test", str(tmp_path))
|
|
assert actual == value
|
|
assert not list(tmp_path.iterdir())
|
|
|
|
|
|
def test_nonempty_images_still_rewrite_alongside_null_fields(tmp_path):
|
|
import base64
|
|
import io
|
|
from PIL import Image
|
|
buf = io.BytesIO()
|
|
Image.new("RGBA", (2, 2), (255, 255, 255, 0)).save(buf, format="PNG")
|
|
value = {"image_middle_base64": base64.b64encode(buf.getvalue()).decode(),
|
|
"image_high_base64": None}
|
|
actual = rewrite_base64_to_url(value, "https://example.test", str(tmp_path))
|
|
assert actual["image_middle_url"].startswith("https://example.test/static/annotations/")
|
|
assert actual["image_high_base64"] is None
|
|
assert "image_high_url" not in actual
|
|
assert len(list(tmp_path.iterdir())) == 1
|
|
|
|
|
|
@pytest.mark.parametrize("path,expected_key", [
|
|
("/api/v1/hairline/generate", "grown_image_base64"),
|
|
("/api/v1/hair/grow", "grown_image_base64"),
|
|
])
|
|
def test_proxy_keeps_original_null_image_fields(tmp_path, monkeypatch, path, expected_key):
|
|
import gateway.config as config
|
|
import gateway.forward as forward
|
|
|
|
monkeypatch.setattr(config, "get_config", lambda: {
|
|
"dispatch": {"request_timeout_seconds": 10}, "shared_password": "test-token",
|
|
"public_base_url": "https://example.test", "static_dir": str(tmp_path),
|
|
})
|
|
monkeypatch.setattr(forward, "acquire_worker", AsyncMock(return_value=SimpleNamespace(url="http://worker")))
|
|
monkeypatch.setattr(forward, "release_worker", AsyncMock())
|
|
worker_response = httpx.Response(200, json={"code": 0, "data": {"hairline_images": [
|
|
{"grown_image_base64": None}]}})
|
|
monkeypatch.setattr(forward, "get_client", lambda: SimpleNamespace(request=AsyncMock(return_value=worker_response)))
|
|
|
|
async def receive():
|
|
return {"type": "http.request", "body": b"", "more_body": False}
|
|
|
|
request = Request({"type": "http", "method": "POST", "path": path, "headers": []}, receive)
|
|
response = asyncio.run(forward.proxy_request(request, path))
|
|
assert response.status_code == 200
|
|
assert json.loads(response.body)["data"]["hairline_images"] == [{expected_key: None}]
|