112 lines
3.1 KiB
Python
112 lines
3.1 KiB
Python
"""微博商品图片后端下载与内嵌测试。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
|
|
import pytest
|
|
|
|
from web.backend.services import weibo_image_service
|
|
|
|
PNG_BYTES = b"\x89PNG\r\n\x1a\n" + b"test-image-data"
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
body: bytes = PNG_BYTES,
|
|
status_code: int = 200,
|
|
content_length: str | None = None,
|
|
):
|
|
self.status_code = status_code
|
|
self.headers = (
|
|
{"Content-Length": content_length} if content_length is not None else {}
|
|
)
|
|
self.body = body
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return None
|
|
|
|
def iter_content(self, chunk_size: int):
|
|
del chunk_size
|
|
yield self.body
|
|
|
|
|
|
def test_fetch_image_url_returns_embedded_png(monkeypatch):
|
|
captured: dict = {}
|
|
|
|
def fake_get(url, **kwargs):
|
|
captured["url"] = url
|
|
captured["headers"] = kwargs["headers"]
|
|
return FakeResponse()
|
|
|
|
monkeypatch.setattr(weibo_image_service.requests, "get", fake_get)
|
|
|
|
result = weibo_image_service.fetch_weibo_image_data_url(
|
|
"http://wx3.sinaimg.cn/large/demo.png"
|
|
)
|
|
|
|
assert captured["url"] == "https://wx3.sinaimg.cn/large/demo.png"
|
|
assert captured["headers"]["Referer"] == "https://vipclub.e.weibo.com/"
|
|
assert result == (
|
|
"data:image/png;base64," + base64.b64encode(PNG_BYTES).decode("ascii")
|
|
)
|
|
|
|
|
|
def test_fetch_image_url_rejects_non_image(monkeypatch):
|
|
monkeypatch.setattr(
|
|
weibo_image_service.requests,
|
|
"get",
|
|
lambda *_args, **_kwargs: FakeResponse(body=b"<html>error</html>"),
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="不是支持的图片类型"):
|
|
weibo_image_service.fetch_weibo_image_data_url(
|
|
"https://wx3.sinaimg.cn/large/error.jpg"
|
|
)
|
|
|
|
|
|
def test_fetch_image_url_rejects_oversized_content(monkeypatch):
|
|
monkeypatch.setattr(
|
|
weibo_image_service.requests,
|
|
"get",
|
|
lambda *_args, **_kwargs: FakeResponse(
|
|
content_length=str(weibo_image_service.MAX_IMAGE_BYTES + 1)
|
|
),
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="超过 5 MiB"):
|
|
weibo_image_service.fetch_weibo_image_data_url(
|
|
"https://wx3.sinaimg.cn/large/large.png"
|
|
)
|
|
|
|
|
|
def test_embed_goods_images_falls_back_to_normalized_https(monkeypatch):
|
|
goods = [
|
|
{"id": 1, "goods_image": "http://wx3.sinaimg.cn/large/ok.png"},
|
|
{"id": 2, "goods_image": "//wx4.sinaimg.cn/large/fail.png"},
|
|
]
|
|
|
|
def fake_fetch(url: str) -> str:
|
|
if url.endswith("ok.png"):
|
|
return "data:image/png;base64,b2s="
|
|
raise ValueError("download failed")
|
|
|
|
monkeypatch.setattr(
|
|
weibo_image_service,
|
|
"fetch_weibo_image_data_url",
|
|
fake_fetch,
|
|
)
|
|
|
|
result = weibo_image_service.embed_weibo_goods_images(goods, account_id=1)
|
|
|
|
assert result.total == 2
|
|
assert result.loaded == 1
|
|
assert result.failed == 1
|
|
assert goods[0]["goods_image"] == "data:image/png;base64,b2s="
|
|
assert goods[1]["goods_image"] == "https://wx4.sinaimg.cn/large/fail.png"
|